diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 86dbe6253..e97b81c80 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -204,10 +204,10 @@ jobs:
           path: junit-xml
           retention-days: 14
 
-  # Run tests against Temporal Cloud (skipped on forks)
+  # Run the test suite against Temporal Cloud (skipped on forks)
   cloud-test:
     if: ${{ github.event.pull_request.head.repo.full_name == '' || github.event.pull_request.head.repo.full_name == 'temporalio/sdk-python' }}
-    timeout-minutes: 15
+    timeout-minutes: 30
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
@@ -230,16 +230,19 @@ jobs:
       - run: uv tool install poethepoet
       - run: uv sync --all-extras
       - run: poe build-develop
-      - run: poe test -s tests/test_cloud.py --junit-xml=junit-xml/cloud.xml
-        timeout-minutes: 10
+      - run: mkdir junit-xml
+      - run: poe test -s --workflow-environment envconfig --junit-xml=junit-xml/cloud.xml
+        timeout-minutes: 15
         env:
+          TEMPORAL_ADDRESS: sdk-ci.a2dd6.tmprl.cloud:7233
+          TEMPORAL_NAMESPACE: sdk-ci.a2dd6
+          TEMPORAL_API_KEY: ${{ secrets.TEMPORAL_CLIENT_CLOUD_API_KEY }}
+          TEMPORAL_TLS_CLIENT_CERT_DATA: ${{ secrets.TEMPORAL_CLIENT_CERT }}
+          TEMPORAL_TLS_CLIENT_KEY_DATA: ${{ secrets.TEMPORAL_CLIENT_KEY }}
           TEMPORAL_IS_CLOUD_TESTS: true
           TEMPORAL_CLIENT_CLOUD_API_KEY: ${{ secrets.TEMPORAL_CLIENT_CLOUD_API_KEY }}
           TEMPORAL_CLIENT_CLOUD_API_VERSION: 2024-05-13-00
           TEMPORAL_CLIENT_CLOUD_NAMESPACE: sdk-ci.a2dd6
-          TEMPORAL_CLIENT_CLOUD_TARGET: sdk-ci.a2dd6.tmprl.cloud:7233
-          TEMPORAL_CLIENT_CERT: ${{ secrets.TEMPORAL_CLIENT_CERT }}
-          TEMPORAL_CLIENT_KEY: ${{ secrets.TEMPORAL_CLIENT_KEY }}
       - name: "Upload junit-xml artifacts"
         uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4
         if: always()
diff --git a/temporalio/testing/_workflow.py b/temporalio/testing/_workflow.py
index 979222dea..3ab07e5aa 100644
--- a/temporalio/testing/_workflow.py
+++ b/temporalio/testing/_workflow.py
@@ -373,6 +373,27 @@ def client(self) -> temporalio.client.Client:
         """Client to this environment."""
         return self._client
 
+    async def connect_client(self, **kwargs: Any) -> temporalio.client.Client:
+        """Create another client connected to this environment.
+
+        Namespace and connection credentials from this environment's client are
+        used by default.
+        Keyword arguments are forwarded to :py:meth:`temporalio.client.Client.connect`
+        and override those defaults.
+        """
+        config = self.client.service_client.config
+        connect_kwargs: dict[str, Any] = {
+            "namespace": self.client.namespace,
+            "api_key": config.api_key,
+            "tls": config.tls,
+            "rpc_metadata": config.rpc_metadata,
+            "runtime": config.runtime or temporalio.runtime.Runtime.default(),
+        }
+        connect_kwargs.update(kwargs)
+        return await temporalio.client.Client.connect(
+            config.target_host, **connect_kwargs
+        )
+
     async def shutdown(self) -> None:
         """Shut down this environment."""
         pass
diff --git a/tests/conftest.py b/tests/conftest.py
index bfe005ede..9c57bc0d1 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -10,6 +10,7 @@
 from opentelemetry.util._once import Once
 
 from temporalio.client import Client
+from temporalio.envconfig import ClientConfigProfile
 from temporalio.testing import WorkflowEnvironment
 from temporalio.worker import SharedStateManager
 from tests.helpers.worker import ExternalPythonWorker, ExternalWorker
@@ -58,10 +59,43 @@ def pytest_addoption(parser):  # type: ignore[reportMissingParameterType]
         "-E",
         "--workflow-environment",
         default="local",
-        help="Which workflow environment to use ('local', 'time-skipping', or ip:port for existing server)",
+        help="Which workflow environment to use ('local', 'time-skipping', 'envconfig', or ip:port for existing server)",
     )
 
 
+def _uses_envconfig_server(env_type: str) -> bool:
+    return env_type == "envconfig"
+
+
+def pytest_configure(config: pytest.Config) -> None:
+    config.addinivalue_line(
+        "markers",
+        "requires_local_server: test requires local-server-only behavior and cannot run against an envconfig server",
+    )
+
+
+def pytest_collection_modifyitems(
+    config: pytest.Config, items: list[pytest.Item]
+) -> None:
+    if not _uses_envconfig_server(config.getoption("--workflow-environment")):
+        return
+    skip_local_only = pytest.mark.skip(
+        reason="requires a local Temporal server, not the configured envconfig server"
+    )
+    for item in items:
+        if item.get_closest_marker("requires_local_server"):
+            item.add_marker(skip_local_only)
+
+
+async def _create_env_from_envconfig() -> WorkflowEnvironment:
+    config = ClientConfigProfile.load().to_client_connect_config()
+    if not config.get("target_host"):
+        raise ValueError(
+            "An envconfig workflow environment requires TEMPORAL_ADDRESS or an envconfig profile with an address"
+        )
+    return WorkflowEnvironment.from_client(await Client.connect(**config))
+
+
 @pytest.fixture(scope="session")
 def event_loop():
     loop = asyncio.get_event_loop_policy().new_event_loop()  # type: ignore[reportDeprecated]
@@ -100,7 +134,9 @@ def env_type(request: pytest.FixtureRequest) -> str:
 
 @pytest_asyncio.fixture(scope="session")  # type: ignore[reportUntypedFunctionDecorator]
 async def env(env_type: str) -> AsyncGenerator[WorkflowEnvironment, None]:
-    if env_type == "local":
+    if _uses_envconfig_server(env_type):
+        env = await _create_env_from_envconfig()
+    elif env_type == "local":
         env = await WorkflowEnvironment.start_local(
             dev_server_extra_args=[
                 "--dynamic-config-value",
diff --git a/tests/contrib/aws/s3driver/test_s3driver_worker.py b/tests/contrib/aws/s3driver/test_s3driver_worker.py
index 61729535f..fcf17fc16 100644
--- a/tests/contrib/aws/s3driver/test_s3driver_worker.py
+++ b/tests/contrib/aws/s3driver/test_s3driver_worker.py
@@ -62,9 +62,7 @@ async def tmprl_client(
 ) -> AsyncIterator[Client]:
     """Temporal client wired with ExternalStorage backed by the moto S3 server."""
     driver = S3StorageDriver(client=new_aioboto3_client(aioboto3_client), bucket=BUCKET)
-    yield await Client.connect(
-        env.client.service_client.config.target_host,
-        namespace=env.client.namespace,
+    yield await env.connect_client(
         data_converter=dataclasses.replace(
             temporalio.converter.default(),
             external_storage=ExternalStorage(
@@ -112,7 +110,8 @@ async def test_s3_driver_workflow_input_key(
     # worker stores activity input with ri=run_id — same bytes, two S3 objects.
     assert len(keys) == 2
     assert all(
-        f"/ns/default/wt/LargeIOWorkflow/wi/{workflow_id}/ri/" in k for k in keys
+        f"/ns/{tmprl_client.namespace}/wt/LargeIOWorkflow/wi/{workflow_id}/ri/" in k
+        for k in keys
     )
     # Client-side store: ri=null because run ID is not yet known.
     assert sum(1 for k in keys if "/ri/null/" in k) == 1
@@ -138,7 +137,10 @@ async def test_s3_driver_workflow_output_key(
     keys = await _list_keys(aioboto3_client)
     # Activity result and workflow result dedup to same key
     assert len(keys) == 1
-    assert f"/ns/default/wt/LargeIOWorkflow/wi/{workflow_id}/ri/" in keys[0]
+    assert (
+        f"/ns/{tmprl_client.namespace}/wt/LargeIOWorkflow/wi/{workflow_id}/ri/"
+        in keys[0]
+    )
     # Run ID is known for both activity completion and workflow completion
     assert "/ri/null/" not in keys[0]
 
@@ -162,7 +164,8 @@ async def test_s3_driver_workflow_activity_input_key(
     assert len(keys) == 2
     # Both keys are under the workflow wi/ri prefix, not the activity.
     assert all(
-        f"/ns/default/wt/LargeIOWorkflow/wi/{workflow_id}/ri/" in k for k in keys
+        f"/ns/{tmprl_client.namespace}/wt/LargeIOWorkflow/wi/{workflow_id}/ri/" in k
+        for k in keys
     )
     # Activity input is keyed under the scheduling workflow, not the activity.
     assert all("/ai/" not in k for k in keys)
@@ -185,7 +188,10 @@ async def test_s3_driver_workflow_activity_output_key(
     keys = await _list_keys(aioboto3_client)
     # Activity result and workflow result are both LARGE so they deduplicate to one object.
     assert len(keys) == 1
-    assert f"/ns/default/wt/LargeIOWorkflow/wi/{workflow_id}/ri/" in keys[0]
+    assert (
+        f"/ns/{tmprl_client.namespace}/wt/LargeIOWorkflow/wi/{workflow_id}/ri/"
+        in keys[0]
+    )
     # ri=run_id for both stores (run ID is known by the time the activity completes).
     assert "/ri/null/" not in keys[0]
 
@@ -214,7 +220,8 @@ async def test_s3_driver_standalone_activity_input_key(
     assert len(keys) == 2
     # Both keyed under the activity, not a workflow.
     assert all(
-        f"/ns/default/at/large_io_activity/ai/{activity_id}/ri/" in k for k in keys
+        f"/ns/{tmprl_client.namespace}/at/large_io_activity/ai/{activity_id}/ri/" in k
+        for k in keys
     )
     assert all("/wt/" not in k for k in keys)
     # Client-side store does not have run ID information
@@ -244,7 +251,10 @@ async def test_s3_driver_standalone_activity_output_key(
     keys = await _list_keys(aioboto3_client)
     # Only the output is large; keyed under the activity.
     assert len(keys) == 1
-    assert f"/ns/default/at/large_output_activity/ai/{activity_id}/ri/" in keys[0]
+    assert (
+        f"/ns/{tmprl_client.namespace}/at/large_output_activity/ai/{activity_id}/ri/"
+        in keys[0]
+    )
     assert "/ri/null/" not in keys[0]
     assert "/wt/" not in keys[0]
 
@@ -337,7 +347,10 @@ async def test_s3_driver_child_workflow_input_key(
     # Child input is the only large payload — stored under the child's wi/ri.
     assert len(keys) == 1
     # Keyed under the child: child input is stored in the child's context.
-    assert f"/ns/default/wt/ChildWorkflow/wi/{child_workflow_id}/ri/" in keys[0]
+    assert (
+        f"/ns/{tmprl_client.namespace}/wt/ChildWorkflow/wi/{child_workflow_id}/ri/"
+        in keys[0]
+    )
 
 
 async def test_s3_driver_identified_casing(
@@ -359,7 +372,8 @@ async def test_s3_driver_identified_casing(
     assert len(keys) == 2
     # Workflow ID is percent-encoded but casing is preserved verbatim.
     assert all(
-        f"/ns/default/wt/LargeIOWorkflow/wi/{workflow_id}/ri/" in k for k in keys
+        f"/ns/{tmprl_client.namespace}/wt/LargeIOWorkflow/wi/{workflow_id}/ri/" in k
+        for k in keys
     ), "Workflow ID should preserve original case in the key"
 
 
@@ -386,7 +400,8 @@ async def test_s3_driver_content_dedup(
     assert len(keys) == 2
     # Both are under the same workflow wi/ri prefix despite crossing activity boundaries.
     assert all(
-        f"/ns/default/wt/DocumentIngestionWorkflow/wi/{workflow_id}/ri/" in k
+        f"/ns/{tmprl_client.namespace}/wt/DocumentIngestionWorkflow/wi/{workflow_id}/ri/"
+        in k
         for k in keys
     )
     # The two keys differ by content hash only.
@@ -469,9 +484,7 @@ async def test_s3_store_failure_surfaces_in_workflow_history(
         aws_secret_access_key="testing",
     ) as client:
         driver = S3StorageDriver(client=new_aioboto3_client(client), bucket=bad_bucket)
-        bad_client = await Client.connect(
-            env.client.service_client.config.target_host,
-            namespace=env.client.namespace,
+        bad_client = await env.connect_client(
             data_converter=dataclasses.replace(
                 temporalio.converter.default(),
                 external_storage=ExternalStorage(
diff --git a/tests/contrib/langsmith/test_integration.py b/tests/contrib/langsmith/test_integration.py
index a89d1ea4a..426b9a3af 100644
--- a/tests/contrib/langsmith/test_integration.py
+++ b/tests/contrib/langsmith/test_integration.py
@@ -572,6 +572,7 @@ async def test_benign_error_not_marked(
 
 
 class TestComprehensiveTracing:
+    @pytest.mark.requires_local_server
     async def test_comprehensive_with_temporal_runs(
         self, client: Client, env: WorkflowEnvironment
     ) -> None:
@@ -765,6 +766,7 @@ async def user_pipeline() -> WorkflowHandle[Any, Any]:
             "  HandleUpdate:my_unvalidated_update",
         ]
 
+    @pytest.mark.requires_local_server
     async def test_comprehensive_without_temporal_runs(
         self, client: Client, env: WorkflowEnvironment
     ) -> None:
@@ -1138,6 +1140,7 @@ async def run(self) -> str:
 class TestNexusInboundTracing:
     """Verifies nexus handlers receive tracing_context for @traceable collection."""
 
+    @pytest.mark.requires_local_server
     async def test_nexus_direct_traceable_without_temporal_runs(
         self,
         client: Client,
diff --git a/tests/contrib/langsmith/test_plugin.py b/tests/contrib/langsmith/test_plugin.py
index 17c21cb7c..0dad5566f 100644
--- a/tests/contrib/langsmith/test_plugin.py
+++ b/tests/contrib/langsmith/test_plugin.py
@@ -53,6 +53,7 @@ def test_construction_stores_all_config(self) -> None:
 class TestPluginIntegration:
     """End-to-end test using LangSmithPlugin as a Temporal client plugin."""
 
+    @pytest.mark.requires_local_server
     async def test_comprehensive_plugin_trace_hierarchy(
         self, client: Client, env: WorkflowEnvironment
     ) -> None:
diff --git a/tests/contrib/langsmith/test_tracing_env_override.py b/tests/contrib/langsmith/test_tracing_env_override.py
index 9d871e1d3..c8aaa2fb0 100644
--- a/tests/contrib/langsmith/test_tracing_env_override.py
+++ b/tests/contrib/langsmith/test_tracing_env_override.py
@@ -162,6 +162,7 @@ async def test_no_runs_when_langchain_tracing_v2_disabled(
             f"{[r.name for r in collector.runs]}"
         )
 
+    @pytest.mark.requires_local_server
     async def test_no_runs_when_tracing_disabled_for_nexus_start(
         self,
         client: Client,
diff --git a/tests/contrib/openai_agents/test_openai.py b/tests/contrib/openai_agents/test_openai.py
index 96cc25133..23c8939a5 100644
--- a/tests/contrib/openai_agents/test_openai.py
+++ b/tests/contrib/openai_agents/test_openai.py
@@ -471,6 +471,7 @@ async def test_tool_failure_workflow(client: Client):
 
 
 @pytest.mark.parametrize("use_local_model", [True, False])
+@pytest.mark.requires_local_server
 async def test_nexus_tool_workflow(
     client: Client, env: WorkflowEnvironment, use_local_model: bool
 ):
diff --git a/tests/contrib/opentelemetry/test_opentelemetry.py b/tests/contrib/opentelemetry/test_opentelemetry.py
index 71e2fa41d..1bab931ac 100644
--- a/tests/contrib/opentelemetry/test_opentelemetry.py
+++ b/tests/contrib/opentelemetry/test_opentelemetry.py
@@ -457,6 +457,7 @@ async def test_opentelemetry_tracing_update_with_start(
     ]
 
 
+@pytest.mark.requires_local_server
 async def test_opentelemetry_tracing_nexus(client: Client, env: WorkflowEnvironment):
     if env.supports_time_skipping:
         pytest.skip(
diff --git a/tests/contrib/opentelemetry/test_opentelemetry_plugin.py b/tests/contrib/opentelemetry/test_opentelemetry_plugin.py
index 3fd50e89b..337270d0c 100644
--- a/tests/contrib/opentelemetry/test_opentelemetry_plugin.py
+++ b/tests/contrib/opentelemetry/test_opentelemetry_plugin.py
@@ -233,6 +233,7 @@ def validate_update_status(self, status: str) -> None:
             raise ValueError("Status cannot be empty")
 
 
+@pytest.mark.requires_local_server
 async def test_opentelemetry_comprehensive_tracing(
     client: Client,
     env: WorkflowEnvironment,
diff --git a/tests/contrib/workflow_streams/test_workflow_streams.py b/tests/contrib/workflow_streams/test_workflow_streams.py
index 12026ff1a..e7cedd038 100644
--- a/tests/contrib/workflow_streams/test_workflow_streams.py
+++ b/tests/contrib/workflow_streams/test_workflow_streams.py
@@ -2699,6 +2699,7 @@ async def test_subscribe_iterates_through_more_ready(client: Client) -> None:
 
 
 @pytest.mark.asyncio
+@pytest.mark.requires_local_server
 async def test_cross_namespace_nexus_stream(
     client: Client, env: WorkflowEnvironment
 ) -> None:
@@ -2722,8 +2723,7 @@ async def test_cross_namespace_nexus_stream(
         )
     )
 
-    handler_client = await Client.connect(
-        client.service_client.config.target_host,
+    handler_client = await env.connect_client(
         namespace=handler_ns,
     )
 
diff --git a/tests/nexus/test_dynamic_creation_of_user_handler_classes.py b/tests/nexus/test_dynamic_creation_of_user_handler_classes.py
index f7306a46b..22c1a818b 100644
--- a/tests/nexus/test_dynamic_creation_of_user_handler_classes.py
+++ b/tests/nexus/test_dynamic_creation_of_user_handler_classes.py
@@ -10,6 +10,8 @@
 from temporalio.worker import Worker
 from tests.helpers.nexus import make_nexus_endpoint_name
 
+pytestmark = pytest.mark.requires_local_server
+
 
 @workflow.defn
 class MyWorkflow:
diff --git a/tests/nexus/test_nexus_client_updates.py b/tests/nexus/test_nexus_client_updates.py
index f63d5482c..c99822770 100644
--- a/tests/nexus/test_nexus_client_updates.py
+++ b/tests/nexus/test_nexus_client_updates.py
@@ -3,6 +3,7 @@
 import uuid
 
 import nexusrpc
+import pytest
 from nexusrpc.handler import StartOperationContext, service_handler, sync_operation
 
 import temporalio.nexus
@@ -11,6 +12,8 @@
 from temporalio.testing import WorkflowEnvironment
 from temporalio.worker import Worker
 
+pytestmark = pytest.mark.requires_local_server
+
 
 @nexusrpc.service
 class ClientTestService:
@@ -48,9 +51,7 @@ async def test_nexus_client_updates_when_worker_client_changes(
     """Test that Nexus operations get the updated client when worker.client is changed."""
     # Create a second client (simulating a new client after cert rotation)
     # Must use the same runtime
-    client2 = await Client.connect(
-        env.client.service_client.config.target_host,
-        namespace=env.client.namespace,
+    client2 = await env.connect_client(
         data_converter=env.client.data_converter,
         runtime=env.client.service_client.config.runtime,
     )
diff --git a/tests/nexus/test_nexus_worker_shutdown.py b/tests/nexus/test_nexus_worker_shutdown.py
index bd9063237..45dfcb8ad 100644
--- a/tests/nexus/test_nexus_worker_shutdown.py
+++ b/tests/nexus/test_nexus_worker_shutdown.py
@@ -23,6 +23,8 @@
     make_nexus_endpoint_name,
 )
 
+pytestmark = pytest.mark.requires_local_server
+
 
 @nexusrpc.service
 class ShutdownTestService:
diff --git a/tests/nexus/test_signal_link_propagation_e2e.py b/tests/nexus/test_signal_link_propagation_e2e.py
index e489ad8a7..e7906f7a5 100644
--- a/tests/nexus/test_signal_link_propagation_e2e.py
+++ b/tests/nexus/test_signal_link_propagation_e2e.py
@@ -56,6 +56,8 @@
     workflow_event_link_event_type,
 )
 
+pytestmark = pytest.mark.requires_local_server
+
 EventType = temporalio.api.enums.v1.EventType
 
 
diff --git a/tests/nexus/test_standalone_operations.py b/tests/nexus/test_standalone_operations.py
index 8193ba7ba..e0cda9475 100644
--- a/tests/nexus/test_standalone_operations.py
+++ b/tests/nexus/test_standalone_operations.py
@@ -71,6 +71,9 @@
 # ---------------------------------------------------------------------------
 
 
+pytestmark = pytest.mark.requires_local_server
+
+
 @dataclass
 class EchoInput:
     value: str
diff --git a/tests/nexus/test_temporal_operation.py b/tests/nexus/test_temporal_operation.py
index b6b1a92c7..0d1b7d63e 100644
--- a/tests/nexus/test_temporal_operation.py
+++ b/tests/nexus/test_temporal_operation.py
@@ -20,6 +20,8 @@
 from tests.helpers import EventType, assert_event_subsequence, assert_eventually
 from tests.helpers.nexus import make_nexus_endpoint_name
 
+pytestmark = pytest.mark.requires_local_server
+
 
 @dataclass
 class Input:
diff --git a/tests/nexus/test_use_existing_conflict_policy.py b/tests/nexus/test_use_existing_conflict_policy.py
index e7cee9e1c..94a3821ba 100644
--- a/tests/nexus/test_use_existing_conflict_policy.py
+++ b/tests/nexus/test_use_existing_conflict_policy.py
@@ -4,6 +4,7 @@
 import uuid
 from dataclasses import dataclass
 
+import pytest
 from nexusrpc.handler import service_handler
 
 from temporalio import nexus, workflow
@@ -13,6 +14,8 @@
 from temporalio.worker import Worker
 from tests.helpers.nexus import make_nexus_endpoint_name
 
+pytestmark = pytest.mark.requires_local_server
+
 
 @dataclass
 class OpInput:
diff --git a/tests/nexus/test_workflow_caller.py b/tests/nexus/test_workflow_caller.py
index 89ce2719a..f776e6e18 100644
--- a/tests/nexus/test_workflow_caller.py
+++ b/tests/nexus/test_workflow_caller.py
@@ -90,6 +90,9 @@ class OpDefinitionType(IntEnum):
     LONGHAND = 1
 
 
+pytestmark = pytest.mark.requires_local_server
+
+
 @dataclass
 class SyncResponse:
     op_definition_type: OpDefinitionType
@@ -1979,9 +1982,7 @@ async def test_workflow_caller_custom_metrics(client: Client, env: WorkflowEnvir
     )
 
     # New client with the runtime
-    client = await Client.connect(
-        client.service_client.config.target_host,
-        namespace=client.namespace,
+    client = await env.connect_client(
         runtime=runtime,
     )
 
@@ -2052,9 +2053,7 @@ async def test_workflow_caller_buffered_metrics(
     assert not buffer.retrieve_updates()
 
     # Create a new client on the runtime and execute the custom metric workflow
-    client = await Client.connect(
-        client.service_client.config.target_host,
-        namespace=client.namespace,
+    client = await env.connect_client(
         runtime=runtime,
     )
     task_queue = str(uuid.uuid4())
diff --git a/tests/nexus/test_workflow_caller_cancellation_types.py b/tests/nexus/test_workflow_caller_cancellation_types.py
index bf33983a5..fa39009a5 100644
--- a/tests/nexus/test_workflow_caller_cancellation_types.py
+++ b/tests/nexus/test_workflow_caller_cancellation_types.py
@@ -25,6 +25,8 @@
 from tests.helpers import LogCapturer, assert_event_subsequence, assert_eventually
 from tests.helpers.nexus import make_nexus_endpoint_name
 
+pytestmark = pytest.mark.requires_local_server
+
 
 @dataclass
 class TestContext:
diff --git a/tests/nexus/test_workflow_caller_cancellation_types_when_cancel_handler_fails.py b/tests/nexus/test_workflow_caller_cancellation_types_when_cancel_handler_fails.py
index 4cdeeeb15..5a0970c95 100644
--- a/tests/nexus/test_workflow_caller_cancellation_types_when_cancel_handler_fails.py
+++ b/tests/nexus/test_workflow_caller_cancellation_types_when_cancel_handler_fails.py
@@ -30,6 +30,8 @@
     has_event,
 )
 
+pytestmark = pytest.mark.requires_local_server
+
 
 @dataclass
 class TestContext:
diff --git a/tests/nexus/test_workflow_caller_error_chains.py b/tests/nexus/test_workflow_caller_error_chains.py
index 9ff84f405..18868288a 100644
--- a/tests/nexus/test_workflow_caller_error_chains.py
+++ b/tests/nexus/test_workflow_caller_error_chains.py
@@ -25,6 +25,8 @@
 from temporalio.worker import Worker
 from tests.helpers.nexus import make_nexus_endpoint_name
 
+pytestmark = pytest.mark.requires_local_server
+
 
 @dataclass
 class ExpectedError:
diff --git a/tests/nexus/test_workflow_caller_errors.py b/tests/nexus/test_workflow_caller_errors.py
index 9246b9fd7..eb85155fe 100644
--- a/tests/nexus/test_workflow_caller_errors.py
+++ b/tests/nexus/test_workflow_caller_errors.py
@@ -42,6 +42,8 @@
 from tests.helpers import LogCapturer, assert_eq_eventually
 from tests.helpers.nexus import make_nexus_endpoint_name
 
+pytestmark = pytest.mark.requires_local_server
+
 operation_invocation_counts = Counter[str]()
 
 logger = getLogger(__name__)
diff --git a/tests/nexus/test_workflow_run_operation.py b/tests/nexus/test_workflow_run_operation.py
index 851f408ec..51032a23a 100644
--- a/tests/nexus/test_workflow_run_operation.py
+++ b/tests/nexus/test_workflow_run_operation.py
@@ -23,6 +23,8 @@
 from temporalio.worker import Worker
 from tests.helpers.nexus import make_nexus_endpoint_name
 
+pytestmark = pytest.mark.requires_local_server
+
 
 @dataclass
 class Input:
diff --git a/tests/test_client.py b/tests/test_client.py
index d611eda3a..15324cf78 100644
--- a/tests/test_client.py
+++ b/tests/test_client.py
@@ -556,7 +556,7 @@ async def test_interceptor(client: Client, worker: ExternalWorker):
     assert interceptor.traces[4][1].id == handle.id
 
 
-async def test_lazy_client(client: Client, env: WorkflowEnvironment):
+async def test_lazy_client(env: WorkflowEnvironment):
     # TODO(cretz): Fix
     if env.supports_time_skipping:
         pytest.skip(
@@ -564,23 +564,17 @@ async def test_lazy_client(client: Client, env: WorkflowEnvironment):
         )
     # Create another client that is lazy. This test just makes sure the
     # functionality continues to work.
-    lazy_client = await Client.connect(
-        client.service_client.config.target_host,
-        namespace=client.namespace,
-        lazy=True,
-    )
+    lazy_client = await env.connect_client(lazy=True)
     assert not lazy_client.service_client.worker_service_client._bridge_client
     await lazy_client.workflow_service.get_system_info(GetSystemInfoRequest())
     assert lazy_client.service_client.worker_service_client._bridge_client
 
 
-async def test_client_connected_skips_connect_lock(client: Client):
+async def test_client_connected_skips_connect_lock(env: WorkflowEnvironment):
     # Once connected, RPCs must not touch the lazy-connect lock. Acquiring it
     # per-RPC put an event-loop-bound primitive on the hot path, pinning a
     # connected client to the loop it connected on.
-    other = await Client.connect(
-        client.service_client.config.target_host, namespace=client.namespace
-    )
+    other = await env.connect_client()
     svc = other.service_client.worker_service_client
     assert svc._bridge_client
 
@@ -599,18 +593,13 @@ async def acquire(self) -> Literal[True]:
     assert counting.acquire_count == 0
 
 
-def test_client_reuse_across_event_loops(client: Client):
+def test_client_reuse_across_event_loops(env: WorkflowEnvironment):
     # A connected client must not be pinned to the loop (or thread) it
     # connected on. This mirrors the long-lived-loop reuse pattern used by
     # gevent/gunicorn and synchronous services.
-    target_host = client.service_client.config.target_host
-    namespace = client.namespace
-
     connect_loop = asyncio.new_event_loop()
     try:
-        reused_client = connect_loop.run_until_complete(
-            Client.connect(target_host, namespace=namespace)
-        )
+        reused_client = connect_loop.run_until_complete(env.connect_client())
     finally:
         connect_loop.close()
 
@@ -668,21 +657,24 @@ async def test_list_workflows_and_fetch_history(
             )
             expected_id_and_input.append((workflow_id, f'"user{i}"'))
 
-    # List them and get their history
-    actual_id_and_input = sorted(
-        [
-            (
-                hist.workflow_id,
-                hist.events[0]
-                .workflow_execution_started_event_attributes.input.payloads[0]
-                .data.decode(),
-            )
-            async for hist in client.list_workflows(
-                f"WorkflowId = '{workflow_id}'"
-            ).map_histories()
-        ]
-    )
-    assert actual_id_and_input == expected_id_and_input
+    # Visibility is eventually consistent, so wait for all runs before fetching
+    # their histories.
+    async def list_id_and_input() -> list[tuple[str, str]]:
+        return sorted(
+            [
+                (
+                    hist.workflow_id,
+                    hist.events[0]
+                    .workflow_execution_started_event_attributes.input.payloads[0]
+                    .data.decode(),
+                )
+                async for hist in client.list_workflows(
+                    f"WorkflowId = '{workflow_id}'"
+                ).map_histories()
+            ]
+        )
+
+    await assert_eq_eventually(expected_id_and_input, list_id_and_input)
 
     # Verify listing can limit results
     limited = [
@@ -837,6 +829,7 @@ def test_history_from_json():
     )
 
 
+@pytest.mark.requires_local_server
 async def test_schedule_basics(
     client: Client, worker: ExternalWorker, env: WorkflowEnvironment
 ):
@@ -844,8 +837,6 @@ async def test_schedule_basics(
         pytest.skip("Java test server doesn't support schedules")
     elif os.getenv("TEMPORAL_TEST_PROTO3"):
         pytest.skip("Older proto library cannot compare repeated fields")
-    await assert_no_schedules(client)
-
     # Create a schedule with a lot of stuff
     schedule = Schedule(
         action=ScheduleActionStartWorkflow(
@@ -1083,10 +1074,9 @@ async def list_ids() -> list[str]:
     assert list_descs[0].id in [f"{handle.id}-3", f"{handle.id}-4"]
     assert list_descs[1].id in [f"{handle.id}-3", f"{handle.id}-4"]
 
-    # Delete all of the schedules
-    for id in await list_ids():
+    # Delete the schedules created by this test.
+    for id in expected_ids:
         await client.get_schedule_handle(id).delete()
-    await assert_no_schedules(client)
 
 
 async def test_schedule_calendar_spec_defaults(
@@ -1094,8 +1084,6 @@ async def test_schedule_calendar_spec_defaults(
 ):
     if env.supports_time_skipping:
         pytest.skip("Java test server doesn't support schedules")
-    await assert_no_schedules(client)
-
     handle = await client.create_schedule(
         f"schedule-{uuid.uuid4()}",
         Schedule(
@@ -1124,7 +1112,6 @@ async def test_schedule_calendar_spec_defaults(
             assert time == desc.info.next_action_times[i - 1] + timedelta(days=1)
 
     await handle.delete()
-    await assert_no_schedules(client)
 
 
 async def test_schedule_trigger_immediately(
@@ -1132,8 +1119,6 @@ async def test_schedule_trigger_immediately(
 ):
     if env.supports_time_skipping:
         pytest.skip("Java test server doesn't support schedules")
-    await assert_no_schedules(client)
-
     # Create paused schedule that triggers immediately
     handle = await client.create_schedule(
         f"schedule-{uuid.uuid4()}",
@@ -1165,7 +1150,6 @@ async def test_schedule_trigger_immediately(
     )
 
     await handle.delete()
-    await assert_no_schedules(client)
 
 
 async def test_schedule_backfill(
@@ -1173,8 +1157,6 @@ async def test_schedule_backfill(
 ):
     if env.supports_time_skipping:
         pytest.skip("Java test server doesn't support schedules")
-    await assert_no_schedules(client)
-
     begin = datetime(year=2020, month=1, day=20, hour=5)
 
     # Create paused schedule that runs every minute and has two backfills
@@ -1225,7 +1207,6 @@ async def test_schedule_backfill(
         )
     finally:
         await handle.delete()
-        await assert_no_schedules(client)
 
 
 async def test_schedule_create_limited_actions_validation(
@@ -1251,13 +1232,12 @@ async def test_schedule_create_limited_actions_validation(
     assert "are remaining actions set" in str(err.value)
 
 
+@pytest.mark.requires_local_server
 async def test_schedule_workflow_search_attribute_update(
     client: Client, env: WorkflowEnvironment
 ):
     if env.supports_time_skipping:
         pytest.skip("Java test server doesn't support schedules")
-    await assert_no_schedules(client)
-
     # Put search attribute on server
     text_attr_key = SearchAttributeKey.for_text("python-test-schedule-text")
     untyped_keyword_key = SearchAttributeKey.for_keyword("python-test-schedule-keyword")
@@ -1350,7 +1330,6 @@ def update_schedule_typed_attrs(
     assert desc.typed_search_attributes[text_attr_key] == "some-schedule-attr1"
 
     await handle.delete()
-    await assert_no_schedules(client)
 
 
 @pytest.mark.parametrize(
@@ -1362,13 +1341,12 @@ def update_schedule_typed_attrs(
         "partial-new-values-overwrites-and-drops",
     ],
 )
+@pytest.mark.requires_local_server
 async def test_schedule_search_attribute_update(
     client: Client, env: WorkflowEnvironment, test_case: str
 ):
     if env.supports_time_skipping:
         pytest.skip("Java test server doesn't support schedules")
-    await assert_no_schedules(client)
-
     # Put search attributes on server
     key_1 = SearchAttributeKey.for_text("python-test-schedule-sa-update-key-1")
     key_2 = SearchAttributeKey.for_keyword("python-test-schedule-sa-update-key-2")
@@ -1486,15 +1464,6 @@ async def expectation() -> bool:
         raise ValueError(f"Invalid test case: {test_case}")
 
     await handle.delete()
-    await assert_no_schedules(client)
-
-
-async def assert_no_schedules(client: Client) -> None:
-    # Listing appears eventually consistent
-    async def schedule_count() -> int:
-        return len([d async for d in await client.list_schedules()])
-
-    await assert_eq_eventually(0, schedule_count)
 
 
 async def test_build_id_interactions(client: Client, env: WorkflowEnvironment):
@@ -1556,6 +1525,9 @@ async def run(self) -> str:
             return "My First Result"
 
 
+# Cloud does not reliably provide a prior completion result to manually
+# triggered schedule actions.
+@pytest.mark.requires_local_server
 async def test_schedule_last_completion_result(
     client: Client, env: WorkflowEnvironment
 ):
@@ -1590,12 +1562,14 @@ async def get_schedule_result() -> tuple[int, str | None]:
                 result = await workflow_handle.result()
                 return length, result
 
-        assert await get_schedule_result() == (1, "My First Result")
+        expected_first_result: tuple[int, str | None] = (1, "My First Result")
+        await assert_eq_eventually(expected_first_result, get_schedule_result)
         await handle.trigger()
-        assert await get_schedule_result() == (
+        expected_second_result: tuple[int, str | None] = (
             2,
             "From last completion: My First Result",
         )
+        await assert_eq_eventually(expected_second_result, get_schedule_result)
 
         await handle.delete()
 
diff --git a/tests/test_cloud.py b/tests/test_cloud.py
index b701bdf94..d7fefb4da 100644
--- a/tests/test_cloud.py
+++ b/tests/test_cloud.py
@@ -1,18 +1,11 @@
-"""Tests that run against Temporal Cloud."""
+"""Tests that run against the Temporal Cloud Operations API."""
 
-import multiprocessing
 import os
-from collections.abc import AsyncGenerator, Iterator
 
 import pytest
-import pytest_asyncio
 
 from temporalio.api.cloud.cloudservice.v1 import GetNamespaceRequest
-from temporalio.client import Client, CloudOperationsClient
-from temporalio.service import TLSConfig
-from temporalio.testing import WorkflowEnvironment
-from temporalio.worker import SharedStateManager
-from tests.helpers.worker import ExternalPythonWorker, ExternalWorker
+from temporalio.client import CloudOperationsClient
 
 # Skip entire module unless explicitly enabled
 pytestmark = pytest.mark.skipif(
@@ -21,54 +14,6 @@
 )
 
 
-@pytest_asyncio.fixture(scope="module")  # type: ignore[reportUntypedFunctionDecorator]
-async def env() -> AsyncGenerator[WorkflowEnvironment, None]:
-    tls_config: bool | TLSConfig = True
-    client_cert = os.environ.get("TEMPORAL_CLIENT_CERT")
-    client_key = os.environ.get("TEMPORAL_CLIENT_KEY")
-    if client_cert and client_key:
-        tls_config = TLSConfig(
-            client_cert=client_cert.encode(),
-            client_private_key=client_key.encode(),
-        )
-    client = await Client.connect(
-        os.environ["TEMPORAL_CLIENT_CLOUD_TARGET"],
-        namespace=os.environ["TEMPORAL_CLIENT_CLOUD_NAMESPACE"],
-        api_key=os.environ.get("TEMPORAL_CLIENT_CLOUD_API_KEY"),
-        tls=tls_config,
-    )
-    env = WorkflowEnvironment.from_client(client)
-    yield env
-    await env.shutdown()
-
-
-@pytest_asyncio.fixture  # type: ignore[reportUntypedFunctionDecorator]
-async def client(env: WorkflowEnvironment) -> Client:
-    return env.client
-
-
-@pytest_asyncio.fixture(scope="module")  # type: ignore[reportUntypedFunctionDecorator]
-async def worker(
-    env: WorkflowEnvironment,
-) -> AsyncGenerator[ExternalWorker, None]:
-    w = ExternalPythonWorker(env)
-    yield w
-    await w.close()
-
-
-@pytest.fixture(scope="module")
-def shared_state_manager() -> Iterator[SharedStateManager]:
-    mp_mgr = multiprocessing.Manager()
-    mgr = SharedStateManager.create_from_multiprocessing(mp_mgr)
-    try:
-        yield mgr
-    finally:
-        mp_mgr.shutdown()
-
-
-# --- Cloud-specific tests ---
-
-
 async def test_cloud_client_simple():
     client = await CloudOperationsClient.connect(
         api_key=os.environ["TEMPORAL_CLIENT_CLOUD_API_KEY"],
@@ -78,11 +23,3 @@ async def test_cloud_client_simple():
         GetNamespaceRequest(namespace=os.environ["TEMPORAL_CLIENT_CLOUD_NAMESPACE"])
     )
     assert os.environ["TEMPORAL_CLIENT_CLOUD_NAMESPACE"] == result.namespace.namespace
-
-
-# --- Delegated tests ---
-# Import test functions to re-run them against cloud fixtures.
-
-from tests.worker.test_activity import (  # noqa: E402
-    test_activity_info,  # pyright: ignore[reportUnusedImport]  # noqa: F401
-)
diff --git a/tests/test_envconfig.py b/tests/test_envconfig.py
index c1a7e32ab..0b44db731 100644
--- a/tests/test_envconfig.py
+++ b/tests/test_envconfig.py
@@ -1,12 +1,20 @@
 import os
 import textwrap
 from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
 
 import pytest
 
+import temporalio.runtime
+import temporalio.service
 from temporalio.client import Client
 from temporalio.envconfig import ClientConfig, ClientConfigProfile, ClientConfigTLS
 from temporalio.service import TLSConfig
+from temporalio.testing import WorkflowEnvironment
+from tests import conftest
+
+pytestmark = pytest.mark.requires_local_server
 
 # A base TOML config with a default and a custom profile
 TOML_CONFIG_BASE = textwrap.dedent(
@@ -147,6 +155,82 @@ def test_load_profile_from_data_env_overrides():
     assert config.get("target_host") == "env-address"
 
 
+@pytest.mark.parametrize(
+    ("env_type", "expected"),
+    [
+        ("envconfig", True),
+        ("local", False),
+        ("time-skipping", False),
+    ],
+)
+def test_envconfig_server_selection(
+    env_type: str,
+    expected: bool,
+):
+    assert conftest._uses_envconfig_server(env_type) is expected
+
+
+async def test_envconfig_workflow_environment_uses_client_connect_config(
+    monkeypatch: pytest.MonkeyPatch,
+):
+    monkeypatch.setenv("TEMPORAL_ADDRESS", "env-address")
+    monkeypatch.setenv("TEMPORAL_NAMESPACE", "env-namespace")
+    monkeypatch.setenv("TEMPORAL_API_KEY", "env-api-key")
+    monkeypatch.setenv("TEMPORAL_TLS", "false")
+    monkeypatch.setenv("TEMPORAL_GRPC_META_TEST_HEADER", "env-value")
+    client = object()
+    environment = object()
+    connect = AsyncMock(return_value=client)
+    monkeypatch.setattr(Client, "connect", connect)
+    monkeypatch.setattr(
+        WorkflowEnvironment,
+        "from_client",
+        lambda actual_client: environment,
+    )
+
+    assert await conftest._create_env_from_envconfig() is environment
+    connect.assert_awaited_once_with(
+        target_host="env-address",
+        namespace="env-namespace",
+        api_key="env-api-key",
+        tls=False,
+        rpc_metadata={"test-header": "env-value"},
+    )
+
+
+async def test_workflow_environment_connect_client_inherits_connection_options(
+    monkeypatch: pytest.MonkeyPatch,
+):
+    runtime = temporalio.runtime.Runtime.default()
+    source_client = SimpleNamespace(
+        namespace="env-namespace",
+        service_client=SimpleNamespace(
+            config=temporalio.service.ConnectConfig(
+                target_host="env-address",
+                api_key="env-api-key",
+                tls=False,
+                rpc_metadata={"test-header": "env-value"},
+                runtime=runtime,
+            )
+        ),
+    )
+    environment = WorkflowEnvironment(source_client)  # type: ignore[arg-type]
+    connect = AsyncMock(return_value=object())
+    monkeypatch.setattr(Client, "connect", connect)
+
+    await environment.connect_client(lazy=True)
+
+    connect.assert_awaited_once_with(
+        "env-address",
+        namespace="env-namespace",
+        api_key="env-api-key",
+        tls=False,
+        rpc_metadata={"test-header": "env-value"},
+        runtime=runtime,
+        lazy=True,
+    )
+
+
 def test_load_profile_env_overrides(base_config_file: Path):
     """Test that environment variables correctly override file settings."""
     env = {
diff --git a/tests/test_plugins.py b/tests/test_plugins.py
index e8823af27..9414e8df0 100644
--- a/tests/test_plugins.py
+++ b/tests/test_plugins.py
@@ -60,6 +60,7 @@ async def connect_service_client(
         return await next(config)
 
 
+@pytest.mark.requires_local_server
 async def test_client_plugin(client: Client, env: WorkflowEnvironment):
     if env.supports_time_skipping:
         pytest.skip("Client connect is only designed for local")
diff --git a/tests/test_runtime.py b/tests/test_runtime.py
index 55501883d..6416ce930 100644
--- a/tests/test_runtime.py
+++ b/tests/test_runtime.py
@@ -21,6 +21,7 @@
     TelemetryFilter,
     _RuntimeRef,
 )
+from temporalio.testing import WorkflowEnvironment
 from temporalio.worker import Worker
 from tests.helpers import (
     LogHandler,
@@ -37,22 +38,18 @@ async def run(self, name: str) -> str:
         return f"Hello, {name}!"
 
 
-async def test_different_runtimes(client: Client):
+async def test_different_runtimes(env: WorkflowEnvironment):
     # Create two workers in separate runtimes and run workflows on them.
     # Confirm they each have different Prometheus addresses.
     prom_addr1 = f"127.0.0.1:{find_free_port()}"
-    client1 = await Client.connect(
-        client.service_client.config.target_host,
-        namespace=client.namespace,
+    client1 = await env.connect_client(
         runtime=Runtime(
             telemetry=TelemetryConfig(metrics=PrometheusConfig(bind_address=prom_addr1))
         ),
     )
 
     prom_addr2 = f"127.0.0.1:{find_free_port()}"
-    client2 = await Client.connect(
-        client.service_client.config.target_host,
-        namespace=client.namespace,
+    client2 = await env.connect_client(
         runtime=Runtime(
             telemetry=TelemetryConfig(metrics=PrometheusConfig(bind_address=prom_addr2))
         ),
@@ -151,16 +148,14 @@ async def run(self) -> None:
         raise RuntimeError("Intentional error")
 
 
-async def test_runtime_task_fail_log_forwarding(client: Client):
+async def test_runtime_task_fail_log_forwarding(env: WorkflowEnvironment):
     # Client with lo capturing runtime
     log_queue: queue.Queue[logging.LogRecord] = queue.Queue()
     log_queue_list = cast(list[logging.LogRecord], log_queue.queue)
     handler = logging.handlers.QueueHandler(log_queue)
     logger = logging.getLogger(f"log-{uuid.uuid4()}")
     logger.setLevel(logging.WARN)
-    client = await Client.connect(
-        client.service_client.config.target_host,
-        namespace=client.namespace,
+    client = await env.connect_client(
         runtime=Runtime(
             telemetry=TelemetryConfig(
                 logging=LoggingConfig(
@@ -199,7 +194,7 @@ async def has_log() -> bool:
         assert record.temporal_log.fields["run_id"] == handle.result_run_id  # type: ignore
 
 
-async def test_prometheus_histogram_bucket_overrides(client: Client):
+async def test_prometheus_histogram_bucket_overrides(env: WorkflowEnvironment):
     # Set up a Prometheus configuration with custom histogram bucket overrides
     prom_addr = f"127.0.0.1:{find_free_port()}"
     special_value = float(1234.5678)
@@ -229,9 +224,7 @@ async def test_prometheus_histogram_bucket_overrides(client: Client):
     custom_histogram.record(600)
 
     # Create client with overrides
-    client_with_overrides = await Client.connect(
-        client.service_client.config.target_host,
-        namespace=client.namespace,
+    client_with_overrides = await env.connect_client(
         runtime=runtime,
     )
 
@@ -270,7 +263,7 @@ async def check_metrics() -> None:
     await assert_eventually(check_metrics)
 
 
-async def test_opentelemetry_histogram_bucket_overrides(client: Client):
+async def test_opentelemetry_histogram_bucket_overrides(env: WorkflowEnvironment):
     # Set up an OpenTelemetry configuration with custom histogram bucket overrides
     import threading
     from http.server import BaseHTTPRequestHandler, HTTPServer
@@ -336,9 +329,7 @@ def do_POST(self):
 
         # Run a workflow so built-in histograms (e.g. temporal_long_request_latency)
         # are recorded and exported.
-        client_with_overrides = await Client.connect(
-            client.service_client.config.target_host,
-            namespace=client.namespace,
+        client_with_overrides = await env.connect_client(
             runtime=runtime,
         )
         task_queue = f"task-queue-{uuid.uuid4()}"
diff --git a/tests/test_serialization_context.py b/tests/test_serialization_context.py
index 0fde2aa96..8d65d5f1f 100644
--- a/tests/test_serialization_context.py
+++ b/tests/test_serialization_context.py
@@ -217,19 +217,19 @@ async def test_payload_conversion_calls_follow_expected_sequence_and_contexts(
 
         workflow_context = dataclasses.asdict(
             WorkflowSerializationContext(
-                namespace="default",
+                namespace=client.namespace,
                 workflow_id=workflow_id,
             )
         )
         child_workflow_context = dataclasses.asdict(
             WorkflowSerializationContext(
-                namespace="default",
+                namespace=client.namespace,
                 workflow_id=f"{workflow_id}_child",
             )
         )
         activity_context = dataclasses.asdict(
             ActivitySerializationContext(
-                namespace="default",
+                namespace=client.namespace,
                 workflow_id=workflow_id,
                 workflow_type=PayloadConversionWorkflow.__name__,
                 activity_type=passthrough_activity.__name__,
@@ -363,14 +363,14 @@ async def test_heartbeat_details_payload_conversion(client: Client):
 
         workflow_context = dataclasses.asdict(
             WorkflowSerializationContext(
-                namespace="default",
+                namespace=client.namespace,
                 workflow_id=workflow_id,
             )
         )
 
         activity_context = dataclasses.asdict(
             ActivitySerializationContext(
-                namespace="default",
+                namespace=client.namespace,
                 workflow_id=workflow_id,
                 workflow_type=HeartbeatDetailsSerializationContextTestWorkflow.__name__,
                 activity_type=activity_with_heartbeat_details.__name__,
@@ -455,13 +455,13 @@ async def test_local_activity_payload_conversion(client: Client):
 
         workflow_context = dataclasses.asdict(
             WorkflowSerializationContext(
-                namespace="default",
+                namespace=client.namespace,
                 workflow_id=workflow_id,
             )
         )
         local_activity_context = dataclasses.asdict(
             ActivitySerializationContext(
-                namespace="default",
+                namespace=client.namespace,
                 workflow_id=workflow_id,
                 workflow_type=LocalActivityWorkflow.__name__,
                 activity_type=local_activity.__name__,
@@ -572,11 +572,11 @@ async def test_async_activity_completion_payload_conversion(
         workflow_runner=UnsandboxedWorkflowRunner(),  # so that we can use isinstance
     ):
         workflow_context = WorkflowSerializationContext(
-            namespace="default",
+            namespace=client.namespace,
             workflow_id=workflow_id,
         )
         activity_context = ActivitySerializationContext(
-            namespace="default",
+            namespace=client.namespace,
             workflow_id=workflow_id,
             workflow_type=AsyncActivityCompletionSerializationContextTestWorkflow.__name__,
             activity_type=async_activity.__name__,
@@ -649,7 +649,7 @@ def my_method(self) -> None:
 
 def test_subclassed_async_activity_handle(client: Client):
     activity_context = ActivitySerializationContext(
-        namespace="default",
+        namespace=client.namespace,
         workflow_id="workflow-id",
         workflow_type="workflow-type",
         activity_type="activity-type",
@@ -742,7 +742,7 @@ async def test_signal_payload_conversion(
 
         workflow_context = dataclasses.asdict(
             WorkflowSerializationContext(
-                namespace="default",
+                namespace=client.namespace,
                 workflow_id=workflow_id,
             )
         )
@@ -811,7 +811,7 @@ async def test_query_payload_conversion(
 
         workflow_context = dataclasses.asdict(
             WorkflowSerializationContext(
-                namespace="default",
+                namespace=client.namespace,
                 workflow_id=workflow_id,
             )
         )
@@ -909,7 +909,7 @@ async def test_update_payload_conversion(
 
         workflow_context = dataclasses.asdict(
             WorkflowSerializationContext(
-                namespace="default",
+                namespace=client.namespace,
                 workflow_id=workflow_id,
             )
         )
@@ -1016,13 +1016,13 @@ async def test_external_workflow_signal_and_cancel_payload_conversion(
 
         signaler_context = dataclasses.asdict(
             WorkflowSerializationContext(
-                namespace="default",
+                namespace=client.namespace,
                 workflow_id=signaler_workflow_id,
             )
         )
         target_context = dataclasses.asdict(
             WorkflowSerializationContext(
-                namespace="default",
+                namespace=client.namespace,
                 workflow_id=target_workflow_id,
             )
         )
@@ -1157,13 +1157,13 @@ async def test_failure_converter_with_context(client: Client):
 
         workflow_context = dataclasses.asdict(
             WorkflowSerializationContext(
-                namespace="default",
+                namespace=client.namespace,
                 workflow_id=workflow_id,
             )
         )
         activity_context = dataclasses.asdict(
             ActivitySerializationContext(
-                namespace="default",
+                namespace=client.namespace,
                 workflow_id=workflow_id,
                 workflow_type=FailureConverterTestWorkflow.__name__,
                 activity_type=failing_activity.__name__,
@@ -1731,6 +1731,7 @@ async def run(self, _data: str) -> None:
         )
 
 
+@pytest.mark.requires_local_server
 async def test_nexus_payload_codec_operations_lack_context(
     env: WorkflowEnvironment,
 ):
diff --git a/tests/test_service.py b/tests/test_service.py
index 0cf06fae0..954c87092 100644
--- a/tests/test_service.py
+++ b/tests/test_service.py
@@ -149,6 +149,7 @@ async def test_check_health(client: Client):
     assert err.value.status == temporalio.service.RPCStatusCode.NOT_FOUND
 
 
+@pytest.mark.requires_local_server
 async def test_grpc_status(client: Client, env: WorkflowEnvironment):
     if env.supports_time_skipping:
         pytest.skip(
diff --git a/tests/testing/test_workflow.py b/tests/testing/test_workflow.py
index d5f2aae5c..b47d0ac05 100644
--- a/tests/testing/test_workflow.py
+++ b/tests/testing/test_workflow.py
@@ -252,6 +252,7 @@ def assert_proper_error(err: BaseException | None) -> None:
             assert_proper_error(err.value.cause)
 
 
+@pytest.mark.requires_local_server
 async def test_search_attributes_on_dev_server(
     client: Client, env: WorkflowEnvironment
 ):
diff --git a/tests/worker/test_extstore.py b/tests/worker/test_extstore.py
index 998a2bd27..2f8fde5fe 100644
--- a/tests/worker/test_extstore.py
+++ b/tests/worker/test_extstore.py
@@ -181,9 +181,7 @@ async def test_extstore_activity_input_no_retrieve(
     WorkflowFailureError wrapping an ActivityError."""
     driver = BadTestDriver(no_retrieve=True)
 
-    client = await Client.connect(
-        env.client.service_client.config.target_host,
-        namespace=env.client.namespace,
+    client = await env.connect_client(
         data_converter=dataclasses.replace(
             temporalio.converter.default(),
             external_storage=ExternalStorage(
@@ -225,9 +223,7 @@ async def test_extstore_activity_result_no_store(
     terminates with a WorkflowFailureError wrapping an ActivityError."""
     driver = BadTestDriver(no_store=True)
 
-    client = await Client.connect(
-        env.client.service_client.config.target_host,
-        namespace=env.client.namespace,
+    client = await env.connect_client(
         data_converter=dataclasses.replace(
             temporalio.converter.default(),
             external_storage=ExternalStorage(
@@ -274,9 +270,7 @@ async def test_extstore_worker_missing_driver(
     """
     driver = InMemoryTestDriver()
 
-    far_client = await Client.connect(
-        env.client.service_client.config.target_host,
-        namespace=env.client.namespace,
+    far_client = await env.connect_client(
         data_converter=dataclasses.replace(
             temporalio.converter.default(),
             external_storage=ExternalStorage(
@@ -286,10 +280,7 @@ async def test_extstore_worker_missing_driver(
         ),
     )
 
-    worker_client = await Client.connect(
-        env.client.service_client.config.target_host,
-        namespace=env.client.namespace,
-    )
+    worker_client = await env.connect_client()
 
     async with new_worker(
         worker_client, ExtStoreWorkflow, activities=[ext_store_activity]
@@ -315,9 +306,7 @@ async def test_extstore_payload_not_found_fails_workflow(
     """When a non-retryable ApplicationError is raised while retrieving workflow input,
     the workflow must fail terminally (not retry as a task failure).
     """
-    client = await Client.connect(
-        env.client.service_client.config.target_host,
-        namespace=env.client.namespace,
+    client = await env.connect_client(
         data_converter=dataclasses.replace(
             temporalio.converter.default(),
             external_storage=ExternalStorage(
@@ -363,9 +352,7 @@ async def _run_extstore_workflow_and_fetch_history(
     activity_output_size: int = 10,
 ) -> WorkflowHandle:
     """Helper: run ExtStoreWorkflow with the given driver and return its history handle."""
-    extstore_client = await Client.connect(
-        env.client.service_client.config.target_host,
-        namespace=env.client.namespace,
+    extstore_client = await env.connect_client(
         data_converter=dataclasses.replace(
             temporalio.converter.default(),
             external_storage=ExternalStorage(
@@ -515,9 +502,7 @@ async def test_extstore_chained_activities(
     """
     driver = InMemoryTestDriver()
 
-    client = await Client.connect(
-        env.client.service_client.config.target_host,
-        namespace=env.client.namespace,
+    client = await env.connect_client(
         data_converter=dataclasses.replace(
             temporalio.converter.default(),
             external_storage=ExternalStorage(
@@ -567,9 +552,7 @@ def __init__(self, driver_name: str):
     driver2 = InMemoryTestDriver(driver_name="driver2")
     driver3 = DifferentTestDriver(driver_name="driver3")
 
-    client = await Client.connect(
-        env.client.service_client.config.target_host,
-        namespace=env.client.namespace,
+    client = await env.connect_client(
         data_converter=dataclasses.replace(
             temporalio.converter.default(),
             external_storage=ExternalStorage(
@@ -685,9 +668,7 @@ async def test_tmprl1104_with_extstore_download(env: WorkflowEnvironment) -> Non
             payload_size_threshold=512,
         ),
     )
-    client = await Client.connect(
-        env.client.service_client.config.target_host,
-        namespace=env.client.namespace,
+    client = await env.connect_client(
         data_converter=data_converter,
     )
 
@@ -737,9 +718,7 @@ async def test_tmprl1104_with_extstore_upload(env: WorkflowEnvironment) -> None:
             payload_size_threshold=512,
         ),
     )
-    client = await Client.connect(
-        env.client.service_client.config.target_host,
-        namespace=env.client.namespace,
+    client = await env.connect_client(
         data_converter=data_converter,
     )
 
@@ -791,9 +770,7 @@ async def test_tmprl1104_with_extstore_download_and_upload(
             payload_size_threshold=512,
         ),
     )
-    client = await Client.connect(
-        env.client.service_client.config.target_host,
-        namespace=env.client.namespace,
+    client = await env.connect_client(
         data_converter=data_converter,
     )
 
@@ -914,9 +891,7 @@ async def _make_tracking_client(
     env: WorkflowEnvironment,
 ) -> tuple[Client, ContextTrackingStorageDriver]:
     driver = ContextTrackingStorageDriver()
-    client = await Client.connect(
-        env.client.service_client.config.target_host,
-        namespace=env.client.namespace,
+    client = await env.connect_client(
         data_converter=dataclasses.replace(
             temporalio.converter.default(),
             external_storage=ExternalStorage(
diff --git a/tests/worker/test_interceptor.py b/tests/worker/test_interceptor.py
index 431f8280d..4a1da399d 100644
--- a/tests/worker/test_interceptor.py
+++ b/tests/worker/test_interceptor.py
@@ -269,6 +269,7 @@ def update_validated_validator(self, param: str) -> None:
             raise ApplicationError("Invalid update")
 
 
+@pytest.mark.requires_local_server
 async def test_worker_interceptor(client: Client, env: WorkflowEnvironment):
     # TODO(cretz): Fix
     if env.supports_time_skipping:
diff --git a/tests/worker/test_payload_size_limits.py b/tests/worker/test_payload_size_limits.py
index b527f429c..18cc7e75a 100644
--- a/tests/worker/test_payload_size_limits.py
+++ b/tests/worker/test_payload_size_limits.py
@@ -7,7 +7,7 @@
 
 import temporalio.api.enums.v1
 from temporalio import activity, workflow
-from temporalio.client import Client, PayloadLimitsConfig, WorkflowFailureError
+from temporalio.client import PayloadLimitsConfig, WorkflowFailureError
 from temporalio.exceptions import (
     TerminatedError,
     TimeoutError,
@@ -91,9 +91,7 @@ async def test_oversized_payload_fails_task_with_error_log(env: WorkflowEnvironm
         dev_server_download_version=DEV_SERVER_DOWNLOAD_VERSION,
     ) as env:
         worker_logger = logging.getLogger(f"log-{uuid.uuid4()}")
-        worker_client = await Client.connect(
-            env.client.service_client.config.target_host,
-            namespace=env.client.namespace,
+        worker_client = await env.connect_client(
             runtime=_forwarding_runtime(worker_logger),
         )
 
@@ -179,9 +177,7 @@ async def test_disable_payload_error_limit_sends_to_server(env: WorkflowEnvironm
 async def test_payload_size_warning_forwarded(env: WorkflowEnvironment):
     """The connection's warn threshold produces a forwarded [TMPRL1103] warning for over-threshold payloads."""
     worker_logger = logging.getLogger(f"log-{uuid.uuid4()}")
-    worker_client = await Client.connect(
-        env.client.service_client.config.target_host,
-        namespace=env.client.namespace,
+    worker_client = await env.connect_client(
         runtime=_forwarding_runtime(worker_logger),
         payload_limits=PayloadLimitsConfig(payloads_warn_size=1024),
     )
@@ -219,9 +215,7 @@ async def test_memo_size_warning_forwarded(env: WorkflowEnvironment):
     """The connection's memo warn threshold produces a forwarded [TMPRL1103] warning for an
     over-threshold memo."""
     worker_logger = logging.getLogger(f"log-{uuid.uuid4()}")
-    worker_client = await Client.connect(
-        env.client.service_client.config.target_host,
-        namespace=env.client.namespace,
+    worker_client = await env.connect_client(
         runtime=_forwarding_runtime(worker_logger),
         payload_limits=PayloadLimitsConfig(memo_warn_size=1024),
     )
diff --git a/tests/worker/test_replayer.py b/tests/worker/test_replayer.py
index 137b32c3e..52fed7416 100644
--- a/tests/worker/test_replayer.py
+++ b/tests/worker/test_replayer.py
@@ -302,6 +302,7 @@ async def test_replayer_workflow_not_registered(client: Client) -> None:
     assert "SayHelloWorkflow is not registered" in str(err.value)
 
 
+@pytest.mark.requires_local_server
 async def test_replayer_multiple_from_client(
     client: Client, env: WorkflowEnvironment
 ) -> None:
@@ -330,6 +331,14 @@ async def test_replayer_multiple_from_client(
             )
             await handle.result()
 
+    async def visible_run_ids() -> set[str]:
+        return {
+            workflow.run_id
+            async for workflow in client.list_workflows(f"WorkflowId = '{workflow_id}'")
+        }
+
+    await assert_eq_eventually(set(expected_runs_and_non_det), visible_run_ids)
+
     # Run replayer with list iterator mapped to histories and collect results
     async with Replayer(workflows=[SayHelloWorkflow]).workflow_replay_iterator(
         client.list_workflows(f"WorkflowId = '{workflow_id}'").map_histories()
diff --git a/tests/worker/test_update_with_start.py b/tests/worker/test_update_with_start.py
index 2ceb5e91b..0b3368725 100644
--- a/tests/worker/test_update_with_start.py
+++ b/tests/worker/test_update_with_start.py
@@ -192,6 +192,7 @@ async def _do_test(
         id_conflict_policy: WorkflowIDConflictPolicy,
         expect_error_when_workflow_exists: ExpectErrorWhenWorkflowExists,
     ):
+        workflow_id = f"{workflow_id}-{uuid.uuid4()}"
         await self._do_execute_update_test(
             client,
             workflow_id + "-execute-update",
@@ -336,6 +337,7 @@ async def test_update_with_start_sets_first_execution_run_id(
         WorkflowForUpdateWithStartTest,
         activities=[activity_called_by_update],
     ) as worker:
+        workflow_id_prefix = f"wid-{uuid.uuid4()}"
 
         def make_start_op(workflow_id: str):
             return WithStartWorkflowOperation(
@@ -348,7 +350,7 @@ def make_start_op(workflow_id: str):
 
         # conflict policy is FAIL
         # First UWS succeeds and sets the first execution run ID
-        start_op_1 = make_start_op("wid-1")
+        start_op_1 = make_start_op(f"{workflow_id_prefix}-1")
         update_handle_1 = await client.start_update_with_start_workflow(
             WorkflowForUpdateWithStartTest.my_non_blocking_update,
             "1",
@@ -360,7 +362,7 @@ def make_start_op(workflow_id: str):
 
         # Second UWS start fails because the workflow already exists
         # first execution run ID is not set on the second UWS handle
-        start_op_2 = make_start_op("wid-1")
+        start_op_2 = make_start_op(f"{workflow_id_prefix}-1")
 
         for aw in [
             client.start_update_with_start_workflow(
@@ -375,7 +377,7 @@ def make_start_op(workflow_id: str):
                 await aw
 
         # Third UWS start succeeds, but the update fails after acceptance
-        start_op_3 = make_start_op("wid-2")
+        start_op_3 = make_start_op(f"{workflow_id_prefix}-2")
         update_handle_3 = await client.start_update_with_start_workflow(
             WorkflowForUpdateWithStartTest.my_non_blocking_update,
             "fail-after-acceptance",
@@ -394,7 +396,7 @@ def make_start_op(workflow_id: str):
         assert await wf_handle_3.result() == "workflow-result-0"
 
         # Fourth UWS is same as third, but we use execute_update instead of start_update.
-        start_op_4 = make_start_op("wid-3")
+        start_op_4 = make_start_op(f"{workflow_id_prefix}-3")
         with pytest.raises(WorkflowUpdateFailedError):
             await client.execute_update_with_start_workflow(
                 WorkflowForUpdateWithStartTest.my_non_blocking_update,
diff --git a/tests/worker/test_worker.py b/tests/worker/test_worker.py
index 3a4e9165b..b6ce37c8b 100644
--- a/tests/worker/test_worker.py
+++ b/tests/worker/test_worker.py
@@ -399,6 +399,7 @@ def my_signal(self, value: str) -> None:
         workflow.logger.info(f"Signal: {value}")
 
 
+@pytest.mark.requires_local_server
 async def test_custom_slot_supplier(client: Client, env: WorkflowEnvironment):
     if env.supports_time_skipping:
         pytest.skip("Nexus tests don't work under Java test server")
@@ -1206,7 +1207,9 @@ async def test_workflows_can_use_versioning_override(
         )
 
 
-async def test_can_run_autoscaling_polling_worker(client: Client):
+async def test_can_run_autoscaling_polling_worker(
+    client: Client, env: WorkflowEnvironment
+):
     # Create new runtime with Prom server
     prom_addr = f"127.0.0.1:{find_free_port()}"
     runtime = Runtime(
@@ -1214,9 +1217,7 @@ async def test_can_run_autoscaling_polling_worker(client: Client):
             metrics=PrometheusConfig(bind_address=prom_addr),
         )
     )
-    client = await Client.connect(
-        client.service_client.config.target_host,
-        namespace=client.namespace,
+    client = await env.connect_client(
         runtime=runtime,
     )
 
@@ -1467,15 +1468,14 @@ def test_fork_use_worker(
         self.run(mp_fork_ctx)
 
 
-async def test_activity_client_updates_when_worker_client_changes(client: Client):
+async def test_activity_client_updates_when_worker_client_changes(
+    client: Client, env: WorkflowEnvironment
+):
     """Test that activities get the updated client when worker.client is changed."""
     # Create a second client (simulating a new client after cert rotation)
     # Must use the same runtime
-    client2 = await Client.connect(
-        client.service_client.config.target_host,
-        namespace=client.namespace,
+    client2 = await env.connect_client(
         data_converter=client.data_converter,
-        runtime=client.service_client.config.runtime,
     )
 
     captured_clients: list[Client] = []
diff --git a/tests/worker/test_workflow.py b/tests/worker/test_workflow.py
index 9c742d5c6..5bb2b13e1 100644
--- a/tests/worker/test_workflow.py
+++ b/tests/worker/test_workflow.py
@@ -308,6 +308,7 @@ def get_history_info(self) -> HistoryInfo:
         )
 
 
+@pytest.mark.requires_local_server
 async def test_workflow_history_info(
     client: Client, env: WorkflowEnvironment, continue_as_new_suggest_history_count: int
 ):
@@ -2352,6 +2353,7 @@ def do_search_attribute_update_typed(self) -> None:
         )
 
 
+@pytest.mark.requires_local_server
 async def test_workflow_search_attributes(client: Client, env_type: str):
     if env_type != "local":
         pytest.skip("Only testing search attributes on local which disables cache")
@@ -2537,6 +2539,7 @@ async def run(self) -> None:
         # All we need to do is complete
 
 
+@pytest.mark.requires_local_server
 async def test_workflow_no_initial_search_attributes(client: Client, env_type: str):
     if env_type != "local":
         pytest.skip("Only testing search attributes on local which disables cache")
@@ -4957,7 +4960,7 @@ async def run(self) -> None:
         )
 
 
-async def test_workflow_custom_metrics(client: Client):
+async def test_workflow_custom_metrics(client: Client, env: WorkflowEnvironment):
     # Run worker with default runtime which is noop meter just to confirm it
     # doesn't fail
     async with new_worker(
@@ -4983,9 +4986,7 @@ async def test_workflow_custom_metrics(client: Client):
     assert str(err.value).startswith("Invalid value type for key")
 
     # New client with the runtime
-    client = await Client.connect(
-        client.service_client.config.target_host,
-        namespace=client.namespace,
+    client = await env.connect_client(
         runtime=runtime,
     )
 
@@ -5062,7 +5063,7 @@ async def test_workflow_custom_metrics(client: Client):
         )
 
 
-async def test_workflow_buffered_metrics(client: Client):
+async def test_workflow_buffered_metrics(client: Client, env: WorkflowEnvironment):
     # Create runtime with metric buffer
     buffer = MetricBuffer(10000)
     runtime = Runtime(
@@ -5123,9 +5124,7 @@ async def test_workflow_buffered_metrics(client: Client):
     assert runtime_updates2[1].value == 400
 
     # Create a new client on the runtime and execute the custom metric workflow
-    client = await Client.connect(
-        client.service_client.config.target_host,
-        namespace=client.namespace,
+    client = await env.connect_client(
         runtime=runtime,
     )
     async with new_worker(
@@ -5186,12 +5185,10 @@ async def test_workflow_buffered_metrics(client: Client):
     )
 
 
-async def test_workflow_metrics_other_types(client: Client):
+async def test_workflow_metrics_other_types(env: WorkflowEnvironment):
     async def do_stuff(buffer: MetricBuffer) -> None:
         runtime = Runtime(telemetry=TelemetryConfig(metrics=buffer))
-        new_client = await Client.connect(
-            client.service_client.config.target_host,
-            namespace=client.namespace,
+        new_client = await env.connect_client(
             runtime=runtime,
         )
         async with new_worker(new_client, HelloWorkflow) as worker:
@@ -6211,6 +6208,7 @@ async def run(self) -> None:
             await asyncio.sleep(0.1)
 
 
+@pytest.mark.requires_local_server
 async def test_workflow_replace_worker_client(client: Client, env: WorkflowEnvironment):
     if env.supports_time_skipping:
         pytest.skip("Only testing against two real servers")
@@ -6266,11 +6264,11 @@ async def any_task_completed(handle: WorkflowHandle) -> bool:
             await handle2.terminate()
 
 
-async def test_workflow_replace_worker_client_diff_runtimes_fail(client: Client):
+async def test_workflow_replace_worker_client_diff_runtimes_fail(
+    client: Client, env: WorkflowEnvironment
+):
     other_runtime = Runtime(telemetry=TelemetryConfig())
-    other_client = await Client.connect(
-        client.service_client.config.target_host,
-        namespace=client.namespace,
+    other_client = await env.connect_client(
         runtime=other_runtime,
     )
     async with new_worker(client, HelloWorkflow) as worker:
@@ -6711,6 +6709,9 @@ async def my_dynamic_signal(self, _name: str, _args: Sequence[RawValue]) -> None
         await workflow.wait_condition(lambda: self.handlers_may_finish)
 
 
+# Cloud cancels in-flight dynamic handlers before SDK teardown can emit the
+# unfinished-handler warning this test asserts.
+@pytest.mark.requires_local_server
 @pytest.mark.parametrize("handler_type", ["-signal-", "-update-"])
 @pytest.mark.parametrize(
     "handler_registration", ["-late-registered-", "-not-late-registered-"]
@@ -9404,6 +9405,7 @@ async def run(self, name: str) -> str:
         return f"Hello from child, {name}"
 
 
+@pytest.mark.requires_local_server
 async def test_search_attribute_codec(client: Client, env_type: str):
     if env_type != "local":
         pytest.skip("Only testing search attributes on local which disables cache")