Add Nexus Signal-with-Start example

nexus_messaging/callerpattern/README.md

@@ -17,19 +17,23 @@ The caller Workflow:
 
 ### Running
 
-Start a Temporal server:
+This sample requires a Temporal dev server build that supports Workflow Update callbacks. Download the compatible
+binary from the [Temporal CLI pre-release instructions](https://docs.temporal.io/standalone-nexus-operation#temporal-cli-support).
+
+Start the Temporal dev server with the required namespaces pre-created and Workflow Update callbacks enabled:
 
 ```bash
-temporal server start-dev
+./temporal server start-dev \
+  --dynamic-config-value history.enableUpdateCallbacks=true \
+  --dynamic-config-value history.enableCHASMSignalBacklinks=true \
+  --namespace nexus-messaging-handler-namespace \
+  --namespace nexus-messaging-caller-namespace
 ```
 
-Create the namespaces and Nexus endpoint:
+Create the Nexus endpoint:
 
 ```bash
-temporal operator namespace create --namespace nexus-messaging-handler-namespace
-temporal operator namespace create --namespace nexus-messaging-caller-namespace
-
-temporal operator nexus endpoint create \
+./temporal operator nexus endpoint create \
   --name nexus-messaging-nexus-endpoint \
   --target-namespace nexus-messaging-handler-namespace \
   --target-task-queue nexus-messaging-handler-task-queue

nexus_messaging/callerpattern/handler/service_handler.py

@@ -1,7 +1,8 @@
 """
 Nexus operation handler implementation for the entity pattern. Each operation receives a
-user_id, which is mapped to a workflow ID. The operations are synchronous because queries
-and updates against a running workflow complete quickly.
+user_id, which is mapped to a Workflow ID. The Query and Signal operations are synchronous
+because they complete quickly against a running Workflow; set_language is async because it
+is backed by a Workflow Update that may call an activity.
 """
 
 from __future__ import annotations
@@ -77,13 +78,11 @@ async def set_language(
         client: nexus.TemporalNexusClient,
         input: SetLanguageInput,
     ) -> nexus.TemporalOperationResult[Language]:
-        result = await self._get_workflow_handle(
-            client.client, input.user_id
-        ).execute_update(
+        return await client.start_workflow_update(
+            get_workflow_id(input.user_id),
             GreetingWorkflow.set_language_using_activity,
             input,
         )
-        return nexus.TemporalOperationResult.sync(result)
 
     @nexus.temporal_operation
     async def approve(

nexus_messaging/ondemandpattern/README.md

@@ -6,28 +6,36 @@ operations. `NexusRemoteGreetingService` adds a `run_from_remote` operation that
 instance to target.
 
 The caller Workflow:
-1. Starts two remote `GreetingWorkflow` instances via `run_from_remote` (backed by `temporal_operation`)
-2. Queries each for supported languages
-3. Changes the language on each (Arabic and Hindi)
-4. Confirms the changes via queries
-5. Approves both Workflows
-6. Waits for each to complete and returns their results
+1. Attaches approval context for the first user via `attach_approval_context`, before anything has
+   started that user's Workflow
+2. Starts two remote `GreetingWorkflow` instances via `run_from_remote` (backed by `temporal_operation`)
+3. Attaches approval context for the second user, whose Workflow now already exists
+4. Queries each for supported languages
+5. Changes the language on each (Arabic and Hindi)
+6. Confirms the changes via queries
+7. Approves both Workflows
+8. Waits for each to complete and returns their results
 
 ### Running
 
-Start a Temporal server:
+This sample requires a Temporal dev server build that supports Workflow Update callbacks. Download the compatible
+binary from the [Temporal CLI pre-release instructions](https://docs.temporal.io/standalone-nexus-operation#temporal-cli-support).
+
+Start the Temporal dev server with the required namespaces pre-created and Workflow Update callbacks enabled:
 
 ```bash
-temporal server start-dev
+./temporal server start-dev \
+  --dynamic-config-value history.enableUpdateCallbacks=true \
+  --dynamic-config-value history.enableCHASMSignalBacklinks=true \
+  --dynamic-config-value history.enableSignalWithStartFromWorkflow=true \
+  --namespace nexus-messaging-handler-namespace \
+  --namespace nexus-messaging-caller-namespace
 ```
 
-Create the namespaces and Nexus endpoint:
+Create the Nexus endpoint:
 
 ```bash
-temporal operator namespace create --namespace nexus-messaging-handler-namespace
-temporal operator namespace create --namespace nexus-messaging-caller-namespace
-
-temporal operator nexus endpoint create \
+./temporal operator nexus endpoint create \
   --name nexus-messaging-nexus-endpoint \
   --target-namespace nexus-messaging-handler-namespace \
   --target-task-queue nexus-messaging-handler-task-queue
@@ -48,8 +56,10 @@ uv run python -m nexus_messaging.ondemandpattern.caller.app
 Expected output:
 
 ```
+Attached approval context before the workflow existed: UserId One
 started remote greeting workflow: UserId One
 started remote greeting workflow: UserId Two
+Attached approval context to the running workflow: UserId Two
 Supported languages for UserId One: [<Language.CHINESE: 2>, <Language.ENGLISH: 3>]
 Supported languages for UserId Two: [<Language.CHINESE: 2>, <Language.ENGLISH: 3>]
 UserId One changed language: ENGLISH -> ARABIC

nexus_messaging/ondemandpattern/caller/workflows.py

@@ -8,6 +8,7 @@
 
 from nexus_messaging.ondemandpattern.service import (
     ApproveInput,
+    AttachApprovalContextInput,
     GetLanguageInput,
     GetLanguagesInput,
     Language,
@@ -38,10 +39,33 @@ async def run(self) -> list[str]:
         # users we want to process. The first calls start two workflows, one for each
         # user. Subsequent calls perform different actions between the two users.
 
+        # Attach information before the Workflow exists. Because attach_approval_context
+        # is backed by Signal-with-Start on the handler, this call creates the Workflow
+        # and delivers the note to it.
+        await self.nexus_client.execute_operation(
+            NexusRemoteGreetingService.attach_approval_context,
+            AttachApprovalContextInput(
+                note="queued for localization review by the nightly batch",
+                user_id=REMOTE_WORKFLOW_ONE,
+            ),
+        )
+        log.append(
+            f"Attached approval context before the workflow existed: "
+            f"{REMOTE_WORKFLOW_ONE}"
+        )
+        workflow.logger.info(
+            "attached approval context for %s, creating the workflow",
+            REMOTE_WORKFLOW_ONE,
+        )
+
         # This is an async Nexus operation -- starts a workflow on the handler and
         # returns a handle. Unlike the sync operations below, this does not block
         # until the workflow completes. It is backed by temporal_operation on the
         # handler side.
+
+        # The Workflow for this user is already running due to the call above. The
+        # handler sets the conflict policy to USE_EXISTING, so this call attaches the
+        # operation's completion callback to the running execution.
         handle_one = await self.nexus_client.start_operation(
             NexusRemoteGreetingService.run_from_remote,
             RunFromRemoteInput(user_id=REMOTE_WORKFLOW_ONE),
@@ -56,6 +80,23 @@ async def run(self) -> list[str]:
         log.append(f"started remote greeting workflow: {REMOTE_WORKFLOW_TWO}")
         workflow.logger.info("started remote greeting workflow %s", REMOTE_WORKFLOW_TWO)
 
+        # This user's Workflow was created by run_from_remote just above, so here
+        # Signal-with-Start skips the start and only delivers the Signal.
+        await self.nexus_client.execute_operation(
+            NexusRemoteGreetingService.attach_approval_context,
+            AttachApprovalContextInput(
+                note="translation approved by the localization team",
+                user_id=REMOTE_WORKFLOW_TWO,
+            ),
+        )
+        log.append(
+            f"Attached approval context to the running workflow: {REMOTE_WORKFLOW_TWO}"
+        )
+        workflow.logger.info(
+            "attached approval context for %s, messaging the existing workflow",
+            REMOTE_WORKFLOW_TWO,
+        )
+
         # Query the remote workflows for supported languages.
         languages_output = await self.nexus_client.execute_operation(
             NexusRemoteGreetingService.get_languages,

nexus_messaging/ondemandpattern/handler/service_handler.py

@@ -10,11 +10,16 @@
 import nexusrpc
 from temporalio import nexus
 from temporalio.client import Client, WorkflowHandle
+from temporalio.common import WorkflowIDConflictPolicy
 
-from nexus_messaging.ondemandpattern.handler.workflows import GreetingWorkflow
+from nexus_messaging.ondemandpattern.handler.workflows import (
+    ATTACH_APPROVAL_CONTEXT_SIGNAL,
+    GreetingWorkflow,
+)
 from nexus_messaging.ondemandpattern.service import (
     ApproveInput,
     ApproveOutput,
+    AttachApprovalContextInput,
     GetLanguageInput,
     GetLanguagesInput,
     GetLanguagesOutput,
@@ -39,7 +44,8 @@ def _get_workflow_handle(
             GreetingWorkflow.run, self._get_workflow_id(user_id)
         )
 
-    # Starts a new GreetingWorkflow with the caller-specified user ID.
+    # Starts a new GreetingWorkflow with the caller-specified user ID,
+    # or attached to one already running.
     # This is an async Nexus operation backed by temporal_operation.
     @nexus.temporal_operation
     async def run_from_remote(
@@ -51,6 +57,10 @@ async def run_from_remote(
         return await client.start_workflow(
             GreetingWorkflow.run,
             id=self._get_workflow_id(input.user_id),
+            # Since attach_approval_context can create the GreetingWorkflow first,
+            # this operation needs to attach to the running execution rather than
+            # fail (default behavior).
+            id_conflict_policy=WorkflowIDConflictPolicy.USE_EXISTING,
         )
 
     @nexus.temporal_operation
@@ -86,13 +96,11 @@ async def set_language(
         client: nexus.TemporalNexusClient,
         input: SetLanguageInput,
     ) -> nexus.TemporalOperationResult[Language]:
-        result = await self._get_workflow_handle(
-            client.client, input.user_id
-        ).execute_update(
+        return await client.start_workflow_update(
+            self._get_workflow_id(input.user_id),
             GreetingWorkflow.set_language_using_activity,
             input,
         )
-        return nexus.TemporalOperationResult.sync(result)
 
     @nexus.temporal_operation
     async def approve(
@@ -105,3 +113,20 @@ async def approve(
             GreetingWorkflow.approve, input
         )
         return nexus.TemporalOperationResult.sync(ApproveOutput())
+
+    # Signals a Workflow, starting the Workflow first if it is not already running.
+    @nexus.temporal_operation
+    async def attach_approval_context(
+        self,
+        _ctx: nexus.TemporalStartOperationContext,
+        client: nexus.TemporalNexusClient,
+        input: AttachApprovalContextInput,
+    ) -> nexus.TemporalOperationResult[None]:
+        await client.client.start_workflow(
+            GreetingWorkflow.run,
+            id=self._get_workflow_id(input.user_id),
+            task_queue=nexus.info().task_queue,
+            start_signal=ATTACH_APPROVAL_CONTEXT_SIGNAL,
+            start_signal_args=[input],
+        )
+        return nexus.TemporalOperationResult.sync(None)

nexus_messaging/ondemandpattern/handler/workflows.py

@@ -7,19 +7,23 @@
 
 import asyncio
 from datetime import timedelta
+from typing import Optional
 
 from temporalio import workflow
 from temporalio.exceptions import ApplicationError
 
 from nexus_messaging.ondemandpattern.handler.activities import call_greeting_service
 from nexus_messaging.ondemandpattern.service import (
     ApproveInput,
+    AttachApprovalContextInput,
     GetLanguagesInput,
     GetLanguagesOutput,
     Language,
     SetLanguageInput,
 )
 
+ATTACH_APPROVAL_CONTEXT_SIGNAL = "attach_approval_context"
+
 
 @workflow.defn
 class GreetingWorkflow:
@@ -30,6 +34,7 @@ def __init__(self) -> None:
             Language.ENGLISH: "Hello, world",
         }
         self.language = Language.ENGLISH
+        self.approval_context: Optional[str] = None
         self.lock = asyncio.Lock()
 
     @workflow.run
@@ -54,9 +59,24 @@ def get_language(self) -> Language:
 
     @workflow.signal
     def approve(self, input: ApproveInput) -> None:
-        workflow.logger.info("Approval signal received for user %s", input.user_id)
+        workflow.logger.info(
+            "Approval signal received for user %s (context: %s)",
+            input.user_id,
+            self.approval_context,
+        )
         self.approved_for_release = True
 
+    # Attaches supporting information for the eventual approval. Delivered with
+    # Signal-with-Start, so this may be the message that creates the workflow.
+    @workflow.signal(name=ATTACH_APPROVAL_CONTEXT_SIGNAL)
+    def attach_approval_context(self, input: AttachApprovalContextInput) -> None:
+        workflow.logger.info(
+            "attach_approval_context signal received for user %s: %s",
+            input.user_id,
+            input.note,
+        )
+        self.approval_context = input.note
+
     @workflow.update
     def set_language(self, input: SetLanguageInput) -> Language:
         workflow.logger.info("setLanguage update received for user %s", input.user_id)

nexus_messaging/ondemandpattern/service.py

@@ -58,6 +58,12 @@ class ApproveOutput:
     pass
 
 
+@dataclass
+class AttachApprovalContextInput:
+    note: str
+    user_id: str
+
+
 @nexusrpc.service
 class NexusRemoteGreetingService:
     # Starts a new GreetingWorkflow with the given workflow ID (asynchronous).
@@ -70,3 +76,6 @@ class NexusRemoteGreetingService:
     set_language: nexusrpc.Operation[SetLanguageInput, Language]
     # Approves the specified workflow, allowing it to complete.
     approve: nexusrpc.Operation[ApproveInput, ApproveOutput]
+    # Attaches supporting information for the Workflow approval, either by
+    # messaging a running Workflow or creating a Workflow.
+    attach_approval_context: nexusrpc.Operation[AttachApprovalContextInput, None]

tests/conftest.py

@@ -45,6 +45,12 @@ async def env(request) -> AsyncGenerator[WorkflowEnvironment, None]:
             dev_server_extra_args=[
                 "--dynamic-config-value",
                 "activity.enableCallbacks=true",
+                "--dynamic-config-value",
+                "history.enableUpdateCallbacks=true",
+                "--dynamic-config-value",
+                "history.enableCHASMSignalBacklinks=true",
+                "--dynamic-config-value",
+                "history.enableSignalWithStartFromWorkflow=true",
             ],
             dev_server_download_version="v1.7.4-standalone-nexus-operations",
         )

tests/nexus_messaging/ondemandpattern_test.py

@@ -12,6 +12,7 @@
 from nexus_messaging.ondemandpattern.caller.workflows import CallerRemoteWorkflow
 from nexus_messaging.ondemandpattern.service import (
     ApproveInput,
+    AttachApprovalContextInput,
     GetLanguageInput,
     GetLanguagesInput,
     Language,
@@ -37,12 +38,27 @@ async def run(self) -> None:
 
         workflow_id = f"test-remote-{workflow.uuid4()}"
 
-        # Start a remote workflow.
+        # Signal-with-Start: no Workflow exists for this user yet, so this operation
+        # creates it and delivers the Signal.
+        await nexus_client.execute_operation(
+            NexusRemoteGreetingService.attach_approval_context,
+            AttachApprovalContextInput(note="created by signal", user_id=workflow_id),
+        )
+
+        # Start a remote Workflow. The Signal-with-Start above already created it, so
+        # this attaches to the running execution (USE_EXISTING conflict policy).
         handle = await nexus_client.start_operation(
             NexusRemoteGreetingService.run_from_remote,
             RunFromRemoteInput(user_id=workflow_id),
         )
 
+        # Signal-with-Start again, this time against the already-running Workflow, so
+        # only the Signal is delivered.
+        await nexus_client.execute_operation(
+            NexusRemoteGreetingService.attach_approval_context,
+            AttachApprovalContextInput(note="signal only", user_id=workflow_id),
+        )
+
         # Query for supported languages.
         languages_output = await nexus_client.execute_operation(
             NexusRemoteGreetingService.get_languages,