test: migrate service async tests

tests/wf_mcp/service/test_adapters.py

@@ -1,7 +1,5 @@
 from __future__ import annotations
 
-import asyncio
-
 from wf_mcp.broker import WfMcpService
 from wf_mcp.models import ConnectionConfig
 from wf_mcp.storage import FileStore
@@ -13,21 +11,21 @@ from ..test_support import (
 from .conftest import ContentOnlyOutputAdapter
 
 
-def test_service_catalog_preserves_json_schema_description_metadata() -> None:
+async def test_service_catalog_preserves_json_schema_description_metadata() -> None:
     service = WfMcpService(store=FileStore(local_temp_root() / "schema_doc_store"))
     service.register_connection(
         ConnectionConfig(id="demo.personal", server="demo", account="personal")
     )
     service.register_adapter("demo", FakeAdapter())
 
-    asyncio.run(service.refresh_connection_catalog("demo.personal"))
+    await service.refresh_connection_catalog("demo.personal")
 
     payload = service.get_catalog().as_payload()
     node = payload["nodes"][0]
     assert node["input_schema"]["properties"]["text"]["description"] == "Text to echo"
 
 
-def test_service_preserves_content_only_tool_output_schema_for_workflows() -> None:
+async def test_service_preserves_content_only_tool_output_schema_for_workflows() -> None:
     service = WfMcpService(store=FileStore(local_temp_root() / "content_only_store"))
     service.register_connection(
         ConnectionConfig(
@@ -36,7 +34,7 @@ def test_service_preserves_content_only_tool_output_schema_for_workflows() -> No
     )
     service.register_adapter("everything", ContentOnlyOutputAdapter())
 
-    asyncio.run(service.refresh_connection_catalog("everything.default"))
+    await service.refresh_connection_catalog("everything.default")
 
     payload = service.get_catalog().as_payload()
     node = payload["nodes"][0]
@@ -44,14 +42,14 @@ def test_service_preserves_content_only_tool_output_schema_for_workflows() -> No
     assert node["output_schema"]["required"] == ["content"]
 
 
-def test_service_wrapped_tool_adapter_model_validates_simple_schema_types() -> None:
+async def test_service_wrapped_tool_adapter_model_validates_simple_schema_types() -> None:
     service = WfMcpService(store=FileStore(local_temp_root() / "adapter_model_store"))
     service.register_connection(
         ConnectionConfig(id="demo.personal", server="demo", account="personal")
     )
     service.register_adapter("demo", FakeAdapter())
 
-    asyncio.run(service.refresh_connection_catalog("demo.personal"))
+    await service.refresh_connection_catalog("demo.personal")
 
     source = service.capability_sources["demo.personal"]
     spec = source.capabilities.node_specs["demo.personal.echo_tool"]

tests/wf_mcp/service/test_catalog.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import asyncio
 import shutil
 
 from wf_authoring import NodeSpec
@@ -268,7 +267,7 @@ def test_service_catalog_split_keeps_system_specs_out_of_backend_catalog() -> No
     assert "wf.std.runtime_error" in available_names
 
 
-def test_service_hydrates_planner_specs_from_stored_catalog() -> None:
+async def test_service_hydrates_planner_specs_from_stored_catalog() -> None:
     store = local_temp_root() / "restart_planner_store"
     shutil.rmtree(store, ignore_errors=True)
     first_service = WfMcpService(store=FileStore(store))
@@ -276,7 +275,7 @@ def test_service_hydrates_planner_specs_from_stored_catalog() -> None:
         ConnectionConfig(id="demo.personal", server="demo", account="personal")
     )
     first_service.register_adapter("demo", FakeAdapter())
-    asyncio.run(first_service.refresh_connection_catalog("demo.personal"))
+    await first_service.refresh_connection_catalog("demo.personal")
 
     second_service = WfMcpService(store=FileStore(store))
     second_service.register_connection(
@@ -288,11 +287,9 @@ def test_service_hydrates_planner_specs_from_stored_catalog() -> None:
         node["qualified_name"]
         for node in second_service.get_planner_catalog().as_payload()["nodes"]
     }
-    run = asyncio.run(
+    run = await second_service.run_workflow_from_plan(
-        second_service.run_workflow_from_plan(
+        single_echo_plan("hydrated_plan", "demo.personal.echo_tool"),
-            single_echo_plan("hydrated_plan", "demo.personal.echo_tool"),
+        {"text": "hello"},
-            {"text": "hello"},
-        )
     )
 
     assert "demo.personal.echo_tool" in planner_names
@@ -412,7 +409,7 @@ def test_source_catalog_service_excludes_hidden_sources_from_planner_catalog() -
     assert "hidden.source.echo_tool" not in planner_names
 
 
-def test_source_catalog_hydrates_connection_source_from_snapshot_directly() -> None:
+async def test_source_catalog_hydrates_connection_source_from_snapshot_directly() -> None:
     root = local_temp_root() / "source_catalog_hydrate_direct"
     shutil.rmtree(root, ignore_errors=True)
     first_service = WfMcpService(store=FileStore(root))
@@ -420,7 +417,7 @@ def test_source_catalog_hydrates_connection_source_from_snapshot_directly() -> N
         ConnectionConfig(id="demo.personal", server="demo", account="personal")
     )
     first_service.register_adapter("demo", FakeAdapter())
-    asyncio.run(first_service.refresh_connection_catalog("demo.personal"))
+    await first_service.refresh_connection_catalog("demo.personal")
 
     second_service = WfMcpService(store=FileStore(root))
     second_service.register_connection(

tests/wf_mcp/service/test_content_access.py

@@ -1,7 +1,5 @@
 from __future__ import annotations
 
-import asyncio
-
 import pytest
 
 from wf_mcp.broker.service.content_access import ContentAccessService
@@ -87,17 +85,17 @@ def _register_local_docs(source_catalog: SourceCatalogService) -> None:
     )
 
 
-def test_content_access_reads_local_documentation_resource() -> None:
+async def test_content_access_reads_local_documentation_resource() -> None:
     content_access, events = _make_content_access()
 
-    result = asyncio.run(content_access.read_resource("test.docs.example"))
+    result = await content_access.read_resource("test.docs.example")
 
     assert result["contents"][0]["uri"] == "wf://docs/example"
     assert result["contents"][0]["text"] == "# Example"
     assert "resource_read_completed" in [e.kind for e in events.list_events()]
 
 
-def test_content_access_reads_upstream_resource_with_events() -> None:
+async def test_content_access_reads_upstream_resource_with_events() -> None:
     service = WfMcpService(
         store=FileStore(local_temp_root() / "content_upstream_resource")
     )
@@ -105,11 +103,9 @@ def test_content_access_reads_upstream_resource_with_events() -> None:
         ConnectionConfig(id="demo.personal", server="demo", account="personal")
     )
     service.register_adapter("demo", FakeAdapter())
-    asyncio.run(service.refresh_connection_catalog("demo.personal"))
+    await service.refresh_connection_catalog("demo.personal")
 
-    result = asyncio.run(
+    result = await service.content_access.read_resource("demo.personal.resource.welcome")
-        service.content_access.read_resource("demo.personal.resource.welcome")
-    )
 
     assert result["contents"][0]["text"] == "Welcome from the fake adapter resource."
     event_kinds = [e.kind for e in service.list_events()]
@@ -117,7 +113,7 @@ def test_content_access_reads_upstream_resource_with_events() -> None:
     assert "resource_read_completed" in event_kinds
 
 
-def test_content_access_renders_upstream_prompt_with_events() -> None:
+async def test_content_access_renders_upstream_prompt_with_events() -> None:
     service = WfMcpService(
         store=FileStore(local_temp_root() / "content_upstream_prompt")
     )
@@ -125,13 +121,11 @@ def test_content_access_renders_upstream_prompt_with_events() -> None:
         ConnectionConfig(id="demo.personal", server="demo", account="personal")
     )
     service.register_adapter("demo", FakeAdapter())
-    asyncio.run(service.refresh_connection_catalog("demo.personal"))
+    await service.refresh_connection_catalog("demo.personal")
 
-    result = asyncio.run(
+    result = await service.content_access.render_prompt(
-        service.content_access.render_prompt(
+        "demo.personal.prompt.summarize",
-            "demo.personal.prompt.summarize",
+        arguments={"text": "hello world"},
-            arguments={"text": "hello world"},
-        )
     )
 
     assert "hello world" in result["messages"][0]["content"]["text"]
@@ -140,12 +134,12 @@ def test_content_access_renders_upstream_prompt_with_events() -> None:
     assert "prompt_get_completed" in event_kinds
 
 
-def test_content_access_renders_local_documentation_prompt() -> None:
+async def test_content_access_renders_local_documentation_prompt() -> None:
     content_access, events = _make_content_access(
         store_root="content_access_local_prompt"
     )
 
-    result = asyncio.run(content_access.render_prompt("test.docs.guide"))
+    result = await content_access.render_prompt("test.docs.guide")
 
     assert result["description"] == "Test documentation prompt."
     assert result["messages"][0]["role"] == "user"
@@ -153,10 +147,10 @@ def test_content_access_renders_local_documentation_prompt() -> None:
     assert "prompt_get_completed" in [e.kind for e in events.list_events()]
 
 
-def test_content_access_raises_on_unknown_resource() -> None:
+async def test_content_access_raises_on_unknown_resource() -> None:
     content_access, _ = _make_content_access(
         store_root="content_access_missing_resource"
     )
 
     with pytest.raises(KeyError):
-        asyncio.run(content_access.read_resource("nonexistent.resource"))
+        await content_access.read_resource("nonexistent.resource")

tests/wf_mcp/service/test_events.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import asyncio
 from typing import Any, cast
 
 from wf_authoring import build_async_registry
@@ -19,7 +18,7 @@ from ..test_support import (
 )
 
 
-def test_service_records_tool_call_events() -> None:
+async def test_service_records_tool_call_events() -> None:
     from wf_core import END, RunStatus
 
     from ..test_support import input_binding, output_binding
@@ -31,7 +30,7 @@ def test_service_records_tool_call_events() -> None:
     )
     service.register_adapter("demo", FakeAdapter())
 
-    asyncio.run(service.refresh_connection_catalog("demo.personal"))
+    await service.refresh_connection_catalog("demo.personal")
 
     plan = raw_plan(
         name="tool_event_plan",
@@ -59,7 +58,7 @@ def test_service_records_tool_call_events() -> None:
         edges=[{"from": "echo", "outcome": "ok", "to": END}],
     )
 
-    run = asyncio.run(service.run_workflow_from_plan(plan, {"text": "hello"}))
+    run = await service.run_workflow_from_plan(plan, {"text": "hello"})
 
     assert run.status == RunStatus.COMPLETED
     tool_events = [
@@ -73,7 +72,7 @@ def test_service_records_tool_call_events() -> None:
     assert tool_events[1].payload["outcome"] == "ok"
 
 
-def test_service_rejects_text_binding_for_raw_mcp_content_contract() -> None:
+async def test_service_rejects_text_binding_for_raw_mcp_content_contract() -> None:
     from wf_core import END
 
     from ..test_support import input_binding, output_binding
@@ -84,7 +83,7 @@ def test_service_rejects_text_binding_for_raw_mcp_content_contract() -> None:
         ConnectionConfig(id="demo.personal", server="demo", account="personal")
     )
     service.register_adapter("demo", ContentOnlyOutputAdapter())
-    asyncio.run(service.refresh_connection_catalog("demo.personal"))
+    await service.refresh_connection_catalog("demo.personal")
     plan = raw_plan(
         name="raw_content_contract",
         input_schema={
@@ -127,14 +126,14 @@ def test_service_rejects_text_binding_for_raw_mcp_content_contract() -> None:
     )
 
 
-def test_service_can_inspect_resources_and_prompts() -> None:
+async def test_service_can_inspect_resources_and_prompts() -> None:
     service = WfMcpService(store=FileStore(local_temp_root() / "inspect_store"))
     service.register_connection(
         ConnectionConfig(id="demo.personal", server="demo", account="personal")
     )
     service.register_adapter("demo", FakeAdapter())
 
-    asyncio.run(service.refresh_connection_catalog("demo.personal"))
+    await service.refresh_connection_catalog("demo.personal")
 
     resources = service.list_resources(connection_id="demo.personal")
     prompts = service.list_prompts(connection_id="demo.personal")
@@ -153,7 +152,7 @@ def test_service_can_inspect_resources_and_prompts() -> None:
     assert prompt.arguments[0]["name"] == "text"
 
 
-def test_service_reports_connection_statuses() -> None:
+async def test_service_reports_connection_statuses() -> None:
     import shutil
 
     store = local_temp_root() / "status_store"
@@ -180,7 +179,7 @@ def test_service_reports_connection_statuses() -> None:
         }
     ]
 
-    asyncio.run(service.refresh_connection_catalog("demo.personal"))
+    await service.refresh_connection_catalog("demo.personal")
     after = service.connection_statuses()
     assert after[0]["has_snapshot"] is True
     assert after[0]["node_count"] == 1
@@ -188,7 +187,7 @@ def test_service_reports_connection_statuses() -> None:
     assert after[0]["prompt_count"] == 1
 
 
-def test_service_can_proxy_resource_reads_and_prompt_gets() -> None:
+async def test_service_can_proxy_resource_reads_and_prompt_gets() -> None:
     import shutil
 
     store = local_temp_root() / "proxy_store"
@@ -199,16 +198,12 @@ def test_service_can_proxy_resource_reads_and_prompt_gets() -> None:
     )
     service.register_adapter("demo", FakeAdapter())
 
-    asyncio.run(service.refresh_connection_catalog("demo.personal"))
+    await service.refresh_connection_catalog("demo.personal")
 
-    resource_result = asyncio.run(
+    resource_result = await service.read_resource("demo.personal.resource.welcome")
-        service.read_resource("demo.personal.resource.welcome")
+    prompt_result = await service.render_prompt(
-    )
+        "demo.personal.prompt.summarize",
-    prompt_result = asyncio.run(
+        arguments={"text": "hello world"},
-        service.render_prompt(
-            "demo.personal.prompt.summarize",
-            arguments={"text": "hello world"},
-        )
     )
 
     assert (
@@ -227,22 +222,20 @@ def test_service_can_proxy_resource_reads_and_prompt_gets() -> None:
     assert "prompt_get_completed" in event_kinds
 
 
-def test_service_can_invoke_raw_method_and_notification() -> None:
+async def test_service_can_invoke_raw_method_and_notification() -> None:
     service = WfMcpService(store=FileStore(local_temp_root() / "raw_store"))
     service.register_connection(
         ConnectionConfig(id="demo.personal", server="demo", account="personal")
     )
     service.register_adapter("demo", FakeAdapter())
 
-    result = asyncio.run(
+    result = await service.invoke_method(
-        service.invoke_method("demo.personal", "demo.echo", params={"text": "hello"})
+        "demo.personal", "demo.echo", params={"text": "hello"}
     )
-    asyncio.run(
+    await service.send_notification(
-        service.send_notification(
+        "demo.personal",
-            "demo.personal",
+        "notifications/progress",
-            "notifications/progress",
+        params={"progress": 1},
-            params={"progress": 1},
-        )
     )
 
     assert result == {"echoed": "hello"}
@@ -253,7 +246,7 @@ def test_service_can_invoke_raw_method_and_notification() -> None:
     assert "raw_notification_completed" in event_kinds
 
 
-def test_generated_specs_use_injected_tool_executor() -> None:
+async def test_generated_specs_use_injected_tool_executor() -> None:
     class RecordingExecutor:
         def __init__(self) -> None:
             self.payloads: list[dict[str, Any]] = []
@@ -278,21 +271,20 @@ def test_generated_specs_use_injected_tool_executor() -> None:
     )
     service.register_adapter("demo", FakeAdapter())
 
-    asyncio.run(service.refresh_connection_catalog("demo.personal"))
+    await service.refresh_connection_catalog("demo.personal")
     spec = service._get_qualified_spec("demo.personal.echo_tool")
     handler = build_async_registry(spec)[spec.name]
 
-    async def run_node() -> dict[str, Any]:
+    result = await handler(
-        return await handler({"text": "hello"}, RuntimeContext(current_node_id="echo"))
+        {"text": "hello"}, RuntimeContext(current_node_id="echo")
-
+    )
-    result = asyncio.run(run_node())
 
     assert result["outcome"] == "ok"
     assert result["output"]["echoed"] == "hello"
     assert executor.payloads == [{"text": "hello"}]
 
 
-def test_service_records_catalog_refresh_failures() -> None:
+async def test_service_records_catalog_refresh_failures() -> None:
     service = WfMcpService(store=FileStore(local_temp_root() / "refresh_fail_store"))
     service.register_connection(
         ConnectionConfig(id="demo.personal", server="demo", account="personal")
@@ -300,7 +292,7 @@ def test_service_records_catalog_refresh_failures() -> None:
     service.register_adapter("demo", FailingDiscoveryAdapter())
 
     try:
-        asyncio.run(service.refresh_connection_catalog("demo.personal"))
+        await service.refresh_connection_catalog("demo.personal")
     except PermissionError as exc:
         assert str(exc) == "Access is denied"
     else:

tests/wf_mcp/service/test_sources.py

@@ -1,7 +1,5 @@
 from __future__ import annotations
 
-import asyncio
-
 from wf_authoring import NodeSpec
 from wf_mcp.broker import WfMcpService
 from wf_mcp.models import ConnectionConfig
@@ -21,7 +19,7 @@ from ..test_support import (
 from .conftest import raw_plan, single_echo_plan
 
 
-def test_service_compiles_and_runs_raw_plan() -> None:
+async def test_service_compiles_and_runs_raw_plan() -> None:
     service = WfMcpService(store=FileStore(local_temp_root() / "run_store"))
     service.register_connection(
         ConnectionConfig(id="demo.personal", server="demo", account="personal")
@@ -64,13 +62,13 @@ def test_service_compiles_and_runs_raw_plan() -> None:
         edges=[{"from": "echo", "outcome": "ok", "to": "__end__"}],
     )
 
-    run = asyncio.run(service.run_workflow_from_plan(plan, {"text": "hello"}))
+    run = await service.run_workflow_from_plan(plan, {"text": "hello"})
 
     assert run.status == "completed"
     assert run.output["echoed"] == "hello"
 
 
-def test_service_preserves_raw_plan_root_output_bindings() -> None:
+async def test_service_preserves_raw_plan_root_output_bindings() -> None:
     service = WfMcpService(store=FileStore(local_temp_root() / "root_output_store"))
     service.register_connection(
         ConnectionConfig(id="demo.personal", server="demo", account="personal")
@@ -113,12 +111,12 @@ def test_service_preserves_raw_plan_root_output_bindings() -> None:
         edges=[{"from": "echo", "outcome": "ok", "to": "__end__"}],
     )
 
-    run = asyncio.run(service.run_workflow_from_plan(plan, {"text": "hello"}))
+    run = await service.run_workflow_from_plan(plan, {"text": "hello"})
 
     assert run.output["echoed"] == "hello"
 
 
-def test_service_resolves_registered_spec_with_dotted_local_name() -> None:
+async def test_service_resolves_registered_spec_with_dotted_local_name() -> None:
     service = WfMcpService(store=FileStore(local_temp_root() / "dotted_store"))
     service.register_connection(
         ConnectionConfig(id="demo.personal", server="demo", account="personal")
@@ -127,13 +125,13 @@ def test_service_resolves_registered_spec_with_dotted_local_name() -> None:
 
     plan = single_echo_plan("dotted_plan", "demo.personal.echo_tool")
 
-    run = asyncio.run(service.run_workflow_from_plan(plan, {"text": "hello"}))
+    run = await service.run_workflow_from_plan(plan, {"text": "hello"})
 
     assert run.status == "completed"
     assert run.output["echoed"] == "hello"
 
 
-def test_service_runs_logical_source_plan_with_dotted_local_name() -> None:
+async def test_service_runs_logical_source_plan_with_dotted_local_name() -> None:
     import shutil
 
     from wf_artifacts import WorkflowDeployment
@@ -163,26 +161,24 @@ def test_service_runs_logical_source_plan_with_dotted_local_name() -> None:
 
     plan = single_echo_plan("logical_plan", "demo.foo.bar")
 
-    run = asyncio.run(
+    run = await service.run_workflow_from_plan(
-        service.run_workflow_from_plan(
+        plan,
-            plan,
+        {"text": "hello"},
-            {"text": "hello"},
+        deployment=WorkflowDeployment(
-            deployment=WorkflowDeployment(
+            id="logical_dotted.personal",
-                id="logical_dotted.personal",
+            artifact_id="logical_dotted",
-                artifact_id="logical_dotted",
+            artifact_version=1,
-                artifact_version=1,
+            bindings=[
-                bindings=[
+                {"logical_source": "demo", "concrete_source": "demo.personal"}
-                    {"logical_source": "demo", "concrete_source": "demo.personal"}
+            ],
-                ],
+        ),
-            ),
-        )
     )
 
     assert run.status == "completed"
     assert run.output["echoed"] == "hello"
 
 
-def test_service_binds_longest_logical_source_prefix_first() -> None:
+async def test_service_binds_longest_logical_source_prefix_first() -> None:
     service = WfMcpService(store=FileStore(local_temp_root() / "prefix_store"))
     service.register_connection(
         ConnectionConfig(id="demo.personal", server="demo", account="personal")
@@ -194,7 +190,7 @@ def test_service_binds_longest_logical_source_prefix_first() -> None:
 
     plan = single_echo_plan("prefix_plan", "demo.personal.echo_tool")
 
-    run = asyncio.run(service.run_workflow_from_plan(plan, {"text": "hello"}))
+    run = await service.run_workflow_from_plan(plan, {"text": "hello"})
 
     assert run.status == "completed"
 
@@ -302,14 +298,14 @@ def test_service_preserves_planner_hidden_connection_source_on_reregistration()
     assert "demo.personal.echo_tool" not in source.capabilities.node_specs
 
 
-def test_service_refreshes_catalog_from_adapter() -> None:
+async def test_service_refreshes_catalog_from_adapter() -> None:
     service = WfMcpService(store=FileStore(local_temp_root() / "refresh_store"))
     service.register_connection(
         ConnectionConfig(id="demo.personal", server="demo", account="personal")
     )
     service.register_adapter("demo", FakeAdapter())
 
-    asyncio.run(service.refresh_connection_catalog("demo.personal"))
+    await service.refresh_connection_catalog("demo.personal")
 
     source = service.capability_sources["demo.personal"]
     assert "demo.personal.echo_tool" in source.capabilities.node_specs

tests/wf_mcp/service/test_upstream_transport.py

@@ -1,7 +1,5 @@
 from __future__ import annotations
 
-import asyncio
-
 from wf_artifacts import WorkflowDeployment
 from wf_platform import CapabilityBuckets, CapabilitySource, SourcePermissions
 
@@ -61,7 +59,7 @@ def test_wfmcpservice_uses_upstream_transport_for_adapters_and_auth() -> None:
     assert service.list_events()[-1].kind == "auth_saved"
 
 
-def test_upstream_transport_invokes_raw_method_and_records_events() -> None:
+async def test_upstream_transport_invokes_raw_method_and_records_events() -> None:
     events: list[McpEvent] = []
     connections = ConnectionRegistry()
     connections.register(
@@ -73,12 +71,10 @@ def test_upstream_transport_invokes_raw_method_and_records_events() -> None:
     )
     transport.register_adapter("demo", FakeAdapter())
 
-    result = asyncio.run(
+    result = await transport.invoke_method(
-        transport.invoke_method(
+        connections.get("demo.personal"),
-            connections.get("demo.personal"),
+        "demo.echo",
-            "demo.echo",
+        params={"text": "hello"},
-            params={"text": "hello"},
-        )
     )
 
     assert result["echoed"] == "hello"
@@ -88,7 +84,7 @@ def test_upstream_transport_invokes_raw_method_and_records_events() -> None:
     ]
 
 
-def test_upstream_transport_refreshes_catalog_directly() -> None:
+async def test_upstream_transport_refreshes_catalog_directly() -> None:
     events: list[McpEvent] = []
     store = FileStore(local_temp_root() / "upstream_refresh")
     connections = ConnectionRegistry()
@@ -107,12 +103,10 @@ def test_upstream_transport_refreshes_catalog_directly() -> None:
     )
     source_catalog.hydrate_connection_source_from_snapshot(connection)
 
-    asyncio.run(
+    await transport.refresh_connection_catalog(
-        transport.refresh_connection_catalog(
+        connection,
-            connection,
+        source_catalog=source_catalog,
-            source_catalog=source_catalog,
+        record_catalog_change_events=lambda source_id, snapshot, reason: None,
-            record_catalog_change_events=lambda source_id, snapshot, reason: None,
-        )
     )
 
     snapshot = store.load_catalog("demo.personal")
@@ -122,7 +116,7 @@ def test_upstream_transport_refreshes_catalog_directly() -> None:
     assert "catalog_refresh_completed" in [event.kind for event in events]
 
 
-def test_upstream_transport_live_diagnostics_report_missing_connection() -> None:
+async def test_upstream_transport_live_diagnostics_report_missing_connection() -> None:
     transport = UpstreamTransportService(
         store=FileStore(local_temp_root() / "upstream_live_missing"),
         event_sink=lambda event: None,
@@ -156,12 +150,10 @@ def test_upstream_transport_live_diagnostics_report_missing_connection() -> None
         bindings=[{"logical_source": "demo", "concrete_source": "demo.personal"}],
     )
 
-    diagnostics = asyncio.run(
+    diagnostics = await transport.deployment_diagnostics(
-        transport.deployment_diagnostics(
+        deployment=deployment,
-            deployment=deployment,
+        artifacts=[artifact],
-            artifacts=[artifact],
+        source_catalog=source_catalog,
-            source_catalog=source_catalog,
-        )
     )
 
     assert diagnostics[0].code == "source_unreachable"

tests/wf_mcp/service/test_workflow_runtime.py

@@ -1,7 +1,5 @@
 from __future__ import annotations
 
-import asyncio
-
 from wf_core import END, NodeUse, RunStatus
 from wf_mcp.broker import WfMcpService
 from wf_mcp.broker.service.source_catalog import SourceCatalogService
@@ -112,7 +110,7 @@ def test_workflow_runtime_service_prepares_node_registry_and_reducers() -> None:
     assert prepared_subgraphs == {}
 
 
-def test_workflow_runtime_service_runs_plan_and_emits_events() -> None:
+async def test_workflow_runtime_service_runs_plan_and_emits_events() -> None:
     events = []
     runtime = WorkflowRuntimeService(
         source_catalog=_source_catalog(),
@@ -120,11 +118,9 @@ def test_workflow_runtime_service_runs_plan_and_emits_events() -> None:
         emit_event=events.append,
     )
 
-    run = asyncio.run(
+    run = await runtime.run_workflow_from_plan(
-        runtime.run_workflow_from_plan(
+        single_echo_plan("runtime_run", "demo.personal.echo_tool"),
-            single_echo_plan("runtime_run", "demo.personal.echo_tool"),
+        {"text": "hello"},
-            {"text": "hello"},
-        )
     )
 
     assert run.output["echoed"] == "hello"
@@ -135,37 +131,35 @@ def test_workflow_runtime_service_runs_plan_and_emits_events() -> None:
     assert events[1].payload["status"] == "completed"
 
 
-def test_workflow_runtime_service_emits_failed_event_for_failed_run() -> None:
+async def test_workflow_runtime_service_emits_failed_event_for_failed_run() -> None:
     service = WfMcpService(store=FileStore(local_temp_root() / "runtime_failed_event"))
 
-    run = asyncio.run(
+    run = await service.workflow_runtime.run_workflow_from_plan(
-        service.workflow_runtime.run_workflow_from_plan(
+        raw_plan(
-            raw_plan(
+            name="runtime_failed_event",
-                name="runtime_failed_event",
+            input_schema={"type": "object", "properties": {}},
-                input_schema={"type": "object", "properties": {}},
+            state_schema={"type": "object", "properties": {}},
-                state_schema={"type": "object", "properties": {}},
+            output_schema={"type": "object", "properties": {}},
-                output_schema={"type": "object", "properties": {}},
+            start="fail",
-                start="fail",
+            nodes=[
-                nodes=[
+                {
-                    {
+                    "id": "fail",
-                        "id": "fail",
+                    "type": "node",
-                        "type": "node",
+                    "node": "wf.std.runtime_error",
-                        "node": "wf.std.runtime_error",
+                    "input": [
-                        "input": [
+                        {
-                            {
+                            "value": "boom",
-                                "value": "boom",
+                            "target": {
-                                "target": {
+                                "root": "local",
-                                    "root": "local",
+                                "parts": ["message"],
-                                    "parts": ["message"],
+                            },
-                                },
+                        }
-                            }
+                    ],
-                        ],
+                }
-                    }
+            ],
-                ],
+            edges=[{"from": "fail", "outcome": "ok", "to": END}],
-                edges=[{"from": "fail", "outcome": "ok", "to": END}],
+        ),
-            ),
+        {},
-            {},
-        )
     )
 
     assert run.status == RunStatus.FAILED

tests/wf_server/test_local_static_server.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import ast
-import asyncio
 from pathlib import Path
 
 from wf_api.models import RawWorkflowPlan
@@ -74,36 +73,30 @@ def test_wf_server_context_imports_no_wfmcp_service() -> None:
     assert violations == []
 
 
-def test_local_static_server_runs_deployment_and_persists_run(tmp_path) -> None:
+async def test_local_static_server_runs_deployment_and_persists_run(tmp_path) -> None:
     server = build_local_static_workflow_server(tmp_path / "store")
     api = server.api
     plan = _constant_plan()
 
-    artifact_result = asyncio.run(
+    artifact_result = await api.create_artifact_from_plan(
-        api.create_artifact_from_plan(
+        artifact_id="server_constant",
-            artifact_id="server_constant",
+        version=1,
-            version=1,
+        title="Server Constant",
-            title="Server Constant",
+        plan=plan,
-            plan=plan,
+        outcomes=["ok"],
-            outcomes=["ok"],
+        source_bindings={"wf.std": "wf.std"},
-            source_bindings={"wf.std": "wf.std"},
-        )
     )
-    deployment_result = asyncio.run(
+    deployment_result = await api.save_deployment(
-        api.save_deployment(
+        {
-            {
+            "id": "server_constant.default",
-                "id": "server_constant.default",
+            "artifact_id": "server_constant",
-                "artifact_id": "server_constant",
+            "artifact_version": 1,
-                "artifact_version": 1,
+            "bindings": [{"logical_source": "wf.std", "concrete_source": "wf.std"}],
-                "bindings": [{"logical_source": "wf.std", "concrete_source": "wf.std"}],
+        }
-            }
-        )
     )
-    run_result = asyncio.run(
+    run_result = await api.run_deployment(
-        api.run_deployment(
+        deployment_id="server_constant.default",
-            deployment_id="server_constant.default",
+        workflow_input={},
-            workflow_input={},
-        )
     )
 
     assert artifact_result["artifact_id"] == "server_constant"
@@ -116,40 +109,34 @@ def test_local_static_server_runs_deployment_and_persists_run(tmp_path) -> None:
     )
 
 
-def test_local_static_server_inspects_and_reads_bounded_trace(tmp_path) -> None:
+async def test_local_static_server_inspects_and_reads_bounded_trace(tmp_path) -> None:
     server = build_local_static_workflow_server(tmp_path / "store")
     api = server.api
     plan = _constant_plan()
-    asyncio.run(
+    await api.create_artifact_from_plan(
-        api.create_artifact_from_plan(
+        artifact_id="server_trace",
-            artifact_id="server_trace",
+        version=1,
-            version=1,
+        title="Server Trace",
-            title="Server Trace",
+        plan=plan.model_copy(update={"name": "server_trace"}),
-            plan=plan.model_copy(update={"name": "server_trace"}),
+        outcomes=["ok"],
-            outcomes=["ok"],
+        source_bindings={"wf.std": "wf.std"},
-            source_bindings={"wf.std": "wf.std"},
-        )
     )
-    asyncio.run(
+    await api.save_deployment(
-        api.save_deployment(
+        {
-            {
+            "id": "server_trace.default",
-                "id": "server_trace.default",
+            "artifact_id": "server_trace",
-                "artifact_id": "server_trace",
+            "artifact_version": 1,
-                "artifact_version": 1,
+            "bindings": [{"logical_source": "wf.std", "concrete_source": "wf.std"}],
-                "bindings": [{"logical_source": "wf.std", "concrete_source": "wf.std"}],
+        }
-            }
-        )
     )
-    run_result = asyncio.run(
+    run_result = await api.run_deployment(
-        api.run_deployment(deployment_id="server_trace.default", workflow_input={})
+        deployment_id="server_trace.default", workflow_input={}
     )
 
-    summary = asyncio.run(api.inspect_run(run_id=run_result["run_id"]))
+    summary = await api.inspect_run(run_id=run_result["run_id"])
-    trace = asyncio.run(
+    trace = await api.read_run_trace(
-        api.read_run_trace(
+        run_id=run_result["run_id"],
-            run_id=run_result["run_id"],
+        trace_range=server.trace_range(start=0, limit=1),
-            trace_range=server.trace_range(start=0, limit=1),
-        )
     )
 
     assert "trace" not in summary