organize the Helper Functions

those who make the program. the helper functions. they deserve better Ahh

src/wf_api/__init__.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from .listing import matches_query, paged_list_payload
 from .artifacts import WorkflowArtifactApi
 from .backend import TraceRange, WorkflowApiBackend
 from .capabilities import WorkflowCapabilityApi
@@ -41,6 +42,8 @@ from .runtime_dependencies import RuntimeDependencies, resolve_runtime_dependenc
 
 __all__ = [
     "DEFAULT_CALL_STEP_ID",
+    "matches_query",
+    "paged_list_payload",
     "DEFAULT_ERROR_OUTCOME",
     "DEFAULT_ERROR_STEP_ID",
     "DEFAULT_OK_OUTCOME",

src/wf_api/artifact_plans.py

@@ -0,0 +1,41 @@
+from __future__ import annotations
+
+from typing import Any
+
+from wf_artifacts import WorkflowArtifact
+
+from .models import RawWorkflowPlan
+
+
+def raw_plan_from_artifact(artifact: WorkflowArtifact) -> RawWorkflowPlan:
+    """Validate the stored raw workflow plan shape expected by runtime calls."""
+    return RawWorkflowPlan.model_validate(
+        {
+            "name": plan_field(artifact, "name"),
+            "input_schema": plan_field(artifact, "input_schema"),
+            "state_schema": plan_field(artifact, "state_schema"),
+            "output_schema": plan_field(artifact, "output_schema"),
+            "outcomes": artifact.plan.get("outcomes", ["ok"]),
+            "output": artifact.plan.get("output", []),
+            "start": plan_field(artifact, "start"),
+            "nodes": plan_field(artifact, "nodes"),
+            "edges": plan_field(artifact, "edges"),
+        }
+    )
+
+
+def plan_field(artifact: WorkflowArtifact, field_name: str) -> Any:
+    """Return one required raw-plan field with an artifact-specific error."""
+    try:
+        return artifact.plan[field_name]
+    except KeyError as exc:
+        raise ValueError(
+            f"workflow artifact {artifact.id}@{artifact.version} "
+            f"is missing plan field {field_name!r}"
+        ) from exc
+
+
+def plan_nodes(artifact: WorkflowArtifact) -> list[dict[str, Any]]:
+    """Return only dict-shaped node entries from a saved raw plan."""
+    nodes = artifact.plan.get("nodes", [])
+    return [node for node in nodes if isinstance(node, dict)]

src/wf_api/artifact_refs.py

@@ -0,0 +1,13 @@
+from __future__ import annotations
+
+from wf_artifacts import WorkflowArtifact, WorkflowCapabilityRef
+
+
+def artifact_capability_id(artifact: WorkflowArtifact) -> str:
+    """Return the stable workflow capability name for a saved artifact."""
+    return str(
+        WorkflowCapabilityRef(
+            artifact_id=artifact.id,
+            version=artifact.version,
+        )
+    )

src/wf_api/artifacts.py

@@ -7,49 +7,22 @@ WorkflowOperationContext so this module stays protocol-neutral.
 from __future__ import annotations
 
 from collections.abc import Sequence
-from typing import Any, TypeVar
+from typing import Any
 
 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 .artifact_refs import artifact_capability_id
+from .capability_requirements import observed_node_specs
 from .drafts import WorkflowDraftApi
+from .listing import matches_query, paged_list_payload
 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.
@@ -81,7 +54,7 @@ class WorkflowArtifactApi:
         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)
+            return paged_list_payload("nodes", [], cursor=cursor, limit=limit)
         entries = [
             self.context.artifacts.workflow_artifact_catalog_entry(artifact).model_dump(
                 mode="json"
@@ -92,7 +65,7 @@ class WorkflowArtifactApi:
         entries = [
             entry
             for entry in entries
-            if _matches_query(
+            if matches_query(
                 entry.get("name"),
                 entry.get("artifact_id"),
                 entry.get("display_name"),
@@ -102,14 +75,14 @@ class WorkflowArtifactApi:
             )
         ]
         entries.sort(key=lambda entry: str(entry["name"]))
-        return _paged_list_payload("nodes", entries, cursor=cursor, limit=limit)
+        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),
+            capability_id=artifact_capability_id(workflow_artifact),
             payload={
                 "artifact_id": workflow_artifact.id,
                 "version": workflow_artifact.version,
@@ -153,13 +126,13 @@ class WorkflowArtifactApi:
                 for name, capability in (required_capabilities or {}).items()
             },
             source_bindings=source_bindings,
-            observed_node_specs=_observed_node_specs(self.context),
+            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),
+            capability_id=artifact_capability_id(workflow_artifact),
             payload={
                 "artifact_id": workflow_artifact.id,
                 "version": workflow_artifact.version,
@@ -202,13 +175,13 @@ class WorkflowArtifactApi:
                 for name, capability in (required_capabilities or {}).items()
             },
             source_bindings=source_bindings,
-            observed_node_specs=_observed_node_specs(self.context),
+            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),
+            capability_id=artifact_capability_id(workflow_artifact),
             payload={
                 "artifact_id": workflow_artifact.id,
                 "version": workflow_artifact.version,
@@ -303,15 +276,6 @@ class WorkflowArtifactApi:
         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 {
@@ -319,34 +283,6 @@ def _suggested_self_bindings(required_sources: Sequence[str]) -> dict[str, str]:
     }
 
 
-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/capabilities.py

@@ -6,7 +6,6 @@ from typing import Any
 from wf_artifacts import (
     DependencyDiagnostic,
     DiagnosticSeverity,
-    RequiredCapability,
     WorkflowArtifact,
     WorkflowCapabilityRef,
 )
@@ -14,10 +13,13 @@ from wf_authoring import build_async_registry
 from wf_core import RuntimeContext
 from wf_core.models.steps import InputBinding, OutputBinding
 from wf_core.paths import GraphSourcePath
-from wf_platform import CapabilitySource, page_items
+from wf_platform import CapabilitySource
 
+from .artifact_plans import raw_plan_from_artifact
+from .artifact_refs import artifact_capability_id
+from .capability_requirements import required_capability_payloads
 from .drafts import WorkflowDraftApi
-from .models import RawWorkflowPlan
+from .listing import matches_query, paged_list_payload
 from .next_actions import NextActions
 from .operation_context import WorkflowOperationContext
 from .refs import parse_workflow_surface_capability_id
@@ -28,26 +30,6 @@ from .wrapper_hints import (
 )
 
 
-def _matches_query(*values: object, query: str | None) -> bool:
-    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[dict[str, Any]],
-    *,
-    cursor: str | None,
-    limit: int,
-) -> dict[str, Any]:
-    page = page_items(items, cursor=cursor, limit=limit)
-    return {key: list(page.items), "next_cursor": page.next_cursor, "total": page.total}
-
-
 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")
@@ -56,25 +38,6 @@ def _schema_field_names(schema: dict[str, Any]) -> list[str]:
     return sorted(str(name) for name in properties)
 
 
-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,
-        )
-    )
-
-
-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 _source_id_for_capability(
     sources: Mapping[str, CapabilitySource],
     qualified_name: str,
@@ -86,33 +49,6 @@ def _source_id_for_capability(
     return None
 
 
-def _raw_plan_from_artifact(artifact: WorkflowArtifact) -> RawWorkflowPlan:
-    """Validate the stored plan shape expected by the broker workflow runner."""
-    return RawWorkflowPlan.model_validate(
-        {
-            "name": _plan_field(artifact, "name"),
-            "input_schema": _plan_field(artifact, "input_schema"),
-            "state_schema": _plan_field(artifact, "state_schema"),
-            "output_schema": _plan_field(artifact, "output_schema"),
-            "outcomes": artifact.plan.get("outcomes", ["ok"]),
-            "output": artifact.plan.get("output", []),
-            "start": _plan_field(artifact, "start"),
-            "nodes": _plan_field(artifact, "nodes"),
-            "edges": _plan_field(artifact, "edges"),
-        }
-    )
-
-
-def _plan_field(artifact: WorkflowArtifact, field_name: str) -> Any:
-    try:
-        return artifact.plan[field_name]
-    except KeyError as exc:
-        raise ValueError(
-            f"workflow artifact {artifact.id}@{artifact.version} "
-            f"is missing plan field {field_name!r}"
-        ) from exc
-
-
 def _draft_name_from_capability(capability_name: str) -> str:
     """Return a stable draft name when caller does not provide one."""
     return capability_name.replace(".", "_").replace("-", "_")
@@ -158,7 +94,7 @@ class WorkflowCapabilityApi:
             if source.enabled and source.visibility.planner
             if source_id is None or source.id == source_id
             for detail in source.as_inventory().capabilities.node_spec_details
-            if _matches_query(
+            if matches_query(
                 detail.name,
                 detail.description,
                 query=query,
@@ -168,7 +104,7 @@ class WorkflowCapabilityApi:
             self._wrapper_capability_summaries(query=query, source_id=source_id)
         )
         capabilities.sort(key=lambda capability: capability["name"])
-        return _paged_list_payload(
+        return paged_list_payload(
             "capabilities",
             capabilities,
             cursor=cursor,
@@ -298,8 +234,8 @@ class WorkflowCapabilityApi:
         for artifact in self.context.artifact_store.list_artifacts():
             if artifact.kind != "wrapper":
                 continue
-            name = _artifact_capability_id(artifact)
+            name = artifact_capability_id(artifact)
-            if not _matches_query(
+            if not matches_query(
                 name,
                 artifact.description,
                 query=query,
@@ -331,7 +267,7 @@ class WorkflowCapabilityApi:
         if artifact is None:
             return None
         return {
-            "name": _artifact_capability_id(artifact),
+            "name": artifact_capability_id(artifact),
             "source_id": "workflow",
             "kind": "wrapper_artifact",
             "artifact_id": artifact.id,
@@ -342,11 +278,11 @@ class WorkflowCapabilityApi:
             "is_async": True,
             "input_schema": artifact.input_schema,
             "output_schema": artifact.output_schema,
-            "required_capabilities": _required_capability_payloads(
+            "required_capabilities": required_capability_payloads(
                 artifact.required_capability_map()
             ),
             "wrapper_hints": wrapper_hints_for_capability(
-                capability_name=_artifact_capability_id(artifact),
+                capability_name=artifact_capability_id(artifact),
                 input_schema=artifact.input_schema,
                 output_schema=artifact.output_schema,
                 outcomes=list(artifact.outcomes),
@@ -368,7 +304,7 @@ class WorkflowCapabilityApi:
         # Direct capability calls remain wrapper-only. Full saved workflows run
         # through deployments, where native subgraph dependencies and bindings
         # are prepared before core execution.
-        plan = _raw_plan_from_artifact(artifact)
+        plan = raw_plan_from_artifact(artifact)
         deployment = None
         if deployment_id is not None:
             if self.context.artifact_store is None:
@@ -389,7 +325,7 @@ class WorkflowCapabilityApi:
             artifact=artifact,
         )
         return {
-            "qualified_name": _artifact_capability_id(artifact),
+            "qualified_name": artifact_capability_id(artifact),
             "source_id": "workflow",
             "kind": "wrapper_artifact",
             "deployment_id": deployment_id,

src/wf_api/capability_requirements.py

@@ -0,0 +1,67 @@
+from __future__ import annotations
+
+from typing import Any
+
+from wf_artifacts import (
+    RequiredCapability,
+    create_workflow_artifact_from_plan as build_workflow_artifact_from_plan,
+)
+from wf_platform import CapabilityRef, NodeSpecInventory
+
+from .artifact_plans import plan_nodes
+from .operation_context import WorkflowOperationContext
+
+
+def required_capability_payloads(
+    requirements: dict[str, RequiredCapability],
+) -> dict[str, dict[str, Any]]:
+    """Return deterministic JSON payloads for required capabilities."""
+    return {
+        name: capability.model_dump(mode="json")
+        for name, capability in sorted(requirements.items())
+    }
+
+
+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 required_capabilities_for_plan(
+    plan: dict[str, Any],
+    *,
+    source_bindings: dict[str, str] | None,
+    context: WorkflowOperationContext,
+) -> 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(context),
+    )
+    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

src/wf_api/drafts.py

@@ -5,11 +5,8 @@ from typing import Any
 
 from wf_artifacts import (
     DraftWorkspaceStore,
-    RequiredCapability,
-    WorkflowArtifact,
     compile_workflow_draft,
     create_draft_workspace as create_draft_workspace_record,
-    create_workflow_artifact_from_plan as build_workflow_artifact_from_plan,
     get_draft_workspace as get_draft_workspace_record,
     patch_draft_workspace as patch_draft_workspace_record,
     patch_workflow_draft,
@@ -22,8 +19,11 @@ from wf_core.models.steps import (
     OutputBinding,
 )
 from wf_core.paths import GraphSourcePath, LocalPath, StatePath
-from wf_platform import CapabilityRef, NodeSpecInventory
 
+from .capability_requirements import (
+    required_capabilities_for_plan,
+    required_capability_payloads,
+)
 from .constants import (
     DEFAULT_CALL_STEP_ID,
     DEFAULT_ERROR_OUTCOME,
@@ -68,8 +68,8 @@ class WorkflowDraftApi:
         plan = compile_workflow_draft(draft)
         return {
             "compiled_plan": plan,
-            "required_capabilities": _required_capability_payloads(
+            "required_capabilities": required_capability_payloads(
-                _required_capabilities_for_plan(
+                required_capabilities_for_plan(
                     plan,
                     source_bindings=None,
                     context=self.context,
@@ -305,65 +305,6 @@ class WorkflowDraftApi:
         )
 
 
-def _required_capabilities_for_plan(
-    plan: dict[str, Any],
-    *,
-    source_bindings: dict[str, str] | None,
-    context: WorkflowOperationContext,
-) -> 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(context),
-    )
-    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]]:
-    return {
-        name: capability.model_dump(mode="json")
-        for name, capability in sorted(requirements.items())
-    }
-
-
-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 _draft_input_maps(
     *,
     input: Sequence[InputBinding] | None,

src/wf_api/listing.py

@@ -0,0 +1,34 @@
+from __future__ import annotations
+
+from collections.abc import Sequence
+from typing import Any, TypeVar
+
+from wf_platform import page_items
+
+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 shared workflow API list response shape."""
+    page = page_items(items, cursor=cursor, limit=limit)
+    return {
+        key: list(page.items),
+        "next_cursor": page.next_cursor,
+        "total": page.total,
+    }

src/wf_api/runs.py

@@ -11,8 +11,8 @@ from wf_artifacts import (
 )
 from wf_core import RunState
 
+from .artifact_plans import raw_plan_from_artifact
 from .deployments import WorkflowDeploymentApi, _available_sources
-from .models import RawWorkflowPlan
 from .next_actions import NextActions
 from .run_lifecycle import (
     create_pinned_environment,
@@ -68,7 +68,7 @@ class WorkflowRunApi:
                 diagnostics=diagnostics,
             )
 
-        plan = _raw_plan_from_artifact(artifact)
+        plan = raw_plan_from_artifact(artifact)
         run = await self.context.runtime.run_workflow_from_plan(
             plan,
             workflow_input,
@@ -148,7 +148,7 @@ class WorkflowRunApi:
                 diagnostics=diagnostics,
                 trace_count=len(stopped_run.trace),
             )
-        plan = _raw_plan_from_artifact(environment.root_artifact)
+        plan = raw_plan_from_artifact(environment.root_artifact)
         tree = saved_subgraph_tree_from_snapshots(environment.child_artifacts)
         run = await self.context.runtime.resume_workflow_from_plan(
             plan,
@@ -237,33 +237,6 @@ class WorkflowRunApi:
         )
 
 
-def _raw_plan_from_artifact(artifact: WorkflowArtifact) -> RawWorkflowPlan:
-    """Validate the stored plan shape expected by the broker workflow runner."""
-    return RawWorkflowPlan.model_validate(
-        {
-            "name": _plan_field(artifact, "name"),
-            "input_schema": _plan_field(artifact, "input_schema"),
-            "state_schema": _plan_field(artifact, "state_schema"),
-            "output_schema": _plan_field(artifact, "output_schema"),
-            "outcomes": artifact.plan.get("outcomes", ["ok"]),
-            "output": artifact.plan.get("output", []),
-            "start": _plan_field(artifact, "start"),
-            "nodes": _plan_field(artifact, "nodes"),
-            "edges": _plan_field(artifact, "edges"),
-        }
-    )
-
-
-def _plan_field(artifact: WorkflowArtifact, field_name: str) -> Any:
-    try:
-        return artifact.plan[field_name]
-    except KeyError as exc:
-        raise ValueError(
-            f"workflow artifact {artifact.id}@{artifact.version} "
-            f"is missing plan field {field_name!r}"
-        ) from exc
-
-
 def _run_payload(
     *,
     deployment: WorkflowDeployment,

src/wf_mcp/workflow_surface/handlers.py

@@ -20,7 +20,7 @@ from wf_api.models import RawWorkflowPlan
 from wf_api.runs import WorkflowRunApi
 
 from ..broker.service.workflow_operation_context import context_from_service
-from ..shared import paged_list_payload
+from wf_api.listing import paged_list_payload
 from .models import TraceRange
 
 if TYPE_CHECKING:

tests/wf_api/test_artifact_helpers.py

@@ -0,0 +1,88 @@
+from __future__ import annotations
+
+import pytest
+
+from wf_api.artifact_plans import plan_field, plan_nodes, raw_plan_from_artifact
+from wf_api.artifact_refs import artifact_capability_id
+from wf_api.capability_requirements import (
+    observed_node_specs,
+    required_capability_payloads,
+)
+
+from tests.wf_mcp.workflow_surface.conftest import echo_artifact
+
+
+def test_artifact_capability_id_uses_workflow_ref_shape() -> None:
+    artifact = echo_artifact()
+    assert artifact_capability_id(artifact) == (
+        f"workflow.{artifact.id}.v{artifact.version}"
+    )
+
+
+def test_raw_plan_from_artifact_preserves_required_plan_fields() -> None:
+    artifact = echo_artifact()
+    plan = raw_plan_from_artifact(artifact)
+
+    assert plan.name == artifact.plan["name"]
+    assert plan.start == artifact.plan["start"]
+    assert len(plan.nodes) == len(artifact.plan["nodes"])
+
+
+def test_plan_field_reports_missing_field() -> None:
+    artifact = echo_artifact()
+    broken = artifact.model_copy(
+        update={
+            "plan": {
+                key: value for key, value in artifact.plan.items() if key != "start"
+            }
+        }
+    )
+
+    with pytest.raises(ValueError, match="missing plan field 'start'"):
+        plan_field(broken, "start")
+
+
+def test_plan_nodes_returns_only_dict_nodes() -> None:
+    artifact = echo_artifact()
+    modified = artifact.model_copy(
+        update={"plan": {**artifact.plan, "nodes": [{"id": "a"}, "bad"]}}
+    )
+
+    assert plan_nodes(modified) == [{"id": "a"}]
+
+
+def test_required_capability_payloads_sorts_by_name() -> None:
+    artifact = echo_artifact()
+    required_capabilities = artifact.required_capability_map()
+
+    payload = required_capability_payloads(required_capabilities)
+
+    assert list(payload) == sorted(required_capabilities)
+    first = next(iter(payload.values()))
+    assert "ref" in first
+    assert "kind" in first
+
+
+def test_observed_node_specs_projects_enabled_context_specs() -> None:
+    from wf_artifacts import FileWorkflowArtifactStore
+    from wf_mcp.broker import WfMcpService
+    from wf_mcp.models import ConnectionConfig
+    from wf_mcp.storage import FileStore
+    from wf_mcp.broker.service.workflow_operation_context import context_from_service
+    from tests.wf_mcp.test_support import echo_tool, local_temp_root
+
+    artifact_store = FileWorkflowArtifactStore(local_temp_root() / "cap_req_helpers")
+    service = WfMcpService(
+        store=FileStore(artifact_store.root / "mcp"),
+        artifact_store=artifact_store,
+    )
+    service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+    service.register_specs("demo.personal", echo_tool)
+    context = context_from_service(service)
+
+    observed = observed_node_specs(context)
+
+    assert isinstance(observed, dict)
+    assert all(hasattr(detail, "name") for detail in observed.values())

tests/wf_api/test_listing.py

@@ -0,0 +1,26 @@
+from __future__ import annotations
+
+from wf_api.listing import matches_query, paged_list_payload
+
+
+def test_matches_query_accepts_empty_or_missing_query() -> None:
+    assert matches_query("Alpha", query=None) is True
+    assert matches_query("Alpha", query="  ") is True
+
+
+def test_matches_query_searches_non_none_values_case_insensitively() -> None:
+    assert matches_query(None, "Demo Echo", query="echo") is True
+    assert matches_query(None, "Demo Echo", query="missing") is False
+
+
+def test_paged_list_payload_preserves_common_shape() -> None:
+    payload = paged_list_payload(
+        "nodes",
+        [{"name": "a"}, {"name": "b"}, {"name": "c"}],
+        cursor=None,
+        limit=2,
+    )
+
+    assert payload["nodes"] == [{"name": "a"}, {"name": "b"}]
+    assert payload["total"] == 3
+    assert payload["next_cursor"] is not None