Nexus: worker, workflow-backed operations, and workflow caller #813

dandavison · 2025-04-02T19:06:35Z

Initial Python Temporal Nexus implementation.

Temporal SDK PR to accompany

Nexus SDK PR Initial implementation nexus-rpc/sdk-python#1
samples-python PR Nexus samples samples-python#174

Contains Nexus worker, components for users to define workflow-backed Nexus operations, and the ability to start and cancel a Nexus operation from a workflow.

Notes for reviewers

nexusrpc.handler.start_workflow is a top-level static function, but currently there's still a public contextvar object nexusrpc.handler.temporal_operation_context. Let's settle on approach there. If we're using module-level getters I think they need to be named something like get_client() etc, despite the fact that we have activity.metric_meter() ? Little bit more discussion needed there.
I'll do another cleanup pass on docstrings, API docs, etc.

This reverts commit 86a9a61.

cretz · 2025-07-03T18:21:44Z

temporalio/nexus/__init__.py

+from ._operation_context import (
+    _TemporalCancelOperationContext as _TemporalCancelOperationContext,
+)
+from ._operation_context import (
+    _TemporalStartOperationContext as _TemporalStartOperationContext,
+)


We do not expose that to users hence the underscore prefix, it just happens to be in the same file. In this multi-file module, are we intending to expose these two underscore-prefixed types to users? If not, we should remove them.

cretz · 2025-07-03T18:24:26Z

temporalio/nexus/_decorators.py

+import nexusrpc
+from nexusrpc import InputT, OutputT
+from nexusrpc.handler import (
+    OperationHandler,
+    StartOperationContext,
+)
+
+from temporalio.nexus._operation_context import (
+    WorkflowRunOperationContext,
+)
+from temporalio.nexus._operation_handlers import (
+    WorkflowRunOperationHandler,
+)
+from temporalio.nexus._token import (
+    WorkflowHandle,
+)
+from temporalio.nexus._util import (
+    get_callable_name,
+    get_workflow_run_start_method_input_and_output_type_annotations,
+    set_operation_factory,
+)


With most other Python code in the SDK, we chose to fully qualify things where they are used if they are from a non-stdlib module. Not that we have to, but I think it can make it easier to reason about when reading, prevents some ambiguity, and makes code consistent across the codebase.

cretz · 2025-07-03T18:25:41Z

temporalio/nexus/_operation_context.py

+):
+    ctx = _try_temporal_context()
+    if ctx is None:
+        raise RuntimeError("Not in Nexus operation context.")


cretz · 2025-07-03T18:29:24Z

temporalio/nexus/_operation_context.py

+
+
+@dataclass(frozen=True)
+class WorkflowRunOperationContext(StartOperationContext):


What I mean by "accessing a Nexus context should be done the same way regardless of whether it's Temporal-backed" is that you access the Nexus context the same way, regardless of whether the handler runs in a Temporal worker or some other worker. That is unrelated to the Temporal context IMO. Regardless, if you must mash the Nexus context with the Temporal context for workflow run, does the Temporal context need to be optional?

cretz · 2025-07-03T18:31:36Z

temporalio/nexus/_operation_context.py

+        return workflow_handle
+
+
+@dataclass(frozen=True)


The problem here is that we are mixing public and private attributes, so to users it looks like a normal dataclass they can instantiate, but it's not. At the least, we should document that users should never instantiate this, though we need some kind of Nexus test harness probably.

cretz · 2025-07-03T19:35:45Z

temporalio/worker/_workflow_instance.py

+            self, self._next_seq("nexus_operation"), input, operation_handle_fn()
+        )
+        handle._apply_schedule_command()
+        self._pending_nexus_operations[handle._seq] = handle


When do you delete from this dictionary?

cretz · 2025-07-03T19:38:08Z

temporalio/worker/_workflow_instance.py

+        # TODO(nexus-preview): How should this behave?
+        # Java has a separate class that only exists if the operation token exists:
+        # https://github.com/temporalio/sdk-java/blob/master/temporal-sdk/src/main/java/io/temporal/internal/sync/NexusOperationExecutionImpl.java#L26
+        # And Go similar:
+        # https://github.com/temporalio/sdk-go/blob/master/internal/workflow.go#L2770-L2771


I don't understand the question exactly here

cretz · 2025-07-03T19:40:31Z

temporalio/worker/_workflow_instance.py

+        v.service = self._input.service
+        v.operation = self._input.operation_name
+        v.input.CopyFrom(
+            self._instance._payload_converter.to_payload(self._input.input)


Can we do payload conversion before we add the command? We need fallible things to fail before we call the side-effecting add_command, because otherwise the command will be half initialized

cretz · 2025-07-03T19:41:31Z

temporalio/worker/_workflow_instance.py

+    async def result(self) -> OutputT:
+        return await self._task


What is the use case for this call? It does not seem to be in the base class

cretz · 2025-07-03T19:42:40Z

tests/nexus/test_handler.py

+    await _test_start_operation(test_case, True, env)
+
+
+async def _test_start_operation(


Otherwise I'd have to create a caller workflow and wouldn't have access to HTTP responses

This is what we ask our users to do. We should test as our users might, it helps us find things. Is what we're asking our users to do too onerous for us to also do? (understood it may not make this PR)

This reverts commit 38a50fa.

bergundy

Submitting what I have so far since I saw changes were pushed.

bergundy · 2025-07-03T20:52:21Z

pyproject.toml

@@ -208,3 +216,6 @@ exclude = [
 [tool.uv]
 # Prevent uv commands from building the package by default
 package = false
+
+[tool.uv.sources]
+nexus-rpc = { git = "https://github.com/nexus-rpc/sdk-python" }


This should be removed before merging, right? We need a release of the Nexus SDK to pypi.

bergundy · 2025-07-03T20:55:11Z

temporalio/client.py

+        # The following options should not be considered part of the public API. They
+        # are deliberately not exposed in overloads, and are not subject to any
+        # backwards compatibility guarantees.
+        nexus_completion_callbacks: Sequence[NexusCompletionCallback] = [],


Nexus callbacks are only one type of callback that we were planning on supporting. I wouldn't mention nexus in the name here.

bergundy · 2025-07-03T20:56:33Z

temporalio/client.py

+            temporalio.api.common.v1.Link.WorkflowEvent
+        ] = [],
+        request_id: Optional[str] = None,
+        stack_level: int = 2,


You are also missing the conflict options which are required for USE_EXISTING to make sense for workflow callers. Can be done in a followup PR since this was functionality that was added as a second step for Go and Java.

I would for now throw if a user tries to set USE_EXISTING in the workflow run operation handler.

bergundy · 2025-07-03T20:57:52Z

temporalio/client.py

+        if input.request_id:
+            req.request_id = input.request_id
+
+        req.completion_callbacks.extend(


Put the links on the callbacks so they can be associated.

bergundy · 2025-07-03T20:59:19Z

temporalio/client.py

@@ -7231,6 +7260,17 @@ def api_key(self, value: Optional[str]) -> None:
        self.service_client.update_api_key(value)


+@dataclass(frozen=True)
+class NexusCompletionCallback:


Don't make callbacks Nexus specific and don't call them Completion callbacks since we will want to use these callbacks for other purposes too.

bergundy · 2025-07-03T21:54:33Z

temporalio/worker/_nexus.py

+                completion = temporalio.bridge.proto.nexus.NexusTaskCompletion(
+                    task_token=task.task.task_token
+                )
+                completion.error.failure.message = "Worker shutting down"


If we do decide to keep this code, it should be a (retryable) INTERNAL handler error.

bergundy · 2025-07-03T22:25:09Z

temporalio/worker/_nexus.py

+                ),
+            )
+        else:
+            # TODO(nexus-preview): ack_cancel completions?


We discussed this on slack, there's no reason to ack the cancels apart from testing Core.

bergundy · 2025-07-03T23:11:20Z

temporalio/worker/_nexus.py

+            await self._bridge_worker().complete_nexus_task(completion)
+        except Exception:
+            logger.exception("Failed to send Nexus task completion")
+


Suggested change

finally:

try:

del self._running_tasks[task_token]

except KeyError:

logger.exception("Failed to remove task for completed Nexus operation cancel handler")

bergundy · 2025-07-03T23:12:06Z

temporalio/worker/_nexus.py

+            try:
+                del self._running_tasks[task_token]
+            except KeyError:
+                logger.exception("Failed to remove completed Nexus operation")


Suggested change

logger.exception("Failed to remove completed Nexus operation")

logger.exception("Failed to remove task for completed Nexus operation start handler")

bergundy · 2025-07-03T23:13:53Z

temporalio/worker/_nexus.py

+            await self._bridge_worker().complete_nexus_task(completion)
+        except Exception:
+            logger.exception("Failed to send Nexus task completion")
+        finally:


Might be worth putting this try finally block around the entire function to ensure we can avoid leaks.

bergundy · 2025-07-03T23:20:05Z

temporalio/converter.py

+                logger.warning(
+                    f"Unknown Nexus HandlerErrorType: {nexus_handler_failure_info.type}"
+                )
+                _type = nexusrpc.HandlerErrorType.INTERNAL


In other SDKs we preserve the original string type for forwards compatibility.

bergundy · 2025-07-03T23:32:57Z

temporalio/worker/_nexus.py

+                if err.retryable
+                else temporalio.api.enums.v1.NexusHandlerErrorRetryBehavior.NEXUS_HANDLER_ERROR_RETRY_BEHAVIOR_NON_RETRYABLE


This is wrong, retryable can be None in which case it would translate to unspecified.

bergundy · 2025-07-03T23:42:03Z

temporalio/worker/_nexus.py

+    elif isinstance(err, ApplicationError):
+        handler_err = nexusrpc.HandlerError(
+            # TODO(nexus-prerelease): what should message be?
+            err.message,


Yeah, we will figure this out. I will try to find time next week to spec it out.

bergundy · 2025-07-03T23:55:03Z

temporalio/worker/_workflow_instance.py

+        handle._apply_schedule_command()
+        self._pending_nexus_operations[handle._seq] = handle
+
+        while True:


I find this loop confusing, can you document what this is trying to do?

Will do. For reassurance it's essentially identical to the existing implementation of ChildWorkflow: https://github.com/temporalio/sdk-python/blob/main/temporalio/worker/_workflow_instance.py#L1715-L1724

bergundy · 2025-07-03T23:58:20Z

temporalio/workflow.py

+
+
+def create_nexus_client(
+    endpoint: str, service: Union[Type[ServiceT], str]


I would consider not making endpoint a positional so we can add a client that targets a task queue in the future.

bergundy · 2025-07-04T00:03:56Z

tests/nexus/test_workflow_run_operation.py

+        return input
+
+
+class MyOperation(WorkflowRunOperationHandler):


Why do you need this class? You can use the decorator, right?

bergundy

Approved with comments. I didn't get to look at the tests but the rest mostly looks good to me. We should revisit error rehydration and some of the additions that were made to Go and Java that weren't added here.

dandavison force-pushed the nexus branch 2 times, most recently from a845a6f to 7a121bb Compare April 5, 2025 15:08

dandavison force-pushed the nexus branch 10 times, most recently from 862e4f9 to 8ac0192 Compare April 17, 2025 12:17

dandavison force-pushed the nexus branch 4 times, most recently from 8940d51 to 520aecf Compare April 26, 2025 16:16

dandavison force-pushed the nexus branch 2 times, most recently from 2fc971f to 8bd6011 Compare April 29, 2025 13:11

dandavison force-pushed the nexus branch from 18e268a to 2eb967a Compare May 6, 2025 15:55

dandavison changed the title ~~Nexus prototype~~ Nexus May 8, 2025

dandavison force-pushed the nexus branch from 7876ecc to 7fe6868 Compare May 8, 2025 22:07

dandavison force-pushed the nexus branch 2 times, most recently from 5d54f48 to 8ddfb94 Compare May 24, 2025 01:49

dandavison force-pushed the nexus branch 7 times, most recently from fa3e8ec to f7bf47b Compare May 27, 2025 20:34

dandavison added 15 commits July 3, 2025 08:02

Cleanup error test

0fb7837

Revert "Delete redundant test"

7656a02

This reverts commit 86a9a61.

Evolve context API

075ec7c

Rename as temporalio.nexus.cancel_workflow

6c1e894

Fix test

d5fce2a

Cleanup

8f3681b

Impprovements from code review comments

8117ea1

Expose nexus.LoggerAdapter

7b62955

Add outbound links for sync responses also

4e34462

Don't expose separate workflow.start_nexus_operation

8889dde

Remove unnecessary type hint

41b9709

New Nexus client constructor

c2d4825

Remove unused test helper methods

a4fd205

Clean up token type

a5f67d2

Refactor start timeout test

f7f8a51

cretz reviewed Jul 3, 2025

View reviewed changes

dandavison added 4 commits July 3, 2025 15:59

Cancellation timeout test

01960ed

Create running_task for cancellation op handler

5103f80

Test creation of worker from ServiceHandler instances

03513b5

Test creation of worker from programmatically-created ServiceHandler

b6bcd6c

dandavison force-pushed the nexus branch from f6b2c34 to b6bcd6c Compare July 3, 2025 19:59

Reapply "Convert nexus_handler_failure_info as nexusrpc.HandlerError"

a840316

This reverts commit 38a50fa.

bergundy reviewed Jul 3, 2025

View reviewed changes

bergundy reviewed Jul 4, 2025

View reviewed changes

bergundy approved these changes Jul 4, 2025

View reviewed changes



		@dataclass(frozen=True)
		class WorkflowRunOperationContext(StartOperationContext):

		await _test_start_operation(test_case, True, env)


		async def _test_start_operation(

	logger.exception("Failed to remove completed Nexus operation")
	logger.exception("Failed to remove task for completed Nexus operation start handler")

		if err.retryable
		else temporalio.api.enums.v1.NexusHandlerErrorRetryBehavior.NEXUS_HANDLER_ERROR_RETRY_BEHAVIOR_NON_RETRYABLE



		def create_nexus_client(
		endpoint: str, service: Union[Type[ServiceT], str]

Nexus: worker, workflow-backed operations, and workflow caller #813

Are you sure you want to change the base?

Nexus: worker, workflow-backed operations, and workflow caller #813

Conversation

dandavison commented Apr 2, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Notes for reviewers

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

bergundy left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

bergundy left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

dandavison commented Apr 2, 2025 •

edited

Loading