temp draft for signal handling

Signed-off-by: Tim Li <ltim@uber.com>

Author: timl3136

cadence/_internal/workflow/workflow_engine.py

@@ -184,9 +184,11 @@ def _process_decision_events(
                         "replay_mode": decision_events.replay,
                     },
                 )
+                event_attr_name = event.WhichOneof("attributes")
+
                 # start workflow on workflow started event
                 if (
-                    event.WhichOneof("attributes")
+                    event_attr_name
                     == "workflow_execution_started_event_attributes"
                 ):
                     started_attrs: WorkflowExecutionStartedEventAttributes = (
@@ -195,6 +197,20 @@ def _process_decision_events(
                     if started_attrs and hasattr(started_attrs, "input"):
                         self._workflow_instance.start(started_attrs.input)
 
+                # invoke signal handler on signal event
+                elif (
+                    event_attr_name
+                    == "workflow_execution_signaled_event_attributes"
+                ):
+                    signaled_attrs = (
+                        event.workflow_execution_signaled_event_attributes
+                    )
+                    self._workflow_instance.handle_signal(
+                        signaled_attrs.signal_name,
+                        signaled_attrs.input,
+                        event.event_id,
+                    )
+
                 # Process through state machines (DecisionsHelper now delegates to DecisionManager)
                 self._decision_manager.handle_history_event(event)
 

cadence/_internal/workflow/workflow_intance.py

@@ -1,11 +1,15 @@
+import logging
 from asyncio import CancelledError, InvalidStateError, Task
-from typing import Optional
+from typing import Optional, Type
+
 from cadence._internal.workflow.deterministic_event_loop import DeterministicEventLoop
 from cadence.api.v1.common_pb2 import Payload
 from cadence.data_converter import DataConverter
 from cadence.error import WorkflowFailure
 from cadence.workflow import WorkflowDefinition
 
+logger = logging.getLogger(__name__)
+
 
 class WorkflowInstance:
     def __init__(
@@ -29,6 +33,72 @@ def start(self, payload: Payload):
             )
             self._task = self._loop.create_task(run_method(*workflow_input))
 
+    def handle_signal(
+        self, signal_name: str, payload: Payload, event_id: int
+    ) -> None:
+        """Handle an incoming signal by invoking the registered signal handler.
+
+        Looks up the signal definition by name, decodes the payload using the
+        data converter and parameter type hints, and invokes the handler on the
+        workflow instance. Async handlers are scheduled as tasks on the
+        deterministic event loop so they execute during run_once().
+
+        Args:
+            signal_name: The name of the signal to handle.
+            payload: The encoded signal input payload.
+            event_id: The history event ID (used for logging context).
+        """
+        # Guard: reject signals after workflow completion (matches Java client)
+        if self.is_done():
+            logger.warning(
+                "Signal received after workflow is completed, ignoring",
+                extra={"signal_name": signal_name, "event_id": event_id},
+            )
+            return
+
+        signal_def = self._definition.signals.get(signal_name)
+        if signal_def is None:
+            logger.warning(
+                "Received signal with no registered handler, ignoring",
+                extra={"signal_name": signal_name, "event_id": event_id},
+            )
+            return
+
+        # Decode payload using parameter type hints from the signal definition.
+        # Deserialization errors are caught and logged rather than crashing the
+        # decision (matches Java client DataConverterException handling).
+        type_hints: list[Type | None] = [p.type_hint for p in signal_def.params]
+        try:
+            if type_hints:
+                decoded_args = self._data_converter.from_data(payload, type_hints)
+            else:
+                decoded_args = []
+        except Exception:
+            logger.error(
+                "Failed to deserialize signal payload, dropping signal",
+                extra={"signal_name": signal_name, "event_id": event_id},
+                exc_info=True,
+            )
+            return
+
+        # Invoke the handler on the workflow instance.
+        # signal_def._wrapped is the unbound class method, so we pass
+        # self._instance as the first argument.
+        # Handler invocation errors are caught and logged rather than crashing
+        # the decision task (matches Java client InvocationTargetException handling).
+        try:
+            if signal_def.is_async:
+                coro = signal_def(self._instance, *decoded_args)
+                self._loop.create_task(coro)
+            else:
+                signal_def(self._instance, *decoded_args)
+        except Exception:
+            logger.error(
+                "Signal handler raised an exception",
+                extra={"signal_name": signal_name, "event_id": event_id},
+                exc_info=True,
+            )
+
     def run_once(self):
         self._loop.run_until_yield()