fix: send unknown System Nexus operations to the generic interceptor hook

.github/workflows/ci.yml

@@ -229,7 +229,7 @@ jobs:
       - run: uv add --dev --python 3.10 "googleapis-common-protos==1.70.0"
       - run: uv add --python 3.10 "protobuf<4"
       - run: uv sync --all-extras
-      - run: cargo install --locked nexgen --version 0.2.8 --features advanced --force
+      - run: cargo install --locked nexgen --version 0.2.11 --features advanced --force
       - run: poe build-develop
       - run: poe gen-protos
       - name: Check generation unchanged

changelog/fixed/wandering-walrus.md

@@ -0,0 +1 @@
+System Nexus operations with no generated interceptor now go to the generic `start_system_nexus_operation` hook of the Workflow outbound interceptors, instead of raising.

scripts/gen_nexus_system_api.py

@@ -25,7 +25,7 @@
 workflow_output_dir = base_dir / "temporalio" / "nexus" / "system" / "workflow_service"
 notification_output_dir = base_dir / "temporalio" / "nexus" / "notifications"
 workflow_init_path = base_dir / "temporalio" / "workflow" / "__init__.py"
-NEX_GEN_VERSION = "0.2.8"
+NEX_GEN_VERSION = "0.2.11"
 proto_files = [
     gen_protos.api_proto_dir
     / "temporal"

temporalio/nexus/notifications/__init__.py

@@ -1,4 +1,4 @@
-# Generated by nexgen v0.2.8. DO NOT EDIT!
+# Generated by nexgen v0.2.11. DO NOT EDIT!
 
 from __future__ import annotations
 

temporalio/nexus/notifications/models.py

@@ -1,4 +1,4 @@
-# Generated by nexgen v0.2.8. DO NOT EDIT!
+# Generated by nexgen v0.2.11. DO NOT EDIT!
 
 from __future__ import annotations
 

temporalio/nexus/system/workflow_service/__init__.py

@@ -1,4 +1,4 @@
-# Generated by nexgen v0.2.8. DO NOT EDIT!
+# Generated by nexgen v0.2.11. DO NOT EDIT!
 
 from __future__ import annotations
 

temporalio/nexus/system/workflow_service/_system_nexus_interceptor.py

@@ -1,4 +1,4 @@
-# Generated by nexgen v0.2.8. DO NOT EDIT!
+# Generated by nexgen v0.2.11. DO NOT EDIT!
 
 from __future__ import annotations
 
@@ -40,9 +40,7 @@ async def _start_system_nexus_operation(
             "temporalio.workflow.NexusOperationHandle[_OutputT]",
             await interceptor.start_signal_with_start_workflow(typed_input.input),
         )
-    raise ValueError(
-        f"unsupported System Nexus operation: {input.service}/{input.operation_name}"
-    )
+    return await interceptor.start_system_nexus_operation(input)
 
 
 class _SystemNexusWorkflowOutboundInterceptorBase(abc.ABC):
@@ -51,6 +49,14 @@ def _next_system_nexus_interceptor(
         self,
     ) -> _SystemNexusWorkflowOutboundInterceptorBase: ...
 
+    @abc.abstractmethod
+    async def start_system_nexus_operation(
+        self,
+        input: StartNexusOperationInput[_InputT, _OutputT],
+    ) -> temporalio.workflow.NexusOperationHandle[_OutputT]:
+        """Intercept a System Nexus operation through the generic hook."""
+        ...
+
     async def start_signal_with_start_workflow(
         self, request: models.SignalWithStartWorkflowRequest
     ) -> temporalio.workflow.NexusOperationHandle[

temporalio/nexus/system/workflow_service/models.py

@@ -1,4 +1,4 @@
-# Generated by nexgen v0.2.8. DO NOT EDIT!
+# Generated by nexgen v0.2.11. DO NOT EDIT!
 
 from __future__ import annotations
 
@@ -18,7 +18,8 @@
 
 @dataclasses.dataclass(slots=True, kw_only=True)
 class SignalWithStartWorkflowRequest:
-    """
+    """Request fields for signaling a workflow, starting it first if needed.
+
     .. warning::
         This API is experimental and subject to change.
     """
@@ -223,7 +224,8 @@ def to_transfer_type(
 
 @dataclasses.dataclass(slots=True)
 class UserMetadata:
-    """
+    """Static metadata for a workflow execution.
+
     .. warning::
         This API is experimental and subject to change.
     """
@@ -276,7 +278,8 @@ def to_transfer_type(
 
 @dataclasses.dataclass(slots=True)
 class SignalWithStartWorkflowResponse:
-    """
+    """Result of signaling a workflow and starting it if needed.
+
     .. warning::
         This API is experimental and subject to change.
     """

temporalio/nexus/system/workflow_service/operations/__init__.py

@@ -1,3 +1,3 @@
-# Generated by nexgen v0.2.8. DO NOT EDIT!
+# Generated by nexgen v0.2.11. DO NOT EDIT!
 
 from __future__ import annotations

temporalio/nexus/system/workflow_service/operations/signal_with_start_workflow.py

@@ -1,4 +1,4 @@
-# Generated by nexgen v0.2.8. DO NOT EDIT!
+# Generated by nexgen v0.2.11. DO NOT EDIT!
 
 from __future__ import annotations
 

temporalio/nexus/system/workflow_service/services.py

@@ -1,4 +1,4 @@
-# Generated by nexgen v0.2.8. DO NOT EDIT!
+# Generated by nexgen v0.2.11. DO NOT EDIT!
 
 from __future__ import annotations
 

tests/nexus/test_system_nexus_fallthrough.py

@@ -0,0 +1,85 @@
+from __future__ import annotations
+
+import uuid
+from collections.abc import Generator
+from datetime import timedelta
+from typing import Any
+
+from temporalio import workflow
+from temporalio.client import Client
+from temporalio.nexus.system import TEMPORAL_SYSTEM_ENDPOINT
+from temporalio.worker import (
+    Interceptor,
+    StartNexusOperationInput,
+    WorkflowInboundInterceptor,
+    WorkflowInterceptorClassInput,
+    WorkflowOutboundInterceptor,
+)
+from tests.helpers import new_worker
+
+PLUGIN_SERVICE = "test.plugin.v1.EchoService"
+
+
+class _DoneHandle(workflow.NexusOperationHandle[Any]):
+    def __init__(self, value: Any) -> None:
+        self._value = value
+
+    def cancel(self) -> bool:
+        return False
+
+    def __await__(self) -> Generator[Any, Any, Any]:
+        return self._result().__await__()
+
+    async def _result(self) -> Any:
+        return self._value
+
+    @property
+    def operation_token(self) -> str | None:
+        return None
+
+
+class _EchoOutboundInterceptor(WorkflowOutboundInterceptor):
+    async def start_system_nexus_operation(
+        self, input: StartNexusOperationInput[Any, Any]
+    ) -> workflow.NexusOperationHandle[Any]:
+        if input.service == PLUGIN_SERVICE:
+            return _DoneHandle(f"{input.operation_name}: {input.input}")
+        return await super().start_system_nexus_operation(input)
+
+
+class _EchoInboundInterceptor(WorkflowInboundInterceptor):
+    def init(self, outbound: WorkflowOutboundInterceptor) -> None:
+        super().init(_EchoOutboundInterceptor(outbound))
+
+
+class EchoServiceInterceptor(Interceptor):
+    def workflow_interceptor_class(
+        self, input: WorkflowInterceptorClassInput
+    ) -> type[WorkflowInboundInterceptor] | None:
+        return _EchoInboundInterceptor
+
+
+@workflow.defn
+class PluginServiceWorkflow:
+    @workflow.run
+    async def run(self) -> str:
+        client = workflow.create_nexus_client(
+            service=PLUGIN_SERVICE, endpoint=TEMPORAL_SYSTEM_ENDPOINT
+        )
+        handle: workflow.NexusOperationHandle[Any] = await client.start_operation(
+            "Echo", "hello"
+        )
+        return await handle
+
+
+async def test_unknown_system_nexus_operation_reaches_generic_hook(client: Client):
+    async with new_worker(
+        client, PluginServiceWorkflow, interceptors=[EchoServiceInterceptor()]
+    ) as worker:
+        result = await client.execute_workflow(
+            PluginServiceWorkflow.run,
+            id=f"wf-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+            execution_timeout=timedelta(seconds=10),
+        )
+    assert result == "Echo: hello"