feat: signal_workflow client method (#45)

<!-- Describe what has changed in this PR -->
**What changed?**

Implement client method to signal_workflow

<!-- Tell your future self why have you made these changes -->
**Why?**

It's a core feature that enables signals in workflow


<!-- How have you verified this change? Tested locally? Added a unit
test? Checked in staging env? -->
**How did you test it?**

Integration test


<!-- Assuming the worst case, what can be broken when deploying this
change to production? -->
**Potential risks**

<!-- Is it notable for release? e.g. schema updates, configuration or
data migration required? If so, please mention it, and also update
CHANGELOG.md -->
**Release notes**

<!-- Is there any documentation updates should be made for config,
https://cadenceworkflow.io/docs/operation-guide/setup/ ? If so, please
open an PR in https://github.com/cadence-workflow/cadence-docs -->
**Documentation Changes**

---------

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

Author: timl3136

cadence/client.py

@@ -15,6 +15,7 @@
 from grpc.aio import Channel, ClientInterceptor, secure_channel, insecure_channel
 from cadence.api.v1.service_workflow_pb2_grpc import WorkflowAPIStub
 from cadence.api.v1.service_workflow_pb2 import (
+    SignalWorkflowExecutionRequest,
     StartWorkflowExecutionRequest,
     StartWorkflowExecutionResponse,
     SignalWithStartWorkflowExecutionRequest,
@@ -232,6 +233,51 @@ async def start_workflow(
         except Exception:
             raise
 
+    async def signal_workflow(
+        self,
+        workflow_id: str,
+        run_id: str,
+        signal_name: str,
+        *signal_args: Any,
+    ) -> None:
+        """
+        Send a signal to a running workflow execution.
+
+        Args:
+            workflow_id: The workflow ID
+            run_id: The run ID (can be empty string to signal current run)
+            signal_name: Name of the signal
+            *signal_args: Arguments to pass to the signal handler
+
+        Raises:
+            ValueError: If signal encoding fails
+            Exception: If the gRPC call fails
+        """
+        signal_payload = None
+        if signal_args:
+            try:
+                signal_payload = self.data_converter.to_data(list[Any](signal_args))
+            except Exception as e:
+                raise ValueError(f"Failed to encode signal input: {e}")
+
+        workflow_execution = WorkflowExecution()
+        workflow_execution.workflow_id = workflow_id
+        if run_id:
+            workflow_execution.run_id = run_id
+
+        signal_request = SignalWorkflowExecutionRequest(
+            domain=self.domain,
+            workflow_execution=workflow_execution,
+            identity=self.identity,
+            request_id=str(uuid.uuid4()),
+            signal_name=signal_name,
+        )
+
+        if signal_payload:
+            signal_request.signal_input.CopyFrom(signal_payload)
+
+        await self.workflow_stub.SignalWorkflowExecution(signal_request)
+
     async def signal_with_start_workflow(
         self,
         workflow: Union[str, WorkflowDefinition],

tests/integration_tests/test_client.py

@@ -140,6 +140,61 @@ async def test_workflow_stub_start_and_describe(helper: CadenceHelper):
         )
 
 
+@pytest.mark.usefixtures("helper")
+async def test_signal_workflow(helper: CadenceHelper):
+    """Test signal_workflow method.
+
+    This integration test verifies:
+    1. Starting a workflow execution
+    2. Sending a signal to the running workflow
+    3. Signal appears in the workflow's history
+    """
+    async with helper.client() as client:
+        workflow_type = "test-workflow-signal"
+        task_list_name = "test-task-list-signal"
+        workflow_id = "test-workflow-signal-789"
+        execution_timeout = timedelta(minutes=5)
+        signal_name = "test-signal"
+        signal_arg = {"action": "update", "value": 42}
+
+        execution = await client.start_workflow(
+            workflow_type,
+            task_list=task_list_name,
+            execution_start_to_close_timeout=execution_timeout,
+            workflow_id=workflow_id,
+        )
+
+        await client.signal_workflow(
+            execution.workflow_id,
+            execution.run_id,
+            signal_name,
+            signal_arg,
+        )
+
+        # Fetch workflow history to verify signal was recorded
+        history_response = await client.workflow_stub.GetWorkflowExecutionHistory(
+            GetWorkflowExecutionHistoryRequest(
+                domain=DOMAIN_NAME,
+                workflow_execution=execution,
+                skip_archival=True,
+            )
+        )
+
+        # Verify signal event appears in history
+        signal_events = [
+            event
+            for event in history_response.history.events
+            if event.HasField("workflow_execution_signaled_event_attributes")
+        ]
+
+        assert len(signal_events) == 1, "Expected exactly one signal event in history"
+        signal_event = signal_events[0]
+        assert (
+            signal_event.workflow_execution_signaled_event_attributes.signal_name
+            == signal_name
+        ), f"Expected signal name '{signal_name}'"
+
+
 @pytest.mark.usefixtures("helper")
 async def test_signal_with_start_workflow(helper: CadenceHelper):
     """Test signal_with_start_workflow method.