third slice: artifacts and deployments...

docs/superpowers/plans/2026-06-02-wf-api-slice-4c-artifacts-deployments.md

@@ -0,0 +1,678 @@
+# wf_api Slice 4C: Artifacts And Deployments 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:** Move saved artifact and deployment operations out of `WorkflowSurfaceHandlers` into protocol-neutral `wf_api` domain services.
+
+**Architecture:** Add `WorkflowArtifactApi` and `WorkflowDeploymentApi` that depend on `WorkflowOperationContext`, not `WfMcpService`. Keep `WorkflowSurfaceHandlers` public method signatures unchanged and delegate artifact/deployment methods to the new services. Extend operation-context protocols for event emission and live source checks so `wf_api` does not import MCP event factories, adapters, auth, or connections.
+
+**Tech Stack:** Python 3.14+, `wf_api.operation_context`, `wf_api.drafts`, `wf_artifacts`, `wf_platform`, `wf_core`, pytest, ruff, basedpyright.
+
+---
+
+## Scope
+
+### Move In This Slice
+
+Move these methods from `WorkflowSurfaceHandlers`:
+
+```text
+list_artifacts
+save_artifact
+create_artifact_from_plan
+create_artifact_from_draft
+create_artifact_from_workspace
+create_wrapper_from_workspace
+inspect_artifact
+list_deployments
+inspect_deployment
+save_deployment
+delete_deployment
+validate_deployment
+```
+
+Move or duplicate only the helpers required by those methods:
+
+```text
+_available_sources
+_suggested_self_bindings
+_observed_node_specs
+_capability_name
+_artifact_capability_id
+_deployment_summary
+```
+
+### Do Not Move In This Slice
+
+Do not move:
+
+```text
+list_capabilities
+inspect_capability
+call_capability
+_wrapper_artifact_for_capability_name
+_wrapper_capability_summaries
+_wrapper_capability_detail
+_call_wrapper_artifact
+run_deployment
+resume_run
+inspect_run
+read_run_trace
+_raw_plan_from_artifact
+_run_payload
+_interrupt_payload
+```
+
+Reasons:
+
+- Capability methods still combine live source specs, saved wrappers, and direct test calls. Move them in Slice 4E.
+- Run methods depend on durable run checkpoints, runtime preparation, trace slicing, and saved subgraph execution. Move them in Slice 4D.
+- `_raw_plan_from_artifact` is still needed by wrapper capability calls and run methods. Leave it in `handlers.py` until those domains move or extract it separately.
+
+### Invariants
+
+- No public payload changes.
+- No MCP tool schema changes.
+- `WorkflowSurfaceHandlers` still exposes the same methods.
+- `wf_api` imports no `wf_mcp`.
+- Event construction stays adapter-owned.
+- Live upstream checks stay adapter-owned.
+- Temporary private helper duplication is allowed when capability/run methods still need a helper in `handlers.py`.
+
+---
+
+## Task 1: Extend Operation Context For Events And Live Checks
+
+**Files:**
+- Modify: `src/wf_api/operation_context.py`
+- Create: `src/wf_mcp/broker/service/workflow_live_checks.py`
+- Modify: `src/wf_mcp/broker/service/workflow_operation_context.py`
+- Modify: `src/wf_mcp/workflow_surface/handlers.py`
+- Test: `tests/wf_api/test_operation_context.py`
+
+- [ ] **Step 1: Add protocol methods**
+
+In `src/wf_api/operation_context.py`, update `WorkflowEventRecorder`:
+
+```python
+class WorkflowEventRecorder(Protocol):
+    """Records workflow lifecycle events without exposing MCP event types."""
+
+    def record_event(self, event: object) -> None:
+        """Record one adapter-native event object."""
+        ...
+
+    def record_workflow_event(
+        self,
+        event_type: str,
+        *,
+        capability_id: str,
+        payload: dict[str, Any],
+    ) -> None:
+        """Record one workflow lifecycle event by protocol-neutral fields."""
+        ...
+```
+
+Update `WorkflowLiveSourceChecker`:
+
+```python
+class WorkflowLiveSourceChecker(Protocol):
+    """Optional hook for validating live external source availability."""
+
+    async def deployment_diagnostics(
+        self,
+        *,
+        deployment: WorkflowDeployment,
+        artifacts: Sequence[WorkflowArtifact],
+    ) -> list[DependencyDiagnostic]:
+        """Return opt-in live-source diagnostics for a deployment tree."""
+        ...
+```
+
+Add imports:
+
+```python
+from collections.abc import Mapping, Sequence
+from wf_artifacts import DependencyDiagnostic, WorkflowDeployment
+```
+
+Remove or keep `available_sources()` only if still used by tests. New moved code should use `deployment_diagnostics(...)`.
+
+- [ ] **Step 2: Move MCP live-check helper out of handlers**
+
+Create `src/wf_mcp/broker/service/workflow_live_checks.py` and move these
+handler-level live-check pieces into it:
+
+```text
+LIVE_SOURCE_CHECK_TIMEOUT_SECONDS
+_LIVE_SOURCE_CHECK_FAILURES
+live_source_diagnostics
+_required_live_sources
+```
+
+Rename `_live_source_diagnostics(...)` to public module-private-adapter helper
+`live_source_diagnostics(...)`.
+
+The new module should own the MCP-only imports:
+
+```python
+import asyncio
+
+import anyio
+import httpx
+from mcp.client.streamable_http import StreamableHTTPError
+from mcp.shared.exceptions import McpError
+
+from wf_artifacts import DependencyDiagnostic, DiagnosticSeverity, WorkflowArtifact, WorkflowDeployment
+from wf_mcp.broker.service.adapters import require_adapter
+from wf_mcp.broker.service.core import WfMcpService
+```
+
+Keep the existing docstring explaining that live checks perform opt-in upstream
+I/O. This split is required to avoid a circular import:
+
+```text
+handlers.py -> workflow_operation_context.py -> handlers.py
+```
+
+After the move, update `handlers.py` to import `live_source_diagnostics` from
+the new module for as long as `validate_deployment` still lives in handlers.
+When Task 4 delegates `validate_deployment`, remove that handler import if it
+is unused.
+
+- [ ] **Step 3: Implement MCP adapter methods**
+
+In `src/wf_mcp/broker/service/workflow_operation_context.py`, import:
+
+```python
+from collections.abc import Sequence
+
+from wf_artifacts import DependencyDiagnostic, WorkflowArtifact, WorkflowDeployment
+from wf_mcp.events import make_event
+from wf_mcp.broker.service.workflow_live_checks import live_source_diagnostics
+```
+
+Then update event recorder:
+
+```python
+def record_workflow_event(
+    self,
+    event_type: str,
+    *,
+    capability_id: str,
+    payload: dict[str, Any],
+) -> None:
+    self.service._record_event(  # noqa: SLF001
+        make_event(event_type, capability_id=capability_id, payload=payload)
+    )
+```
+
+Update live source checker:
+
+```python
+async def deployment_diagnostics(
+    self,
+    *,
+    deployment: WorkflowDeployment,
+    artifacts: Sequence[WorkflowArtifact],
+) -> list[DependencyDiagnostic]:
+    return await live_source_diagnostics(
+        self.service,
+        deployment=deployment,
+        artifacts=artifacts,
+    )
+```
+
+Do not import handler modules from `workflow_operation_context.py`.
+
+- [ ] **Step 4: Update operation context tests**
+
+In `tests/wf_api/test_operation_context.py`, add a test that calls:
+
+```python
+operation_context.events.record_workflow_event(
+    "workflow_artifact_saved",
+    capability_id="workflow.demo.v1",
+    payload={"artifact_id": "demo", "version": 1},
+)
+```
+
+Then assert the service recorded an event with stable fields individually. Do not assert full dict equality.
+
+- [ ] **Step 5: Run focused tests**
+
+```powershell
+uv run pytest tests/wf_api/test_operation_context.py -q
+```
+
+Expected: pass.
+
+---
+
+## Task 2: Create `wf_api.artifacts`
+
+**Files:**
+- Create: `src/wf_api/artifacts.py`
+- Modify: `src/wf_api/__init__.py`
+- Test: `tests/wf_api/test_artifact_api.py`
+
+- [ ] **Step 1: Create service skeleton**
+
+Create `src/wf_api/artifacts.py`:
+
+```python
+from __future__ import annotations
+
+from collections.abc import Sequence
+from typing import Any
+
+from wf_artifacts import (
+    ArtifactKind,
+    RequiredCapability,
+    WorkflowArtifact,
+    create_workflow_artifact_from_plan as build_workflow_artifact_from_plan,
+)
+from wf_platform import CapabilityRef, NodeSpecInventory
+
+from .drafts import WorkflowDraftApi
+from .models import RawWorkflowPlan
+from .operation_context import WorkflowOperationContext
+
+
+class WorkflowArtifactApi:
+    """Saved workflow artifact operations.
+
+    Event construction is intentionally delegated through
+    WorkflowOperationContext so this module stays protocol-neutral.
+    """
+
+    def __init__(self, context: WorkflowOperationContext) -> None:
+        self.context = context
+        self.drafts = WorkflowDraftApi(context)
+
+    def _artifact_store(self):
+        if self.context.artifact_store is None:
+            raise KeyError("workflow artifact store is not configured")
+        return self.context.artifact_store
+```
+
+- [ ] **Step 2: Move artifact methods**
+
+Move these bodies from `WorkflowSurfaceHandlers`, replacing `self.service` access:
+
+```text
+list_artifacts
+save_artifact
+create_artifact_from_plan
+create_artifact_from_draft
+inspect_artifact
+```
+
+Required replacements:
+
+```python
+self.service.artifact_store -> self._artifact_store()
+self.service.workflow_artifact_catalog_entry(artifact) -> self.context.artifacts.workflow_artifact_catalog_entry(artifact)
+self.service._record_event(make_event(...)) -> self.context.events.record_workflow_event(...)
+_observed_node_specs(self.service) -> _observed_node_specs(self.context)
+```
+
+Keep return payloads byte-for-byte equivalent except for dictionary ordering.
+
+- [ ] **Step 3: Move workspace artifact methods**
+
+Move:
+
+```text
+create_artifact_from_workspace
+create_wrapper_from_workspace
+```
+
+Use `self.context.draft_workspace_store` through `self.drafts` or a local store helper. Preserve current behavior:
+
+- validate workspace draft first
+- return `saved: False` with diagnostics when invalid
+- call `create_artifact_from_draft(...)` when valid
+- wrapper path passes `kind="wrapper"`
+
+- [ ] **Step 4: Add helpers**
+
+Add private helpers to `wf_api.artifacts`:
+
+```text
+_required_capability_payloads
+_suggested_self_bindings
+_observed_node_specs
+_plan_nodes
+_artifact_capability_id
+```
+
+Duplicate `_required_capability_payloads`, `_observed_node_specs`, and `_plan_nodes` from `wf_api.drafts` for now instead of importing private draft helpers. We can consolidate after Slice 4C if duplication becomes annoying.
+
+Do not remove `_artifact_capability_id` from `handlers.py`; wrapper capability methods still need it until Slice 4E.
+
+- [ ] **Step 5: Export artifact service**
+
+In `src/wf_api/__init__.py`:
+
+```python
+from .artifacts import WorkflowArtifactApi
+```
+
+Add `"WorkflowArtifactApi"` to `__all__`.
+
+- [ ] **Step 6: Add focused tests**
+
+Create `tests/wf_api/test_artifact_api.py` with tests that instantiate `WorkflowArtifactApi(context_from_service(service))`:
+
+- `save_artifact` stores a `WorkflowArtifact` and returns `saved: True`.
+- `create_artifact_from_plan` saves an artifact and includes observed node specs.
+- `create_artifact_from_workspace` returns `saved: False` when workspace validation fails.
+- `create_wrapper_from_workspace` saves `kind == "wrapper"`.
+- Handler delegation for `inspect_artifact` returns the same stable fields as direct `WorkflowArtifactApi.inspect_artifact`.
+
+Use field-by-field assertions unless asserting a known closed model shape.
+
+- [ ] **Step 7: Run artifact tests**
+
+```powershell
+uv run pytest tests/wf_api/test_artifact_api.py tests/wf_api/test_drafts_service.py -q
+```
+
+Expected: pass.
+
+---
+
+## Task 3: Create `wf_api.deployments`
+
+**Files:**
+- Create: `src/wf_api/deployments.py`
+- Modify: `src/wf_api/__init__.py`
+- Test: `tests/wf_api/test_deployment_api.py`
+
+- [ ] **Step 1: Create service skeleton**
+
+Create `src/wf_api/deployments.py`:
+
+```python
+from __future__ import annotations
+
+from typing import Any
+
+from wf_artifacts import (
+    AvailableCapability,
+    AvailableSource,
+    DependencyDiagnostic,
+    WorkflowArtifact,
+    WorkflowDeployment,
+    hash_json_schema,
+    validate_deployment_dependencies,
+)
+from wf_platform import CapabilitySource
+
+from .next_actions import NextActions
+from .operation_context import WorkflowOperationContext
+from .saved_subgraphs import resolve_saved_subgraph_tree, validate_saved_subgraph_tree
+
+
+class WorkflowDeploymentApi:
+    """Saved deployment operations and dependency validation."""
+
+    def __init__(self, context: WorkflowOperationContext) -> None:
+        self.context = context
+
+    def _artifact_store(self):
+        if self.context.artifact_store is None:
+            raise KeyError("workflow artifact store is not configured")
+        return self.context.artifact_store
+```
+
+- [ ] **Step 2: Move deployment methods**
+
+Move these bodies from `WorkflowSurfaceHandlers`:
+
+```text
+list_deployments
+inspect_deployment
+save_deployment
+delete_deployment
+validate_deployment
+```
+
+Required replacements:
+
+```python
+self.service.artifact_store -> self._artifact_store()
+self.service._record_event(make_event(...)) -> self.context.events.record_workflow_event(...)
+_available_sources(self.service) -> _available_sources(self.context.capability_sources)
+```
+
+For `validate_deployment(live_check=True)`, use:
+
+```python
+if live_check and self.context.live_sources is not None:
+    diagnostics.extend(
+        await self.context.live_sources.deployment_diagnostics(
+            deployment=deployment,
+            artifacts=[artifact, *tree.artifacts_by_ref.values()],
+        )
+    )
+```
+
+If `live_check=True` and `live_sources is None`, preserve static validation only. Do not invent a new warning payload in this slice.
+
+- [ ] **Step 3: Move deployment validation helper**
+
+Move `_deployment_validation` logic into `WorkflowDeploymentApi` as a private method:
+
+```python
+def _deployment_validation(
+    self,
+    deployment_id: str,
+) -> tuple[WorkflowDeployment, WorkflowArtifact, list[DependencyDiagnostic], SavedSubgraphTree]:
+    ...
+```
+
+Use `self._artifact_store()` and `_available_sources(self.context.capability_sources)`.
+
+- [ ] **Step 4: Add helper functions**
+
+Add private helpers:
+
+```text
+_available_sources
+_capability_name
+_deployment_summary
+```
+
+Adapt `_available_sources` to accept `Mapping[str, CapabilitySource]` instead of `WfMcpService`.
+
+Do not remove `_available_sources` or `_capability_name` from `handlers.py` if run/capability methods still use them.
+
+- [ ] **Step 5: Export deployment service**
+
+In `src/wf_api/__init__.py`:
+
+```python
+from .deployments import WorkflowDeploymentApi
+```
+
+Add `"WorkflowDeploymentApi"` to `__all__`.
+
+- [ ] **Step 6: Add focused tests**
+
+Create `tests/wf_api/test_deployment_api.py` with tests that instantiate `WorkflowDeploymentApi(context_from_service(service))`:
+
+- `save_deployment` stores and returns stable deployment fields.
+- `list_deployments` returns compact summaries.
+- `delete_deployment` removes one deployment.
+- `validate_deployment(live_check=False)` returns `runnable` for a valid binding.
+- `validate_deployment(live_check=True)` calls the operation-context live checker. A simple fake context may be easier than MCP service setup for this test.
+- Handler delegation for `validate_deployment` returns the same stable status/diagnostic fields as direct `WorkflowDeploymentApi.validate_deployment`.
+
+- [ ] **Step 7: Run deployment tests**
+
+```powershell
+uv run pytest tests/wf_api/test_deployment_api.py tests/wf_mcp/workflow_surface/test_deployments.py -q
+```
+
+Expected: pass.
+
+---
+
+## Task 4: Wire `WorkflowSurfaceHandlers`
+
+**Files:**
+- Modify: `src/wf_mcp/workflow_surface/handlers.py`
+
+- [ ] **Step 1: Add imports**
+
+```python
+from wf_api.artifacts import WorkflowArtifactApi
+from wf_api.deployments import WorkflowDeploymentApi
+```
+
+- [ ] **Step 2: Instantiate services**
+
+In `WorkflowSurfaceHandlers.__init__`, avoid building multiple independent contexts:
+
+```python
+context = context_from_service(service)
+self._drafts = WorkflowDraftApi(context)
+self._artifacts = WorkflowArtifactApi(context)
+self._deployments = WorkflowDeploymentApi(context)
+```
+
+- [ ] **Step 3: Replace moved artifact methods with delegates**
+
+Replace bodies for:
+
+```text
+list_artifacts
+save_artifact
+create_artifact_from_plan
+create_artifact_from_draft
+create_artifact_from_workspace
+create_wrapper_from_workspace
+inspect_artifact
+```
+
+Example:
+
+```python
+async def inspect_artifact(self, *, artifact_id: str, version: int) -> dict[str, Any]:
+    return await self._artifacts.inspect_artifact(
+        artifact_id=artifact_id,
+        version=version,
+    )
+```
+
+- [ ] **Step 4: Replace moved deployment methods with delegates**
+
+Replace bodies for:
+
+```text
+list_deployments
+inspect_deployment
+save_deployment
+delete_deployment
+validate_deployment
+```
+
+Example:
+
+```python
+async def validate_deployment(
+    self,
+    *,
+    deployment_id: str,
+    live_check: bool = False,
+) -> dict[str, Any]:
+    return await self._deployments.validate_deployment(
+        deployment_id=deployment_id,
+        live_check=live_check,
+    )
+```
+
+- [ ] **Step 5: Remove only unused imports/helpers**
+
+After delegation, run:
+
+```powershell
+rg -n "_available_sources|_suggested_self_bindings|_observed_node_specs|_deployment_summary|_artifact_capability_id|_capability_name" src/wf_mcp/workflow_surface/handlers.py
+```
+
+Remove a helper from `handlers.py` only if it has no remaining caller there.
+
+Expected likely result:
+
+- `_suggested_self_bindings`, `_observed_node_specs`, `_deployment_summary` can probably be removed.
+- `_artifact_capability_id`, `_capability_name`, `_available_sources` may still be needed by capability/run methods. Keep them if referenced.
+
+---
+
+## Task 5: Verification
+
+- [ ] **Step 1: Run focused wf_api tests**
+
+```powershell
+uv run pytest tests/wf_api/test_artifact_api.py tests/wf_api/test_deployment_api.py tests/wf_api/test_drafts_service.py -q
+```
+
+Expected: pass.
+
+- [ ] **Step 2: Run workflow surface tests**
+
+```powershell
+uv run pytest tests/wf_mcp/workflow_surface -q
+```
+
+Expected: pass.
+
+- [ ] **Step 3: Run import-direction test**
+
+```powershell
+uv run pytest tests/wf_api/test_import_direction.py -q
+```
+
+Expected: pass; `wf_api` has no `wf_mcp` imports.
+
+- [ ] **Step 4: Run ruff on touched files**
+
+```powershell
+uv run ruff check src/wf_api/artifacts.py src/wf_api/deployments.py src/wf_api/operation_context.py src/wf_api/__init__.py src/wf_mcp/broker/service/workflow_operation_context.py src/wf_mcp/workflow_surface/handlers.py tests/wf_api/test_artifact_api.py tests/wf_api/test_deployment_api.py
+```
+
+Expected: all checks pass.
+
+- [ ] **Step 5: Run basedpyright on touched files**
+
+```powershell
+uv run basedpyright --level error src/wf_api/artifacts.py src/wf_api/deployments.py src/wf_api/operation_context.py src/wf_mcp/broker/service/workflow_operation_context.py src/wf_mcp/workflow_surface/handlers.py tests/wf_api/test_artifact_api.py tests/wf_api/test_deployment_api.py
+```
+
+Expected: `0 errors`.
+
+- [ ] **Step 6: Optional full suite**
+
+```powershell
+uv run pytest -q
+```
+
+Expected: full suite passes with the project’s existing skipped/xfailed counts.
+
+---
+
+## Self-Review Checklist
+
+- `wf_api.artifacts` imports no `wf_mcp`.
+- `wf_api.deployments` imports no `wf_mcp`.
+- Event construction remains in `wf_mcp.broker.service.workflow_operation_context`.
+- Live upstream adapter/auth probing remains in `wf_mcp`.
+- `WorkflowSurfaceHandlers` public artifact/deployment method signatures are unchanged.
+- `create_draft_workspace_from_capability` still lives in `WorkflowSurfaceHandlers`.
+- Capability methods still live in `WorkflowSurfaceHandlers`.
+- Run methods still live in `WorkflowSurfaceHandlers`.
+- No public payload shape changed.
+- No MCP schema changed.
+- Temporary helper duplication is documented and deliberate.

src/wf_api/__init__.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from .artifacts import WorkflowArtifactApi
 from .backend import TraceRange, WorkflowApiBackend
 from .constants import (
     DEFAULT_CALL_STEP_ID,
@@ -8,6 +9,7 @@ from .constants import (
     DEFAULT_OK_OUTCOME,
     RUNTIME_ERROR_CAPABILITY,
 )
+from .deployments import WorkflowDeploymentApi
 from .drafts import WorkflowDraftApi
 from .next_actions import NextActionPatchExample, NextActionTool, NextActions
 from .refs import WorkflowSurfaceCapabilityId, parse_workflow_surface_capability_id
@@ -52,7 +54,9 @@ __all__ = [
     "TraceRange",
     "WorkflowApi",
     "WorkflowApiBackend",
+    "WorkflowArtifactApi",
     "WorkflowArtifactCataloger",
+    "WorkflowDeploymentApi",
     "WorkflowDraftApi",
     "WorkflowEventRecorder",
     "WorkflowLiveSourceChecker",

src/wf_api/artifacts.py

@@ -0,0 +1,352 @@
+"""Saved workflow artifact operations.
+
+Event construction is intentionally delegated through
+WorkflowOperationContext so this module stays protocol-neutral.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from typing import Any, TypeVar
+
+from wf_artifacts import (
+    ArtifactKind,
+    RequiredCapability,
+    WorkflowArtifact,
+    WorkflowCapabilityRef,
+    create_workflow_artifact_from_plan as build_workflow_artifact_from_plan,
+)
+from wf_platform import NodeSpecInventory, page_items
+
+from .drafts import WorkflowDraftApi
+from .models import RawWorkflowPlan
+from .operation_context import WorkflowOperationContext
+
+T = TypeVar("T")
+
+
+def _matches_query(*values: object, query: str | None) -> bool:
+    """Return whether a compact discovery row matches a human search query."""
+    if query is None:
+        return True
+    needle = query.strip().casefold()
+    if not needle:
+        return True
+    return any(needle in str(value).casefold() for value in values if value is not None)
+
+
+def _paged_list_payload(
+    key: str,
+    items: Sequence[T],
+    *,
+    cursor: str | None,
+    limit: int,
+) -> dict[str, Any]:
+    """Build the common workflow-surface list response shape."""
+    page = page_items(items, cursor=cursor, limit=limit)
+    return {
+        key: list(page.items),
+        "next_cursor": page.next_cursor,
+        "total": page.total,
+    }
+
+
+class WorkflowArtifactApi:
+    """Saved workflow artifact operations.
+
+    Event construction is intentionally delegated through
+    WorkflowOperationContext so this module stays protocol-neutral.
+    """
+
+    def __init__(self, context: WorkflowOperationContext) -> None:
+        self.context = context
+        self.drafts = WorkflowDraftApi(context)
+
+    def _artifact_store(self):
+        if self.context.artifact_store is None:
+            raise KeyError("workflow artifact store is not configured")
+        return self.context.artifact_store
+
+    async def list_artifacts(
+        self,
+        *,
+        query: str | None = None,
+        kind: ArtifactKind | None = None,
+        cursor: str | None = None,
+        limit: int = 50,
+    ) -> dict[str, Any]:
+        """Return compact paged saved artifact summaries.
+
+        Saved artifacts can contain full raw workflow plans, so list results
+        deliberately stay summary-only. Use inspect/run tools for detail.
+        """
+        if self.context.artifact_store is None:
+            return _paged_list_payload("nodes", [], cursor=cursor, limit=limit)
+        entries = [
+            self.context.artifacts.workflow_artifact_catalog_entry(artifact).model_dump(
+                mode="json"
+            )
+            for artifact in self.context.artifact_store.list_artifacts()
+            if kind is None or artifact.kind == kind
+        ]
+        entries = [
+            entry
+            for entry in entries
+            if _matches_query(
+                entry.get("name"),
+                entry.get("artifact_id"),
+                entry.get("display_name"),
+                entry.get("description"),
+                entry.get("kind"),
+                query=query,
+            )
+        ]
+        entries.sort(key=lambda entry: str(entry["name"]))
+        return _paged_list_payload("nodes", entries, cursor=cursor, limit=limit)
+
+    async def save_artifact(self, artifact: dict[str, Any]) -> dict[str, Any]:
+        workflow_artifact = WorkflowArtifact.model_validate(artifact)
+        self._artifact_store().save_artifact(workflow_artifact)
+        self.context.events.record_workflow_event(
+            "workflow_artifact_saved",
+            capability_id=_artifact_capability_id(workflow_artifact),
+            payload={
+                "artifact_id": workflow_artifact.id,
+                "version": workflow_artifact.version,
+            },
+        )
+        return {
+            "artifact_id": workflow_artifact.id,
+            "version": workflow_artifact.version,
+            "saved": True,
+        }
+
+    async def create_artifact_from_plan(
+        self,
+        *,
+        artifact_id: str,
+        version: int,
+        title: str,
+        plan: RawWorkflowPlan | dict[str, Any],
+        outcomes: Sequence[str],
+        kind: ArtifactKind = "workflow",
+        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]:
+        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=typed_plan.model_dump(mode="json", by_alias=True),
+            outcomes=tuple(outcomes),
+            required_capabilities={
+                name: RequiredCapability.model_validate(capability)
+                for name, capability in (required_capabilities or {}).items()
+            },
+            source_bindings=source_bindings,
+            observed_node_specs=_observed_node_specs(self.context),
+            created_from_catalog_version=created_from_catalog_version,
+        )
+        self._artifact_store().save_artifact(workflow_artifact)
+        self.context.events.record_workflow_event(
+            "workflow_artifact_saved",
+            capability_id=_artifact_capability_id(workflow_artifact),
+            payload={
+                "artifact_id": workflow_artifact.id,
+                "version": workflow_artifact.version,
+                "created_from_plan": True,
+            },
+        )
+        return {
+            "artifact_id": workflow_artifact.id,
+            "version": workflow_artifact.version,
+            "saved": True,
+        }
+
+    async def create_artifact_from_draft(
+        self,
+        *,
+        artifact_id: str,
+        version: int,
+        title: str,
+        draft: dict[str, Any],
+        outcomes: Sequence[str],
+        kind: ArtifactKind = "workflow",
+        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]:
+        from wf_artifacts import compile_workflow_draft
+
+        plan = compile_workflow_draft(draft)
+        workflow_artifact = build_workflow_artifact_from_plan(
+            artifact_id=artifact_id,
+            version=version,
+            title=title,
+            kind=kind,
+            description=description,
+            plan=plan,
+            outcomes=tuple(outcomes),
+            required_capabilities={
+                name: RequiredCapability.model_validate(capability)
+                for name, capability in (required_capabilities or {}).items()
+            },
+            source_bindings=source_bindings,
+            observed_node_specs=_observed_node_specs(self.context),
+            created_from_catalog_version=created_from_catalog_version,
+        )
+        self._artifact_store().save_artifact(workflow_artifact)
+        self.context.events.record_workflow_event(
+            "workflow_artifact_saved",
+            capability_id=_artifact_capability_id(workflow_artifact),
+            payload={
+                "artifact_id": workflow_artifact.id,
+                "version": workflow_artifact.version,
+                "created_from_draft": True,
+            },
+        )
+        required_sources = sorted(
+            {
+                capability.logical_source
+                for capability in workflow_artifact.required_capability_map().values()
+            }
+        )
+        return {
+            "artifact_id": workflow_artifact.id,
+            "version": workflow_artifact.version,
+            "saved": True,
+            "required_logical_sources": required_sources,
+            "suggested_bindings": _suggested_self_bindings(required_sources),
+        }
+
+    async def create_artifact_from_workspace(
+        self,
+        *,
+        workspace_id: str,
+        artifact_id: str,
+        version: int,
+        title: str,
+        outcomes: Sequence[str],
+        kind: ArtifactKind = "workflow",
+        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]:
+        store = self.context.draft_workspace_store
+        if store is None:
+            raise KeyError("draft workspace store is not configured")
+        workspace = store.get_workspace(workspace_id)
+        validation = await self.drafts.validate_draft(draft=workspace.draft)
+        if validation["status"] != "valid":
+            return {
+                "saved": False,
+                "workspace_id": workspace_id,
+                "revision": workspace.revision,
+                "status": validation["status"],
+                "diagnostics": validation["diagnostics"],
+            }
+        return await self.create_artifact_from_draft(
+            artifact_id=artifact_id,
+            version=version,
+            title=title,
+            kind=kind,
+            description=description,
+            draft=workspace.draft,
+            outcomes=outcomes,
+            required_capabilities=required_capabilities,
+            source_bindings=source_bindings,
+            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,
+        )
+
+    async def inspect_artifact(
+        self, *, artifact_id: str, version: int
+    ) -> dict[str, Any]:
+        artifact = self._artifact_store().get_artifact(artifact_id, version)
+        return artifact.model_dump(mode="json")
+
+
+def _required_capability_payloads(
+    requirements: dict[str, RequiredCapability],
+) -> dict[str, dict[str, Any]]:
+    return {
+        name: capability.model_dump(mode="json")
+        for name, capability in sorted(requirements.items())
+    }
+
+
+def _suggested_self_bindings(required_sources: Sequence[str]) -> dict[str, str]:
+    """Suggest local bindings for built-in sources that deploy to themselves."""
+    return {
+        source: source for source in required_sources if source in {"wf.std", "wf.mcp"}
+    }
+
+
+def _observed_node_specs(
+    context: WorkflowOperationContext,
+) -> dict[str, NodeSpecInventory]:
+    """Project current executable specs into serializable observed contracts."""
+    observed: dict[str, NodeSpecInventory] = {}
+    for source in context.capability_sources.values():
+        inventory = source.as_inventory()
+        observed.update(
+            {detail.name: detail for detail in inventory.capabilities.node_spec_details}
+        )
+    return observed
+
+
+def _plan_nodes(artifact: WorkflowArtifact) -> list[dict[str, Any]]:
+    nodes = artifact.plan.get("nodes", [])
+    return [node for node in nodes if isinstance(node, dict)]
+
+
+def _artifact_capability_id(artifact: WorkflowArtifact) -> str:
+    """Use the same stable name shape as workflow artifact catalog entries."""
+    return str(
+        WorkflowCapabilityRef(
+            artifact_id=artifact.id,
+            version=artifact.version,
+        )
+    )
+
+
+__all__ = [
+    "WorkflowArtifactApi",
+]

src/wf_api/deployments.py

@@ -0,0 +1,211 @@
+"""Saved deployment operations and dependency validation."""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Any
+
+from wf_artifacts import (
+    AvailableCapability,
+    AvailableSource,
+    DependencyDiagnostic,
+    WorkflowArtifact,
+    WorkflowDeployment,
+    validate_deployment_dependencies,
+)
+from wf_platform import CapabilitySource, hash_json_schema
+
+from .next_actions import NextActions
+from .operation_context import WorkflowOperationContext
+from .saved_subgraphs import resolve_saved_subgraph_tree, validate_saved_subgraph_tree
+
+
+class WorkflowDeploymentApi:
+    """Saved deployment operations and dependency validation."""
+
+    def __init__(self, context: WorkflowOperationContext) -> None:
+        self.context = context
+
+    def _artifact_store(self):
+        if self.context.artifact_store is None:
+            raise KeyError("workflow artifact store is not configured")
+        return self.context.artifact_store
+
+    async def list_deployments(self) -> dict[str, Any]:
+        if self.context.artifact_store is None:
+            return {"deployments": []}
+        return {
+            "deployments": [
+                _deployment_summary(deployment)
+                for deployment in self.context.artifact_store.list_deployments()
+            ]
+        }
+
+    async def inspect_deployment(self, *, deployment_id: str) -> dict[str, Any]:
+        return self._artifact_store().get_deployment(deployment_id).model_dump(
+            mode="json"
+        )
+
+    async def save_deployment(self, deployment: dict[str, Any]) -> dict[str, Any]:
+        workflow_deployment = WorkflowDeployment.model_validate(deployment)
+        self._artifact_store().save_deployment(workflow_deployment)
+        self.context.events.record_workflow_event(
+            "workflow_deployment_saved",
+            capability_id=f"deployment.{workflow_deployment.id}",
+            payload={
+                "deployment_id": workflow_deployment.id,
+                "artifact_id": workflow_deployment.artifact_id,
+                "artifact_version": workflow_deployment.artifact_version,
+            },
+        )
+        return {
+            "deployment_id": workflow_deployment.id,
+            "artifact_id": workflow_deployment.artifact_id,
+            "artifact_version": workflow_deployment.artifact_version,
+            "saved": True,
+        }
+
+    async def delete_deployment(self, *, deployment_id: str) -> dict[str, Any]:
+        """Delete one mutable deployment environment binding."""
+        self._artifact_store().delete_deployment(deployment_id)
+        self.context.events.record_workflow_event(
+            "workflow_deployment_deleted",
+            capability_id=f"deployment.{deployment_id}",
+            payload={"deployment_id": deployment_id},
+        )
+        return {"deployment_id": deployment_id, "deleted": True}
+
+    async def validate_deployment(
+        self,
+        *,
+        deployment_id: str,
+        live_check: bool = False,
+    ) -> dict[str, Any]:
+        deployment, artifact, diagnostics, tree = self.deployment_validation(
+            deployment_id
+        )
+        if live_check and self.context.live_sources is not None:
+            diagnostics.extend(
+                await self.context.live_sources.deployment_diagnostics(
+                    deployment=deployment,
+                    artifacts=[artifact, *tree.artifacts_by_ref.values()],
+                )
+            )
+        return {
+            "deployment_id": deployment.id,
+            "artifact_id": artifact.id,
+            "artifact_version": artifact.version,
+            "status": "unrunnable" if diagnostics else "runnable",
+            "diagnostics": [
+                diagnostic.model_dump(mode="json") for diagnostic in diagnostics
+            ],
+            "next_actions": NextActions.from_deployment_validation(
+                deployment_id=deployment.id,
+                diagnostics=diagnostics,
+            ).model_dump(mode="json"),
+        }
+
+    def deployment_validation(
+        self,
+        deployment_id: str,
+    ) -> tuple[
+        WorkflowDeployment,
+        WorkflowArtifact,
+        list[DependencyDiagnostic],
+        Any,  # SavedSubgraphTree
+    ]:
+        store = self._artifact_store()
+        deployment = store.get_deployment(deployment_id)
+        artifact = store.get_artifact(
+            deployment.artifact_id,
+            deployment.artifact_version,
+        )
+        available_sources = _available_sources(self.context.capability_sources)
+        diagnostics = validate_deployment_dependencies(
+            artifact=artifact,
+            deployment=deployment,
+            sources=available_sources,
+        )
+        tree = resolve_saved_subgraph_tree(
+            root_artifact=artifact,
+            artifact_store=store,
+        )
+        diagnostics.extend(
+            validate_saved_subgraph_tree(
+                tree=tree,
+                deployment=deployment,
+                sources=available_sources,
+            )
+        )
+        return deployment, artifact, diagnostics, tree
+
+
+def _available_sources(
+    capability_sources: Mapping[str, CapabilitySource],
+) -> list[AvailableSource]:
+    """Convert broker capability sources into artifact validation snapshots."""
+    sources: list[AvailableSource] = []
+    for source in capability_sources.values():
+        node_spec_details = {
+            detail.name: detail
+            for detail in source.as_inventory().capabilities.node_spec_details
+        }
+        capabilities = {
+            capability_name: AvailableCapability(
+                name=capability_name,
+                kind="node_spec",
+                input_schema_hash=hash_json_schema(detail.input_schema),
+                output_schema_hash=hash_json_schema(detail.output_schema),
+            )
+            for spec in source.capabilities.node_specs.values()
+            if (capability_name := _capability_name(spec.name)) is not None
+            if (detail := node_spec_details.get(spec.name)) is not None
+        }
+        capabilities.update(
+            {
+                capability_name: AvailableCapability(
+                    name=capability_name,
+                    kind="reducer",
+                )
+                for reducer in source.capabilities.reducers.values()
+                if (capability_name := _capability_name(reducer.name)) is not None
+            }
+        )
+        sources.append(
+            AvailableSource(
+                id=source.id,
+                enabled=source.enabled,
+                capabilities=capabilities,
+            )
+        )
+    return sources
+
+
+def _capability_name(qualified_name: str) -> str | None:
+    """Return the local name of one qualified capability ref if it is valid."""
+    from wf_api.refs import parse_workflow_surface_capability_id
+    from wf_artifacts import WorkflowCapabilityRef
+
+    try:
+        parsed = parse_workflow_surface_capability_id(qualified_name)
+    except ValueError:
+        return None
+    if isinstance(parsed, WorkflowCapabilityRef):
+        return None
+    return parsed.name
+
+
+def _deployment_summary(deployment: WorkflowDeployment) -> dict[str, Any]:
+    """Return compact deployment metadata for progressive list responses."""
+    return {
+        "id": deployment.id,
+        "artifact_id": deployment.artifact_id,
+        "artifact_version": deployment.artifact_version,
+        "binding_count": len(deployment.binding_map()),
+        "drift_policy": deployment.drift_policy.value,
+    }
+
+
+__all__ = [
+    "WorkflowDeploymentApi",
+]

src/wf_api/operation_context.py

@@ -1,15 +1,17 @@
 from __future__ import annotations
 
-from collections.abc import Mapping
+from collections.abc import Mapping, Sequence
 from dataclasses import dataclass
 from typing import Any, Protocol
 
 from wf_artifacts import (
+    DependencyDiagnostic,
     DraftWorkspaceStore,
     RunStore,
     WorkflowArtifact,
     WorkflowArtifactCatalogEntry,
     WorkflowArtifactStore,
+    WorkflowDeployment,
 )
 from wf_authoring import AsyncRegistryHandler
 from wf_core import RunState
@@ -23,7 +25,17 @@ class WorkflowEventRecorder(Protocol):
     """Records workflow lifecycle events without exposing MCP event types."""
 
     def record_event(self, event: object) -> None:
-        """Record one event object supplied by an adapter-owned event factory."""
+        """Record one adapter-native event object."""
+        ...
+
+    def record_workflow_event(
+        self,
+        event_type: str,
+        *,
+        capability_id: str,
+        payload: dict[str, Any],
+    ) -> None:
+        """Record one workflow lifecycle event by protocol-neutral fields."""
         ...
 
 
@@ -85,8 +97,13 @@ class WorkflowRuntimeRunner(Protocol):
 class WorkflowLiveSourceChecker(Protocol):
     """Optional hook for validating live external source availability."""
 
-    async def available_sources(self) -> list[object]:
+    async def deployment_diagnostics(
-        """Return source availability records understood by the caller."""
+        self,
+        *,
+        deployment: WorkflowDeployment,
+        artifacts: Sequence[WorkflowArtifact],
+    ) -> list[DependencyDiagnostic]:
+        """Return opt-in live-source diagnostics for a deployment tree."""
         ...
 
 

src/wf_mcp/broker/service/workflow_live_checks.py

@@ -0,0 +1,107 @@
+"""MCP-adapter-owned live source diagnostics for deployment validation.
+
+Static deployment validation only checks the last known source catalog.
+This probe intentionally performs live upstream I/O, so MCP tools keep it
+disabled by default and only run it when the caller asks for liveness.
+
+This module owns the MCP-only imports to avoid a circular import:
+handlers.py -> workflow_operation_context.py -> handlers.py
+"""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Sequence
+
+import anyio
+import httpx
+from mcp.client.streamable_http import StreamableHTTPError
+from mcp.shared.exceptions import McpError
+
+from wf_artifacts import DependencyDiagnostic, DiagnosticSeverity, WorkflowArtifact, WorkflowDeployment
+
+from .adapters import require_adapter
+from .core import WfMcpService
+
+LIVE_SOURCE_CHECK_TIMEOUT_SECONDS = 8.0
+_LIVE_SOURCE_CHECK_FAILURES = (
+    KeyError,
+    TimeoutError,
+    OSError,
+    anyio.ClosedResourceError,
+    anyio.EndOfStream,
+    anyio.BrokenResourceError,
+    httpx.HTTPError,
+    McpError,
+    StreamableHTTPError,
+)
+
+
+def _required_live_sources(
+    deployment: WorkflowDeployment,
+    artifacts: Sequence[WorkflowArtifact],
+) -> dict[str, str]:
+    """Return concrete upstream source ids to live-check, with one logical ref."""
+    bindings = deployment.binding_map()
+    required: dict[str, str] = {}
+    for artifact in artifacts:
+        for logical_ref, capability in artifact.required_capability_map().items():
+            source_id = bindings.get(capability.logical_source)
+            if source_id is not None:
+                required.setdefault(source_id, logical_ref)
+    return required
+
+
+async def live_source_diagnostics(
+    service: WfMcpService,
+    *,
+    deployment: WorkflowDeployment,
+    artifacts: Sequence[WorkflowArtifact],
+) -> list[DependencyDiagnostic]:
+    """Return opt-in diagnostics for bound upstream sources that cannot answer.
+
+    Static deployment validation only checks the last known source catalog.
+    This probe intentionally performs live upstream I/O, so MCP tools keep it
+    disabled by default and only run it when the caller asks for liveness.
+    """
+    diagnostics: list[DependencyDiagnostic] = []
+    for source_id, logical_ref in _required_live_sources(deployment, artifacts).items():
+        source = service.capability_sources.get(source_id)
+        if (
+            source is None
+            or not source.enabled
+            or not source.permissions.calls_upstream
+        ):
+            continue
+        try:
+            connection = service.connections.get(source_id)
+            adapter = require_adapter(connection, service.adapters)
+            auth = service.load_auth(source_id)
+            await asyncio.wait_for(
+                adapter.list_tools(connection, auth),
+                timeout=LIVE_SOURCE_CHECK_TIMEOUT_SECONDS,
+            )
+        except _LIVE_SOURCE_CHECK_FAILURES as exc:
+            diagnostics.append(
+                DependencyDiagnostic(
+                    severity=DiagnosticSeverity.ERROR,
+                    code="source_unreachable",
+                    logical_ref=logical_ref,
+                    bound_source=source_id,
+                    message=(
+                        f"Live check for upstream source {source_id!r} failed: "
+                        f"{type(exc).__name__}: {exc}"
+                    ),
+                    repair_hint=(
+                        "Start or reconnect the source, fix its transport/auth "
+                        "configuration, or bind this deployment to another source."
+                    ),
+                )
+            )
+    return diagnostics
+
+
+__all__ = [
+    "LIVE_SOURCE_CHECK_TIMEOUT_SECONDS",
+    "live_source_diagnostics",
+]

src/wf_mcp/broker/service/workflow_operation_context.py

@@ -1,8 +1,10 @@
 from __future__ import annotations
 
+from collections.abc import Sequence
 from dataclasses import dataclass
 from typing import Any
 
+from wf_artifacts import DependencyDiagnostic, WorkflowArtifact, WorkflowDeployment
 from wf_api.operation_context import (
     WorkflowArtifactCataloger,
     WorkflowEventRecorder,
@@ -11,8 +13,10 @@ from wf_api.operation_context import (
     WorkflowRuntimeRunner,
     WorkflowSpecProvider,
 )
+from wf_mcp.events import make_event
 
 from .core import WfMcpService
+from .workflow_live_checks import live_source_diagnostics
 
 
 @dataclass(frozen=True, slots=True)
@@ -24,6 +28,17 @@ class WfMcpWorkflowEventRecorder(WorkflowEventRecorder):
     def record_event(self, event: Any) -> None:
         self.service._record_event(event)  # noqa: SLF001
 
+    def record_workflow_event(
+        self,
+        event_type: str,
+        *,
+        capability_id: str,
+        payload: dict[str, Any],
+    ) -> None:
+        self.service._record_event(  # noqa: SLF001
+            make_event(event_type, capability_id=capability_id, payload=payload)
+        )
+
 
 @dataclass(frozen=True, slots=True)
 class WfMcpWorkflowSpecProvider(WorkflowSpecProvider):
@@ -64,20 +79,21 @@ class WfMcpWorkflowRuntimeRunner(WorkflowRuntimeRunner):
 
 @dataclass(frozen=True, slots=True)
 class WfMcpWorkflowLiveSourceChecker(WorkflowLiveSourceChecker):
-    """Placeholder live source checker while Slice 4A only defines the seam.
+    """Adapter-owned live source checker backed by WfMcpService."""
-
-    See `docs/superpowers/plans/2026-06-01-wf-api-extraction-roadmap.md`.
-    Real live source checks still live in workflow handlers until a later
-    capability-domain extraction can move them without dragging MCP adapters
-    into `wf_api`.
-    """
 
     service: WfMcpService
 
-    async def available_sources(self) -> list[object]:
+    async def deployment_diagnostics(
-        # Existing live source availability logic still lives near handlers.
+        self,
-        # Slice 4A only creates the seam; it does not move live-check behavior.
+        *,
-        return []
+        deployment: WorkflowDeployment,
+        artifacts: Sequence[WorkflowArtifact],
+    ) -> list[DependencyDiagnostic]:
+        return await live_source_diagnostics(
+            self.service,
+            deployment=deployment,
+            artifacts=artifacts,
+        )
 
 
 def context_from_service(service: WfMcpService) -> WorkflowOperationContext:

src/wf_mcp/workflow_surface/handlers.py

@@ -1,15 +1,9 @@
 from __future__ import annotations
 
-import asyncio
 from collections.abc import Sequence
 from dataclasses import asdict
 from typing import TYPE_CHECKING, Any
 
-import anyio
-import httpx
-from mcp.client.streamable_http import StreamableHTTPError
-from mcp.shared.exceptions import McpError
-
 from wf_artifacts import (
     ArtifactKind,
     AvailableCapability,
@@ -22,14 +16,9 @@ from wf_artifacts import (
     WorkflowArtifact,
     WorkflowCapabilityRef,
     WorkflowDeployment,
-    compile_workflow_draft,
-    create_workflow_artifact_from_plan as build_workflow_artifact_from_plan,
-    validate_deployment_dependencies,
 )
 from wf_platform import (
-    CapabilityRef,
     CapabilitySource,
-    NodeSpecInventory,
     hash_json_schema,
 )
 from wf_authoring import build_async_registry
@@ -40,25 +29,22 @@ from wf_core.models.steps import (
 )
 from wf_core.paths import GraphSourcePath
 
+from wf_api.artifacts import WorkflowArtifactApi
+from wf_api.deployments import WorkflowDeploymentApi
 from wf_api.drafts import WorkflowDraftApi
 from wf_api.models import RawWorkflowPlan
 from wf_api.next_actions import NextActions
 from wf_api.refs import parse_workflow_surface_capability_id
 from wf_api.saved_subgraphs import (
-    SavedSubgraphTree,
     direct_wrapper_interrupt_diagnostic,
-    resolve_saved_subgraph_tree,
     saved_subgraph_tree_from_snapshots,
-    validate_saved_subgraph_tree,
 )
 from wf_api.wrapper_hints import (
     workflow_output_schema_for_authoring,
     wrapper_hints_for_capability,
 )
 
-from ..broker.service.adapters import require_adapter
 from ..broker.service.workflow_operation_context import context_from_service
-from ..events import make_event
 from ..shared import matches_query, paged_list_payload
 from .models import TraceRange
 from wf_api.run_lifecycle import (
@@ -71,19 +57,6 @@ from wf_api.run_lifecycle import (
     validate_pinned_resume_environment,
 )
 
-LIVE_SOURCE_CHECK_TIMEOUT_SECONDS = 8.0
-_LIVE_SOURCE_CHECK_FAILURES = (
-    KeyError,
-    TimeoutError,
-    OSError,
-    anyio.ClosedResourceError,
-    anyio.EndOfStream,
-    anyio.BrokenResourceError,
-    httpx.HTTPError,
-    McpError,
-    StreamableHTTPError,
-)
-
 if TYPE_CHECKING:
     from wf_core import RunState
 
@@ -95,7 +68,10 @@ class WorkflowSurfaceHandlers:
 
     def __init__(self, service: WfMcpService) -> None:
         self.service = service
-        self._drafts = WorkflowDraftApi(context_from_service(service))
+        context = context_from_service(service)
+        self._drafts = WorkflowDraftApi(context)
+        self._artifacts = WorkflowArtifactApi(context)
+        self._deployments = WorkflowDeploymentApi(context)
 
     async def list_artifacts(
         self,
@@ -112,27 +88,12 @@ class WorkflowSurfaceHandlers:
         """
         if self.service.artifact_store is None:
             return paged_list_payload("nodes", [], cursor=cursor, limit=limit)
-        entries = [
+        return await self._artifacts.list_artifacts(
-            self.service.workflow_artifact_catalog_entry(artifact).model_dump(
-                mode="json"
-            )
-            for artifact in self.service.artifact_store.list_artifacts()
-            if kind is None or artifact.kind == kind
-        ]
-        entries = [
-            entry
-            for entry in entries
-            if matches_query(
-                entry.get("name"),
-                entry.get("artifact_id"),
-                entry.get("display_name"),
-                entry.get("description"),
-                entry.get("kind"),
             query=query,
+            kind=kind,
+            cursor=cursor,
+            limit=limit,
         )
-        ]
-        entries.sort(key=lambda entry: str(entry["name"]))
-        return paged_list_payload("nodes", entries, cursor=cursor, limit=limit)
 
     async def list_capabilities(
         self,
@@ -406,23 +367,7 @@ class WorkflowSurfaceHandlers:
     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")
-        workflow_artifact = WorkflowArtifact.model_validate(artifact)
+        return await self._artifacts.save_artifact(artifact)
-        self.service.artifact_store.save_artifact(workflow_artifact)
-        self.service._record_event(
-            make_event(
-                "workflow_artifact_saved",
-                capability_id=_artifact_capability_id(workflow_artifact),
-                payload={
-                    "artifact_id": workflow_artifact.id,
-                    "version": workflow_artifact.version,
-                },
-            )
-        )
-        return {
-            "artifact_id": workflow_artifact.id,
-            "version": workflow_artifact.version,
-            "saved": True,
-        }
 
     async def create_artifact_from_plan(
         self,
@@ -440,44 +385,18 @@ class WorkflowSurfaceHandlers:
     ) -> dict[str, Any]:
         if self.service.artifact_store is None:
             raise KeyError("workflow artifact store is not configured")
-        typed_plan = (
+        return await self._artifacts.create_artifact_from_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,
+            plan=plan,
+            outcomes=outcomes,
             kind=kind,
             description=description,
-            plan=typed_plan.model_dump(mode="json", by_alias=True),
+            required_capabilities=required_capabilities,
-            outcomes=tuple(outcomes),
-            required_capabilities={
-                name: RequiredCapability.model_validate(capability)
-                for name, capability in (required_capabilities or {}).items()
-            },
             source_bindings=source_bindings,
-            observed_node_specs=_observed_node_specs(self.service),
             created_from_catalog_version=created_from_catalog_version,
         )
-        self.service.artifact_store.save_artifact(workflow_artifact)
-        self.service._record_event(
-            make_event(
-                "workflow_artifact_saved",
-                capability_id=_artifact_capability_id(workflow_artifact),
-                payload={
-                    "artifact_id": workflow_artifact.id,
-                    "version": workflow_artifact.version,
-                    "created_from_plan": True,
-                },
-            )
-        )
-        return {
-            "artifact_id": workflow_artifact.id,
-            "version": workflow_artifact.version,
-            "saved": True,
-        }
 
     async def validate_draft(self, *, draft: dict[str, Any]) -> dict[str, Any]:
         return await self._drafts.validate_draft(draft=draft)
@@ -501,48 +420,18 @@ class WorkflowSurfaceHandlers:
     ) -> dict[str, Any]:
         if self.service.artifact_store is None:
             raise KeyError("workflow artifact store is not configured")
-        plan = compile_workflow_draft(draft)
+        return await self._artifacts.create_artifact_from_draft(
-        workflow_artifact = build_workflow_artifact_from_plan(
             artifact_id=artifact_id,
             version=version,
             title=title,
+            draft=draft,
+            outcomes=outcomes,
             kind=kind,
             description=description,
-            plan=plan,
+            required_capabilities=required_capabilities,
-            outcomes=tuple(outcomes),
-            required_capabilities={
-                name: RequiredCapability.model_validate(capability)
-                for name, capability in (required_capabilities or {}).items()
-            },
             source_bindings=source_bindings,
-            observed_node_specs=_observed_node_specs(self.service),
             created_from_catalog_version=created_from_catalog_version,
         )
-        self.service.artifact_store.save_artifact(workflow_artifact)
-        self.service._record_event(
-            make_event(
-                "workflow_artifact_saved",
-                capability_id=_artifact_capability_id(workflow_artifact),
-                payload={
-                    "artifact_id": workflow_artifact.id,
-                    "version": workflow_artifact.version,
-                    "created_from_draft": True,
-                },
-            )
-        )
-        required_sources = sorted(
-            {
-                capability.logical_source
-                for capability in workflow_artifact.required_capability_map().values()
-            }
-        )
-        return {
-            "artifact_id": workflow_artifact.id,
-            "version": workflow_artifact.version,
-            "saved": True,
-            "required_logical_sources": required_sources,
-            "suggested_bindings": _suggested_self_bindings(required_sources),
-        }
 
     async def patch_draft(
         self,
@@ -756,24 +645,16 @@ class WorkflowSurfaceHandlers:
         source_bindings: dict[str, str] | None = None,
         created_from_catalog_version: str | None = None,
     ) -> dict[str, Any]:
-        workspace = self._draft_store().get_workspace(workspace_id)
+        if self.service.artifact_store is None:
-        validation = await self.validate_draft(draft=workspace.draft)
+            raise KeyError("workflow artifact store is not configured")
-        if validation["status"] != "valid":
+        return await self._artifacts.create_artifact_from_workspace(
-            return {
+            workspace_id=workspace_id,
-                "saved": False,
-                "workspace_id": workspace_id,
-                "revision": workspace.revision,
-                "status": validation["status"],
-                "diagnostics": validation["diagnostics"],
-            }
-        return await self.create_artifact_from_draft(
             artifact_id=artifact_id,
             version=version,
             title=title,
+            outcomes=outcomes,
             kind=kind,
             description=description,
-            draft=workspace.draft,
-            outcomes=outcomes,
             required_capabilities=required_capabilities,
             source_bindings=source_bindings,
             created_from_catalog_version=created_from_catalog_version,
@@ -793,13 +674,14 @@ class WorkflowSurfaceHandlers:
         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(
+        if self.service.artifact_store is None:
+            raise KeyError("workflow artifact store is not configured")
+        return await self._artifacts.create_wrapper_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,
@@ -811,62 +693,35 @@ class WorkflowSurfaceHandlers:
     ) -> dict[str, Any]:
         if self.service.artifact_store is None:
             raise KeyError("workflow artifact store is not configured")
-        artifact = self.service.artifact_store.get_artifact(artifact_id, version)
+        return await self._artifacts.inspect_artifact(
-        return artifact.model_dump(mode="json")
+            artifact_id=artifact_id,
+            version=version,
+        )
 
     async def list_deployments(self) -> dict[str, Any]:
         if self.service.artifact_store is None:
             return {"deployments": []}
-        return {
+        return await self._deployments.list_deployments()
-            "deployments": [
-                _deployment_summary(deployment)
-                for deployment in self.service.artifact_store.list_deployments()
-            ]
-        }
 
     async def inspect_deployment(self, *, deployment_id: str) -> dict[str, Any]:
         if self.service.artifact_store is None:
             raise KeyError("workflow artifact store is not configured")
-        return self.service.artifact_store.get_deployment(deployment_id).model_dump(
+        return await self._deployments.inspect_deployment(
-            mode="json"
+            deployment_id=deployment_id,
         )
 
     async def save_deployment(self, deployment: dict[str, Any]) -> dict[str, Any]:
         if self.service.artifact_store is None:
             raise KeyError("workflow artifact store is not configured")
-        workflow_deployment = WorkflowDeployment.model_validate(deployment)
+        return await self._deployments.save_deployment(deployment)
-        self.service.artifact_store.save_deployment(workflow_deployment)
-        self.service._record_event(
-            make_event(
-                "workflow_deployment_saved",
-                capability_id=f"deployment.{workflow_deployment.id}",
-                payload={
-                    "deployment_id": workflow_deployment.id,
-                    "artifact_id": workflow_deployment.artifact_id,
-                    "artifact_version": workflow_deployment.artifact_version,
-                },
-            )
-        )
-        return {
-            "deployment_id": workflow_deployment.id,
-            "artifact_id": workflow_deployment.artifact_id,
-            "artifact_version": workflow_deployment.artifact_version,
-            "saved": True,
-        }
 
     async def delete_deployment(self, *, deployment_id: str) -> dict[str, Any]:
         """Delete one mutable deployment environment binding."""
         if self.service.artifact_store is None:
             raise KeyError("workflow artifact store is not configured")
-        self.service.artifact_store.delete_deployment(deployment_id)
+        return await self._deployments.delete_deployment(
-        self.service._record_event(
+            deployment_id=deployment_id,
-            make_event(
-                "workflow_deployment_deleted",
-                capability_id=f"deployment.{deployment_id}",
-                payload={"deployment_id": deployment_id},
         )
-        )
-        return {"deployment_id": deployment_id, "deleted": True}
 
     async def validate_deployment(
         self,
@@ -874,30 +729,12 @@ class WorkflowSurfaceHandlers:
         deployment_id: str,
         live_check: bool = False,
     ) -> dict[str, Any]:
-        deployment, artifact, diagnostics, tree = self._deployment_validation(
+        if self.service.artifact_store is None:
-            deployment_id
+            raise KeyError("workflow artifact store is not configured")
+        return await self._deployments.validate_deployment(
+            deployment_id=deployment_id,
+            live_check=live_check,
         )
-        if live_check:
-            diagnostics.extend(
-                await _live_source_diagnostics(
-                    self.service,
-                    deployment=deployment,
-                    artifacts=[artifact, *tree.artifacts_by_ref.values()],
-                )
-            )
-        return {
-            "deployment_id": deployment.id,
-            "artifact_id": artifact.id,
-            "artifact_version": artifact.version,
-            "status": "unrunnable" if diagnostics else "runnable",
-            "diagnostics": [
-                diagnostic.model_dump(mode="json") for diagnostic in diagnostics
-            ],
-            "next_actions": NextActions.from_deployment_validation(
-                deployment_id=deployment.id,
-                diagnostics=diagnostics,
-            ).model_dump(mode="json"),
-        }
 
     async def run_deployment(
         self,
@@ -906,7 +743,7 @@ class WorkflowSurfaceHandlers:
         workflow_input: dict[str, Any],
         trace_range: TraceRange | None = None,
     ) -> dict[str, Any]:
-        deployment, artifact, diagnostics, tree = self._deployment_validation(
+        deployment, artifact, diagnostics, tree = self._deployments.deployment_validation(
             deployment_id
         )
         if diagnostics:
@@ -1091,41 +928,6 @@ class WorkflowSurfaceHandlers:
             raise KeyError("workflow run store is not configured")
         return self.service.run_store
 
-    def _deployment_validation(
-        self,
-        deployment_id: str,
-    ) -> tuple[
-        WorkflowDeployment,
-        WorkflowArtifact,
-        list[DependencyDiagnostic],
-        SavedSubgraphTree,
-    ]:
-        if self.service.artifact_store is None:
-            raise KeyError("workflow artifact store is not configured")
-        deployment = self.service.artifact_store.get_deployment(deployment_id)
-        artifact = self.service.artifact_store.get_artifact(
-            deployment.artifact_id,
-            deployment.artifact_version,
-        )
-        available_sources = _available_sources(self.service)
-        diagnostics = validate_deployment_dependencies(
-            artifact=artifact,
-            deployment=deployment,
-            sources=available_sources,
-        )
-        tree = resolve_saved_subgraph_tree(
-            root_artifact=artifact,
-            artifact_store=self.service.artifact_store,
-        )
-        diagnostics.extend(
-            validate_saved_subgraph_tree(
-                tree=tree,
-                deployment=deployment,
-                sources=available_sources,
-            )
-        )
-        return deployment, artifact, diagnostics, tree
-
 
 def _available_sources(service: WfMcpService) -> list[AvailableSource]:
     """Convert broker capability sources into artifact validation snapshots."""
@@ -1166,102 +968,6 @@ def _available_sources(service: WfMcpService) -> list[AvailableSource]:
     return sources
 
 
-async def _live_source_diagnostics(
-    service: WfMcpService,
-    *,
-    deployment: WorkflowDeployment,
-    artifacts: Sequence[WorkflowArtifact],
-) -> list[DependencyDiagnostic]:
-    """Return opt-in diagnostics for bound upstream sources that cannot answer.
-
-    Static deployment validation only checks the last known source catalog.
-    This probe intentionally performs live upstream I/O, so MCP tools keep it
-    disabled by default and only run it when the caller asks for liveness.
-    """
-    diagnostics: list[DependencyDiagnostic] = []
-    for source_id, logical_ref in _required_live_sources(deployment, artifacts).items():
-        source = service.capability_sources.get(source_id)
-        if (
-            source is None
-            or not source.enabled
-            or not source.permissions.calls_upstream
-        ):
-            continue
-        try:
-            connection = service.connections.get(source_id)
-            adapter = require_adapter(connection, service.adapters)
-            auth = service.load_auth(source_id)
-            await asyncio.wait_for(
-                adapter.list_tools(connection, auth),
-                timeout=LIVE_SOURCE_CHECK_TIMEOUT_SECONDS,
-            )
-        except _LIVE_SOURCE_CHECK_FAILURES as exc:
-            diagnostics.append(
-                DependencyDiagnostic(
-                    severity=DiagnosticSeverity.ERROR,
-                    code="source_unreachable",
-                    logical_ref=logical_ref,
-                    bound_source=source_id,
-                    message=(
-                        f"Live check for upstream source {source_id!r} failed: "
-                        f"{type(exc).__name__}: {exc}"
-                    ),
-                    repair_hint=(
-                        "Start or reconnect the source, fix its transport/auth "
-                        "configuration, or bind this deployment to another source."
-                    ),
-                )
-            )
-    return diagnostics
-
-
-def _required_live_sources(
-    deployment: WorkflowDeployment,
-    artifacts: Sequence[WorkflowArtifact],
-) -> dict[str, str]:
-    """Return concrete upstream source ids to live-check, with one logical ref."""
-    bindings = deployment.binding_map()
-    required: dict[str, str] = {}
-    for artifact in artifacts:
-        for logical_ref, capability in artifact.required_capability_map().items():
-            source_id = bindings.get(capability.logical_source)
-            if source_id is not None:
-                required.setdefault(source_id, logical_ref)
-    return required
-
-
-def _required_capabilities_for_plan(
-    plan: dict[str, Any],
-    *,
-    source_bindings: dict[str, str] | None,
-    service: WfMcpService,
-) -> dict[str, RequiredCapability]:
-    """Infer a draft dependency summary without persisting an artifact."""
-    artifact = build_workflow_artifact_from_plan(
-        artifact_id="draft_preview",
-        version=1,
-        title="Draft Preview",
-        plan=plan,
-        outcomes=("completed",),
-        source_bindings=source_bindings,
-        observed_node_specs=_observed_node_specs(service),
-    )
-    requirements = artifact.required_capability_map()
-    for node in _plan_nodes(artifact):
-        raw_ref = node.get("node")
-        if not isinstance(raw_ref, str) or raw_ref in requirements:
-            continue
-        try:
-            parsed = CapabilityRef.parse(raw_ref)
-        except ValueError:
-            continue
-        requirements[raw_ref] = RequiredCapability(
-            ref=parsed,
-            kind="node_spec",
-        )
-    return requirements
-
-
 def _required_capability_payloads(
     requirements: dict[str, RequiredCapability],
 ) -> dict[str, dict[str, Any]]:
@@ -1271,24 +977,6 @@ def _required_capability_payloads(
     }
 
 
-def _suggested_self_bindings(required_sources: Sequence[str]) -> dict[str, str]:
-    """Suggest local bindings for built-in sources that deploy to themselves."""
-    return {
-        source: source for source in required_sources if source in {"wf.std", "wf.mcp"}
-    }
-
-
-def _observed_node_specs(service: WfMcpService) -> dict[str, NodeSpecInventory]:
-    """Project current executable specs into serializable observed contracts."""
-    observed: dict[str, NodeSpecInventory] = {}
-    for source in service.capability_sources.values():
-        inventory = source.as_inventory()
-        observed.update(
-            {detail.name: detail for detail in inventory.capabilities.node_spec_details}
-        )
-    return observed
-
-
 def _schema_field_names(schema: dict[str, Any]) -> list[str]:
     """Return top-level JSON object property names for compact discovery rows."""
     properties = schema.get("properties")
@@ -1361,11 +1049,6 @@ def _plan_field(artifact: WorkflowArtifact, field_name: str) -> Any:
         ) from exc
 
 
-def _plan_nodes(artifact: WorkflowArtifact) -> list[dict[str, Any]]:
-    nodes = artifact.plan.get("nodes", [])
-    return [node for node in nodes if isinstance(node, dict)]
-
-
 def _run_payload(
     *,
     deployment: WorkflowDeployment,
@@ -1427,14 +1110,3 @@ def _interrupt_payload(run: RunState) -> dict[str, Any] | None:
         if hasattr(workflow_ref, "model_dump"):
             route["workflow_ref"] = workflow_ref.model_dump(mode="json")
     return payload
-
-
-def _deployment_summary(deployment: WorkflowDeployment) -> dict[str, Any]:
-    """Return compact deployment metadata for progressive list responses."""
-    return {
-        "id": deployment.id,
-        "artifact_id": deployment.artifact_id,
-        "artifact_version": deployment.artifact_version,
-        "binding_count": len(deployment.binding_map()),
-        "drift_policy": deployment.drift_policy.value,
-    }

tests/wf_api/test_artifact_api.py

@@ -0,0 +1,273 @@
+"""Tests for wf_api.artifacts module."""
+
+from __future__ import annotations
+
+import asyncio
+from dataclasses import replace
+from typing import Any
+
+from wf_artifacts import FileWorkflowArtifactStore, RequiredCapability, WorkflowArtifact
+from wf_api.artifacts import WorkflowArtifactApi
+from wf_mcp.broker import WfMcpService
+from wf_mcp.models import ConnectionConfig
+from wf_mcp.storage import FileStore
+from wf_mcp.workflow_surface import WorkflowSurfaceHandlers
+from wf_mcp.broker.service.workflow_operation_context import context_from_service
+
+from tests.wf_mcp.test_support import echo_tool, local_temp_root
+
+
+def _echo_draft() -> dict[str, Any]:
+    return {
+        "name": "echo",
+        "input_schema": {
+            "type": "object",
+            "properties": {"text": {"type": "string"}},
+            "required": ["text"],
+        },
+        "state_schema": {"fields": {"echoed": {"type": "string"}}},
+        "output_schema": {
+            "type": "object",
+            "properties": {"echoed": {"type": "string"}},
+            "required": ["echoed"],
+        },
+        "start": "echo",
+        "steps": {
+            "echo": {
+                "use": "demo.personal.echo_tool",
+                "input": [
+                    {
+                        "target": {"root": "local", "parts": ["text"]},
+                        "path": {"root": "input", "parts": ["text"]},
+                    }
+                ],
+                "output": [
+                    {
+                        "source": {"root": "local", "parts": ["echoed"]},
+                        "target": {"root": "state", "parts": ["echoed"]},
+                    }
+                ],
+            }
+        },
+        "routes": {"echo": {"ok": "__end__"}},
+    }
+
+
+def _echo_artifact() -> WorkflowArtifact:
+    plan: dict[str, Any] = {
+        "name": "echo",
+        "input_schema": {
+            "type": "object",
+            "properties": {"text": {"type": "string"}},
+            "required": ["text"],
+        },
+        "state_schema": {"fields": {"echoed": {"type": "string"}}},
+        "output_schema": {
+            "type": "object",
+            "properties": {"echoed": {"type": "string"}},
+            "required": ["echoed"],
+        },
+        "start": "echo",
+        "nodes": [
+            {
+                "id": "echo",
+                "type": "node",
+                "node": "demo.personal.echo_tool",
+                "input": [
+                    {
+                        "path": {"root": "input", "parts": ["text"]},
+                        "target": {"root": "local", "parts": ["text"]},
+                    }
+                ],
+                "output": [
+                    {
+                        "source": {"root": "local", "parts": ["echoed"]},
+                        "target": {"root": "state", "parts": ["echoed"]},
+                    }
+                ],
+            }
+        ],
+        "edges": [{"from": "echo", "outcome": "ok", "to": "__end__"}],
+    }
+    return WorkflowArtifact(
+        id="echo",
+        version=1,
+        title="Echo",
+        input_schema=plan["input_schema"],
+        output_schema=plan["output_schema"],
+        outcomes=("completed",),
+        plan=plan,
+        required_capabilities={
+            "demo.echo_tool": RequiredCapability(
+                ref="demo.echo_tool",
+                kind="node_spec",
+            )
+        },
+    )
+
+
+def _artifact_api(
+    artifact_store: FileWorkflowArtifactStore,
+    *,
+    register_echo: bool = False,
+) -> tuple[WorkflowArtifactApi, WfMcpService]:
+    service = WfMcpService(
+        store=FileStore(artifact_store.root / "artifacts_mcp" / str(id(artifact_store))),
+        artifact_store=artifact_store,
+    )
+    if register_echo:
+        service.register_connection(
+            ConnectionConfig(id="demo.personal", server="demo", account="personal")
+        )
+        service.register_specs("demo.personal", echo_tool)
+    context = context_from_service(service)
+    return WorkflowArtifactApi(context), service
+
+
+def test_save_artifact_stores_and_returns_saved() -> None:
+    artifact_store = FileWorkflowArtifactStore(local_temp_root() / "artifacts_save")
+    api, _service = _artifact_api(artifact_store)
+
+    result = asyncio.run(api.save_artifact(_echo_artifact().model_dump(mode="json")))
+
+    assert result["saved"] is True
+    assert result["artifact_id"] == "echo"
+    assert result["version"] == 1
+    saved = artifact_store.get_artifact("echo", 1)
+    assert saved.id == "echo"
+
+
+def test_list_artifacts_returns_empty_page_without_artifact_store() -> None:
+    artifact_store = FileWorkflowArtifactStore(local_temp_root() / "artifacts_no_store")
+    _api, service = _artifact_api(artifact_store)
+    context = replace(context_from_service(service), artifact_store=None)
+    api = WorkflowArtifactApi(context)
+
+    result = asyncio.run(api.list_artifacts())
+
+    assert result["nodes"] == []
+    assert result["next_cursor"] is None
+    assert result["total"] == 0
+
+
+def test_create_artifact_from_plan_saves_with_observed_node_specs() -> None:
+    artifact_store = FileWorkflowArtifactStore(
+        local_temp_root() / "artifacts_from_plan"
+    )
+    api, _service = _artifact_api(artifact_store, register_echo=True)
+
+    result = asyncio.run(
+        api.create_artifact_from_plan(
+            artifact_id="echo",
+            version=1,
+            title="Echo",
+            plan=_echo_artifact().plan,
+            outcomes=("completed",),
+        )
+    )
+
+    assert result["saved"] is True
+    assert result["artifact_id"] == "echo"
+    saved = artifact_store.get_artifact("echo", 1)
+    assert saved.id == "echo"
+
+
+def test_create_artifact_from_workspace_returns_saved_false_when_invalid() -> None:
+    artifact_store = FileWorkflowArtifactStore(
+        local_temp_root() / "artifacts_workspace_invalid"
+    )
+    api, service = _artifact_api(artifact_store, register_echo=True)
+    draft = _echo_draft()
+    draft["routes"]["echo"] = {"typo": "__end__"}
+    from wf_api.drafts import WorkflowDraftApi
+
+    context = context_from_service(service)
+    drafts_api = WorkflowDraftApi(context)
+    asyncio.run(
+        drafts_api.create_draft_workspace(
+            workspace_id="echo_ws",
+            draft=draft,
+        )
+    )
+
+    result = asyncio.run(
+        api.create_artifact_from_workspace(
+            workspace_id="echo_ws",
+            artifact_id="echo",
+            version=1,
+            title="Echo",
+            outcomes=("completed",),
+        )
+    )
+
+    assert result["saved"] is False
+    assert result["status"] == "invalid"
+
+
+def test_create_wrapper_from_workspace_saves_kind_wrapper() -> None:
+    artifact_store = FileWorkflowArtifactStore(
+        local_temp_root() / "artifacts_wrapper_workspace"
+    )
+    api, service = _artifact_api(artifact_store, register_echo=True)
+    from wf_api.drafts import WorkflowDraftApi
+
+    context = context_from_service(service)
+    drafts_api = WorkflowDraftApi(context)
+    asyncio.run(
+        drafts_api.create_draft_workspace(
+            workspace_id="echo_ws",
+            draft=_echo_draft(),
+        )
+    )
+
+    result = asyncio.run(
+        api.create_wrapper_from_workspace(
+            workspace_id="echo_ws",
+            artifact_id="echo_wrapper",
+            version=1,
+            title="Echo Wrapper",
+            outcomes=("completed",),
+        )
+    )
+
+    assert result["saved"] is True
+    saved = artifact_store.get_artifact("echo_wrapper", 1)
+    assert saved.kind == "wrapper"
+
+
+def test_inspect_artifact_returns_stable_fields() -> None:
+    artifact_store = FileWorkflowArtifactStore(
+        local_temp_root() / "artifacts_inspect"
+    )
+    api, _service = _artifact_api(artifact_store)
+    artifact_store.save_artifact(_echo_artifact())
+
+    result = asyncio.run(api.inspect_artifact(artifact_id="echo", version=1))
+
+    assert result["id"] == "echo"
+    assert result["version"] == 1
+    assert result["title"] == "Echo"
+    assert "plan" in result
+
+
+def test_handler_delegation_for_inspect_artifact() -> None:
+    """WorkflowSurfaceHandlers.inspect_artifact delegates to WorkflowArtifactApi."""
+    artifact_store = FileWorkflowArtifactStore(
+        local_temp_root() / "artifacts_delegation"
+    )
+    service = WfMcpService(
+        store=FileStore(artifact_store.root / "delegation_mcp"),
+        artifact_store=artifact_store,
+    )
+    artifact_store.save_artifact(_echo_artifact())
+
+    h = WorkflowSurfaceHandlers(service)
+    context = context_from_service(service)
+    api = WorkflowArtifactApi(context)
+
+    handler_result = asyncio.run(h.inspect_artifact(artifact_id="echo", version=1))
+    api_result = asyncio.run(api.inspect_artifact(artifact_id="echo", version=1))
+
+    assert handler_result["id"] == api_result["id"]
+    assert handler_result["version"] == api_result["version"]
+    assert handler_result["title"] == api_result["title"]

tests/wf_api/test_deployment_api.py

@@ -0,0 +1,264 @@
+"""Tests for wf_api.deployments module."""
+
+from __future__ import annotations
+
+import asyncio
+from dataclasses import replace
+from typing import Any, cast
+
+from wf_artifacts import (
+    FileWorkflowArtifactStore,
+    RequiredCapability,
+    WorkflowArtifact,
+    WorkflowDeployment,
+)
+from wf_api.deployments import WorkflowDeploymentApi
+from wf_mcp.broker import WfMcpService
+from wf_mcp.models import AuthRecord, ConnectionConfig
+from wf_mcp.capabilities import DiscoveredTool
+from wf_mcp.sdk import BackendAdapter
+from wf_mcp.storage import FileStore
+from wf_mcp.workflow_surface import WorkflowSurfaceHandlers
+from wf_mcp.broker.service.workflow_operation_context import context_from_service
+
+from tests.wf_mcp.test_support import echo_tool, local_temp_root
+
+
+def _echo_artifact() -> WorkflowArtifact:
+    plan: dict[str, Any] = {
+        "name": "echo",
+        "input_schema": {
+            "type": "object",
+            "properties": {"text": {"type": "string"}},
+            "required": ["text"],
+        },
+        "state_schema": {"fields": {"echoed": {"type": "string"}}},
+        "output_schema": {
+            "type": "object",
+            "properties": {"echoed": {"type": "string"}},
+            "required": ["echoed"],
+        },
+        "start": "echo",
+        "nodes": [
+            {
+                "id": "echo",
+                "type": "node",
+                "node": "demo.personal.echo_tool",
+                "input": [
+                    {
+                        "path": {"root": "input", "parts": ["text"]},
+                        "target": {"root": "local", "parts": ["text"]},
+                    }
+                ],
+                "output": [
+                    {
+                        "source": {"root": "local", "parts": ["echoed"]},
+                        "target": {"root": "state", "parts": ["echoed"]},
+                    }
+                ],
+            }
+        ],
+        "edges": [{"from": "echo", "outcome": "ok", "to": "__end__"}],
+    }
+    return WorkflowArtifact(
+        id="echo",
+        version=1,
+        title="Echo",
+        input_schema=plan["input_schema"],
+        output_schema=plan["output_schema"],
+        outcomes=("completed",),
+        plan=plan,
+        required_capabilities={
+            "demo.echo_tool": RequiredCapability(
+                ref="demo.echo_tool",
+                kind="node_spec",
+            )
+        },
+    )
+
+
+def _deployment_api(
+    artifact_store: FileWorkflowArtifactStore,
+    *,
+    register_echo: bool = False,
+) -> tuple[WorkflowDeploymentApi, WfMcpService]:
+    service = WfMcpService(
+        store=FileStore(artifact_store.root / "deployments_mcp" / str(id(artifact_store))),
+        artifact_store=artifact_store,
+    )
+    if register_echo:
+        service.register_connection(
+            ConnectionConfig(id="demo.personal", server="demo", account="personal")
+        )
+        service.register_specs("demo.personal", echo_tool)
+    context = context_from_service(service)
+    return WorkflowDeploymentApi(context), service
+
+
+def test_save_deployment_stores_and_returns_stable_fields() -> None:
+    artifact_store = FileWorkflowArtifactStore(local_temp_root() / "deploy_save")
+    api, _service = _deployment_api(artifact_store)
+
+    result = asyncio.run(
+        api.save_deployment(
+            WorkflowDeployment(
+                id="echo.personal",
+                artifact_id="echo",
+                artifact_version=1,
+                bindings=[{"logical_source": "demo", "concrete_source": "demo.personal"}],
+            ).model_dump(mode="json")
+        )
+    )
+
+    assert result["saved"] is True
+    assert result["deployment_id"] == "echo.personal"
+    assert result["artifact_id"] == "echo"
+    assert result["artifact_version"] == 1
+
+
+def test_list_deployments_returns_compact_summaries() -> None:
+    artifact_store = FileWorkflowArtifactStore(local_temp_root() / "deploy_list")
+    api, _service = _deployment_api(artifact_store)
+    artifact_store.save_deployment(
+        WorkflowDeployment(
+            id="echo.personal",
+            artifact_id="echo",
+            artifact_version=1,
+            bindings=[{"logical_source": "demo", "concrete_source": "demo.personal"}],
+        )
+    )
+
+    result = asyncio.run(api.list_deployments())
+
+    assert len(result["deployments"]) == 1
+    assert result["deployments"][0]["id"] == "echo.personal"
+    assert result["deployments"][0]["binding_count"] == 1
+    assert "bindings" not in result["deployments"][0]
+
+
+def test_list_deployments_returns_empty_without_artifact_store() -> None:
+    artifact_store = FileWorkflowArtifactStore(local_temp_root() / "deploy_no_store")
+    _api, service = _deployment_api(artifact_store)
+    context = replace(context_from_service(service), artifact_store=None)
+    api = WorkflowDeploymentApi(context)
+
+    result = asyncio.run(api.list_deployments())
+
+    assert result["deployments"] == []
+
+
+def test_delete_deployment_removes_one() -> None:
+    artifact_store = FileWorkflowArtifactStore(local_temp_root() / "deploy_delete")
+    api, _service = _deployment_api(artifact_store)
+    artifact_store.save_deployment(
+        WorkflowDeployment(
+            id="echo.personal",
+            artifact_id="echo",
+            artifact_version=1,
+            bindings=[{"logical_source": "demo", "concrete_source": "demo.personal"}],
+        )
+    )
+
+    result = asyncio.run(api.delete_deployment(deployment_id="echo.personal"))
+
+    assert result["deployment_id"] == "echo.personal"
+    assert result["deleted"] is True
+    assert artifact_store.list_deployments() == []
+
+
+def test_validate_deployment_returns_runnable_for_valid_binding() -> None:
+    artifact_store = FileWorkflowArtifactStore(
+        local_temp_root() / "deploy_validate_runnable"
+    )
+    api, service = _deployment_api(artifact_store, register_echo=True)
+    artifact_store.save_artifact(_echo_artifact())
+    artifact_store.save_deployment(
+        WorkflowDeployment(
+            id="echo.personal",
+            artifact_id="echo",
+            artifact_version=1,
+            bindings=[{"logical_source": "demo", "concrete_source": "demo.personal"}],
+        )
+    )
+
+    result = asyncio.run(
+        api.validate_deployment(deployment_id="echo.personal", live_check=False)
+    )
+
+    assert result["status"] == "runnable"
+    assert result["diagnostics"] == []
+
+
+class FailingLivenessAdapter:
+    async def list_tools(
+        self,
+        connection: ConnectionConfig,
+        auth: AuthRecord | None,
+    ) -> list[DiscoveredTool]:
+        raise OSError("stdio process exited")
+
+
+def test_validate_deployment_live_check_calls_live_checker() -> None:
+    artifact_store = FileWorkflowArtifactStore(
+        local_temp_root() / "deploy_validate_live"
+    )
+    api, service = _deployment_api(artifact_store, register_echo=True)
+    artifact_store.save_artifact(_echo_artifact())
+    artifact_store.save_deployment(
+        WorkflowDeployment(
+            id="echo.personal",
+            artifact_id="echo",
+            artifact_version=1,
+            bindings=[{"logical_source": "demo", "concrete_source": "demo.personal"}],
+        )
+    )
+    service.register_adapter(
+        "demo",
+        cast(BackendAdapter, FailingLivenessAdapter()),
+    )
+
+    result = asyncio.run(
+        api.validate_deployment(deployment_id="echo.personal", live_check=True)
+    )
+
+    assert result["status"] == "unrunnable"
+    assert result["diagnostics"][0]["code"] == "source_unreachable"
+
+
+def test_handler_delegation_for_validate_deployment() -> None:
+    """WorkflowSurfaceHandlers.validate_deployment delegates to WorkflowDeploymentApi."""
+    artifact_store = FileWorkflowArtifactStore(
+        local_temp_root() / "deploy_delegation"
+    )
+    service = WfMcpService(
+        store=FileStore(artifact_store.root / "delegation_mcp"),
+        artifact_store=artifact_store,
+    )
+    artifact_store.save_artifact(_echo_artifact())
+    artifact_store.save_deployment(
+        WorkflowDeployment(
+            id="echo.personal",
+            artifact_id="echo",
+            artifact_version=1,
+            bindings=[{"logical_source": "demo", "concrete_source": "demo.personal"}],
+        )
+    )
+
+    h = WorkflowSurfaceHandlers(service)
+    context = context_from_service(service)
+    api = WorkflowDeploymentApi(context)
+
+    handler_result = asyncio.run(
+        h.validate_deployment(deployment_id="echo.personal")
+    )
+    api_result = asyncio.run(
+        api.validate_deployment(deployment_id="echo.personal")
+    )
+
+    assert handler_result["status"] == api_result["status"]
+    assert len(handler_result["diagnostics"]) == len(api_result["diagnostics"])
+    if handler_result["diagnostics"]:
+        assert (
+            handler_result["diagnostics"][0]["code"]
+            == api_result["diagnostics"][0]["code"]
+        )

tests/wf_api/test_operation_context.py

@@ -74,3 +74,25 @@ def test_context_from_service_delegates_specs_and_events(tmp_path: Path) -> None
     operation_context.events.record_event(event)
 
     assert cli_context.service.list_events()[-1] is event
+
+
+def test_context_from_service_record_workflow_event(tmp_path: Path) -> None:
+    config_path = tmp_path / "wf_mcp.config.json"
+    config_path.write_text(
+        json.dumps({"store_root": ".wf_mcp_store", "connections": []}),
+        encoding="utf-8",
+    )
+    cli_context = load_cli_context(config_path)
+    operation_context = context_from_service(cli_context.service)
+
+    operation_context.events.record_workflow_event(
+        "workflow_artifact_saved",
+        capability_id="workflow.demo.v1",
+        payload={"artifact_id": "demo", "version": 1},
+    )
+
+    recorded = cli_context.service.list_events()[-1]
+    assert recorded.kind == "workflow_artifact_saved"
+    assert recorded.capability_id == "workflow.demo.v1"
+    assert recorded.payload["artifact_id"] == "demo"
+    assert recorded.payload["version"] == 1