diff --git a/scripts/gen_payload_visitor.py b/scripts/gen_payload_visitor.py
index 53d61611b..7fab85dcd 100644
--- a/scripts/gen_payload_visitor.py
+++ b/scripts/gen_payload_visitor.py
@@ -124,12 +124,20 @@ def generate(self, roots: list[Descriptor]) -> str:
 
         The generated code defines async visitor functions for each reachable
         protobuf message type starting from WorkflowActivation, including support
-        for repeated fields and map entries, and a convenience entrypoint
-        function `visit`.
+        for repeated fields and map entries. Payload-free roots get no-op methods
+        so the `visit` entrypoint recognizes them as supported.
         """
 
-        for r in roots:
-            self.walk(r)
+        for root in roots:
+            if not self.walk(root):
+                self.methods.append(
+                    f"""\
+    async def _visit_{name_for(root)}(
+        self, fs: VisitorFunctions, o: Any
+    ) -> None:
+        pass
+"""
+                )
 
         header = """
 from __future__ import annotations
diff --git a/temporalio/bridge/_visitor.py b/temporalio/bridge/_visitor.py
index 2bccd3b4a..e16d70500 100644
--- a/temporalio/bridge/_visitor.py
+++ b/temporalio/bridge/_visitor.py
@@ -635,3 +635,8 @@ async def _visit_temporal_api_workflowservice_v1_SignalWithStartWorkflowExecutio
             await self._visit_temporal_api_common_v1_Header(fs, o.header)
         if o.HasField("user_metadata"):
             await self._visit_temporal_api_sdk_v1_UserMetadata(fs, o.user_metadata)
+
+    async def _visit_temporal_api_workflowservice_v1_SignalWithStartWorkflowExecutionResponse(
+        self, fs: VisitorFunctions, o: Any
+    ) -> None:
+        pass
diff --git a/temporalio/nexus/system/__init__.py b/temporalio/nexus/system/__init__.py
index 730dd4258..1f8f7c6d2 100644
--- a/temporalio/nexus/system/__init__.py
+++ b/temporalio/nexus/system/__init__.py
@@ -15,6 +15,7 @@
 import temporalio.api.common.v1
 import temporalio.common
 import temporalio.converter
+import temporalio.exceptions
 from temporalio.bridge._visitor_functions import VisitorFunctions
 from temporalio.converter import BinaryProtoPayloadConverter, CompositePayloadConverter
 from temporalio.converter._payload_converter import (
@@ -154,7 +155,14 @@ async def maybe_visit_payload(
 
     payload_visitor = PayloadVisitor(skip_search_attributes=skip_search_attributes)
     checkpoint = visitor_functions.checkpoint()
-    await payload_visitor.visit(visitor_functions, value)
+    try:
+        await payload_visitor.visit(visitor_functions, value)
+    except ValueError as err:
+        if not str(err).startswith("Unknown root message type: "):
+            raise
+        raise temporalio.exceptions.ApplicationError(
+            f"Unknown Temporal system payload: {value.DESCRIPTOR.full_name}"
+        ) from err
     if checkpoint is not None:
         await visitor_functions.drain_since(checkpoint)
     return payload_converter.to_payload(value)
diff --git a/tests/__init__.py b/tests/__init__.py
index af97849fe..eff2c8adb 100644
--- a/tests/__init__.py
+++ b/tests/__init__.py
@@ -1 +1 @@
-DEV_SERVER_DOWNLOAD_VERSION = "v1.7.4-standalone-nexus-operations"
+DEV_SERVER_DOWNLOAD_VERSION = "v1.8.3-server-1.32.0-162.0"
diff --git a/tests/worker/test_visitor.py b/tests/worker/test_visitor.py
index 76916aaa7..d81f22465 100644
--- a/tests/worker/test_visitor.py
+++ b/tests/worker/test_visitor.py
@@ -38,6 +38,7 @@
     Success,
     WorkflowActivationCompletion,
 )
+from temporalio.exceptions import ApplicationError
 from tests.worker.test_workflow import SimpleCodec
 
 
@@ -266,6 +267,75 @@ async def visit_system_nexus_envelope(self, payload: Payload) -> None:
     assert visitor.system_envelope_count == 1
 
 
+async def test_payload_free_system_nexus_envelope_is_detected():
+    class SystemNexusVisitor(Visitor):
+        def __init__(self) -> None:
+            self.system_envelope_count = 0
+
+        async def visit_system_nexus_envelope(self, payload: Payload) -> None:
+            _ = payload
+            self.system_envelope_count += 1
+
+    data_converter = temporalio.converter.default()
+    payload_converter = nexus_system._get_payload_converter(
+        data_converter.payload_converter,
+        data_converter.failure_converter,
+    )
+    response = workflowservice_pb2.SignalWithStartWorkflowExecutionResponse(
+        run_id="test-run-id"
+    )
+    system_payload = payload_converter.to_payload(response)
+    assert system_payload is not None
+    comp = WorkflowActivationCompletion(
+        run_id="3",
+        successful=Success(
+            commands=[
+                WorkflowCommand(
+                    update_response=UpdateResponse(completed=system_payload),
+                )
+            ]
+        ),
+    )
+    visitor = SystemNexusVisitor()
+
+    await PayloadVisitor().visit(visitor, comp)
+
+    completed = comp.successful.commands[0].update_response.completed
+    assert payload_converter.from_payload(completed) == response
+    assert visitor.system_envelope_count == 1
+
+
+async def test_unknown_system_nexus_payload_raises_application_error():
+    data_converter = temporalio.converter.default()
+    payload_converter = nexus_system._get_payload_converter(
+        data_converter.payload_converter,
+        data_converter.failure_converter,
+    )
+    system_payload = payload_converter.to_payload(
+        workflowservice_pb2.StartWorkflowExecutionResponse(run_id="test-run-id")
+    )
+    assert system_payload is not None
+    comp = WorkflowActivationCompletion(
+        run_id="3",
+        successful=Success(
+            commands=[
+                WorkflowCommand(
+                    update_response=UpdateResponse(completed=system_payload),
+                )
+            ]
+        ),
+    )
+
+    with pytest.raises(ApplicationError) as err:
+        await PayloadVisitor().visit(Visitor(), comp)
+
+    assert (
+        err.value.message == "Unknown Temporal system payload: "
+        "temporal.api.workflowservice.v1.StartWorkflowExecutionResponse"
+    )
+    assert not err.value.non_retryable
+
+
 async def test_concurrent_throughput():
     """Demonstrate that concurrent visitation is faster than serialized for I/O-bound codecs."""
     N_CMDS = 10