Support payload-free system Nexus roots (#1783)

* Support payload-free system Nexus roots

* Fixing time-skipping test error

* Update visit exception thrown

* sdk core update

* Revert "sdk core update"

This reverts commit d2f3219.

Author: JoshuaFrenchwood

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

temporalio/bridge/_visitor.py

@@ -645,3 +645,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

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)

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"

tests/nexus/test_temporal_operation.py

@@ -44,9 +44,6 @@
 # See https://github.com/temporalio/sdk-python/issues/1704.
 pytestmark = pytest.mark.requires_local_server
 
-# Query response links require a newer server than the shared test environment.
-_QUERY_LINK_DEV_SERVER_DOWNLOAD_VERSION = "v1.8.3-server-1.32.0-162.0"
-
 
 @dataclass
 class Input:
@@ -854,14 +851,7 @@ async def run(self, input: Input) -> bool:
         return await client.execute_operation(TestService.query_op, input.value)
 
 
-async def test_temporal_operation_query_workflow() -> None:
-    async with await WorkflowEnvironment.start_local(
-        dev_server_download_version=_QUERY_LINK_DEV_SERVER_DOWNLOAD_VERSION
-    ) as env:
-        await _assert_temporal_operation_query_workflow(env.client, env)
-
-
-async def _assert_temporal_operation_query_workflow(
+async def test_temporal_operation_query_workflow(
     client: Client, env: WorkflowEnvironment
 ) -> None:
     task_queue = str(uuid.uuid4())
@@ -900,15 +890,17 @@ async def _assert_temporal_operation_query_workflow(
             target_history = await target_handle.fetch_history()
             assert not any(event.links for event in target_history.events)
 
-            assert target_handle.result_run_id is not None
-            assert Link(
-                workflow=Link.Workflow(
-                    namespace=client.namespace,
-                    workflow_id=target_workflow_id,
-                    run_id=target_handle.result_run_id,
-                    reason="Query processed",
-                )
-            ) in list(completed_event.links)
+            # The Java time-skipping test server does not return Nexus operation links.
+            if not env.supports_time_skipping:
+                assert target_handle.result_run_id is not None
+                assert Link(
+                    workflow=Link.Workflow(
+                        namespace=client.namespace,
+                        workflow_id=target_workflow_id,
+                        run_id=target_handle.result_run_id,
+                        reason="Query processed",
+                    )
+                ) in list(completed_event.links)
         finally:
             await target_handle.cancel()
 
@@ -1314,14 +1306,8 @@ async def test_temporal_operation_start_activity_raises_error(
                 id=str(uuid.uuid4()),
             )
 
-        operation_err = err.value.__cause__
-        assert isinstance(operation_err, temporalio.exceptions.ApplicationError)
-        assert operation_err.type == "OperationError"
-        assert "nexus operation completed unsuccessfully" in str(operation_err)
-
-        application_err = operation_err.__cause__
+        application_err = err.value.__cause__
         assert isinstance(application_err, temporalio.exceptions.ApplicationError)
-
         assert application_err.type == "test-activity-error-type"
         assert "test-activity-error-message" in str(application_err)
         assert application_err.__cause__ is None

tests/worker/test_visitor.py

@@ -9,6 +9,7 @@
 import temporalio.api.workflowservice.v1.request_response_pb2 as workflowservice_pb2
 import temporalio.bridge.worker
 import temporalio.converter
+import temporalio.exceptions
 import temporalio.nexus.system as nexus_system
 from temporalio.api.common.v1.message_pb2 import (
     Payload,
@@ -266,6 +267,75 @@ async def visit_system_nexus_envelope(self, payload: Payload) -> None:
     assert visitor.system_envelope_count == 1
 
 
+async def test_system_nexus_envelope_without_payloads_is_visited():
+    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
+
+    response = workflowservice_pb2.SignalWithStartWorkflowExecutionResponse(
+        run_id="test-run-id"
+    )
+    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(response)
+    assert system_payload is not None
+    completion = WorkflowActivationCompletion(
+        run_id="3",
+        successful=Success(
+            commands=[
+                WorkflowCommand(
+                    update_response=UpdateResponse(completed=system_payload),
+                )
+            ]
+        ),
+    )
+    visitor = SystemNexusVisitor()
+
+    await PayloadVisitor().visit(visitor, completion)
+
+    completed = completion.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
+    completion = WorkflowActivationCompletion(
+        run_id="3",
+        successful=Success(
+            commands=[
+                WorkflowCommand(
+                    update_response=UpdateResponse(completed=system_payload),
+                )
+            ]
+        ),
+    )
+
+    with pytest.raises(temporalio.exceptions.ApplicationError) as err:
+        await PayloadVisitor().visit(Visitor(), completion)
+
+    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