Add @workflow.signal decorator and in-workflow signal dispatch

Addresses zorporation/durable-workflow#390 (signal portion). Workflows
can now declare @workflow.signal("name") handler methods; the replayer
scans for SignalReceived history events, decodes the envelope, and
dispatches to the handler in history order — interleaved with activity
completions so handler-mutated state is visible to subsequent yields.

Queries and updates are scoped out of this change and tracked as
follow-up issues. Queries need a worker-side query-execution endpoint
since the server cannot replay Python workflows; updates need
UpdateAccepted/UpdateApplied history event processing plus a validator
contract.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: durable-workflow-ops

src/durable_workflow/workflow.py

@@ -34,16 +34,59 @@
 
 
 def defn(*, name: str):  # type: ignore[no-untyped-def]
-    """Register a class as a workflow type under a language-neutral name."""
+    """Register a class as a workflow type under a language-neutral name.
+
+    Scans the class for ``@signal``-decorated methods and builds a signal
+    registry at decoration time so the replayer can dispatch incoming
+    signals without re-inspecting the class on every history event.
+    """
 
     def wrap(cls: type) -> type:
         cls.__workflow_name__ = name  # type: ignore[attr-defined]
+        signals: dict[str, str] = {}
+        for attr in dir(cls):
+            if attr.startswith("_"):
+                continue
+            member = getattr(cls, attr, None)
+            signal_name = getattr(member, "__signal_name__", None)
+            if isinstance(signal_name, str) and signal_name:
+                signals[signal_name] = attr
+        cls.__workflow_signals__ = signals  # type: ignore[attr-defined]
         _REGISTRY[name] = cls
         return cls
 
     return wrap
 
 
+def signal(name: str) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
+    """Mark a workflow method as the handler for an external signal.
+
+    Example::
+
+        @workflow.defn(name="approval")
+        class ApprovalWorkflow:
+            def __init__(self) -> None:
+                self.approved: bool = False
+
+            @workflow.signal("approve")
+            def on_approve(self, by: str) -> None:
+                self.approved = True
+
+    The decorated method is called by the replayer when a matching
+    ``SignalReceived`` history event is observed, with the signal's
+    decoded arguments unpacked into positional parameters. Handler return
+    values are ignored; to expose state back to the workflow's main run
+    loop, mutate ``self.*`` attributes (as ``on_approve`` does above) and
+    yield the usual commands from ``run()``.
+    """
+
+    def wrap(method: Callable[..., Any]) -> Callable[..., Any]:
+        method.__signal_name__ = name  # type: ignore[attr-defined]
+        return method
+
+    return wrap
+
+
 def registry() -> dict[str, type]:
     """Return a copy of workflow types registered in this process."""
     return dict(_REGISTRY)
@@ -428,6 +471,21 @@ def _decode_history_result(payload: dict[str, Any], fallback_codec: str | None)
     return serializer.decode_envelope(payload.get("result"), codec=codec)
 
 
+def _decode_signal_args(payload: dict[str, Any], fallback_codec: str | None) -> list[Any]:
+    codec = payload.get("payload_codec") or fallback_codec
+    raw = payload.get("value")
+    if raw is None:
+        raw = payload.get("input")
+    if raw is None:
+        raw = payload.get("arguments")
+    if raw is None:
+        return []
+    decoded = serializer.decode_envelope(raw, codec=codec)
+    if isinstance(decoded, list):
+        return decoded
+    return [decoded]
+
+
 def replay(
     workflow_cls: type,
     history_events: Iterable[dict[str, Any]],
@@ -452,6 +510,10 @@ def replay(
     ctx = WorkflowContext(run_id=run_id, current_time=workflow_start_time)
 
     resolved_results: list[Any] = []
+    # (resolved_result_index_before_apply, signal_name, decoded_args) —
+    # signals apply before the generator consumes the resolved_result at the
+    # stored index, which preserves history interleaving with activities.
+    pending_signals: list[tuple[int, str, list[Any]]] = []
     for ev in events:
         etype = ev.get("event_type")
         payload = ev.get("payload") or {}
@@ -472,6 +534,26 @@ def replay(
             resolved_results.append(payload.get("version", 0))
         elif etype in ("SearchAttributesUpserted", "search_attributes_upserted"):
             resolved_results.append(None)
+        elif etype in ("SignalReceived", "signal_received"):
+            signal_name = payload.get("signal_name")
+            if isinstance(signal_name, str) and signal_name:
+                pending_signals.append(
+                    (len(resolved_results), signal_name, _decode_signal_args(payload, payload_codec))
+                )
+
+    signal_registry: dict[str, str] = getattr(workflow_cls, "__workflow_signals__", {}) or {}
+
+    def _apply_due_signals() -> None:
+        while pending_signals and pending_signals[0][0] <= result_cursor:
+            _, name, args = pending_signals.pop(0)
+            method_name = signal_registry.get(name)
+            if method_name is None:
+                continue
+            handler = getattr(instance, method_name, None)
+            if handler is None:
+                continue
+            ctx.logger._set_replaying(True)
+            handler(*args)
 
     gen = instance.run(ctx, *start_input)
     if not hasattr(gen, "__next__"):
@@ -488,6 +570,7 @@ def replay(
     advanced_cmd: Any = None
     try:
         while True:
+            _apply_due_signals()
             if advanced_cmd is not None:
                 cmd = advanced_cmd
                 advanced_cmd = None

tests/test_signals.py

@@ -0,0 +1,158 @@
+from __future__ import annotations
+
+from typing import Any
+
+from durable_workflow import serializer, workflow
+from durable_workflow.workflow import (
+    CompleteWorkflow,
+    ScheduleActivity,
+    WorkflowContext,
+    replay,
+)
+
+
+@workflow.defn(name="approval-workflow")
+class ApprovalWorkflow:
+    def __init__(self) -> None:
+        self.approved: bool = False
+        self.approvals: list[str] = []
+
+    @workflow.signal("approve")
+    def on_approve(self, by: str) -> None:
+        self.approved = True
+        self.approvals.append(by)
+
+    def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
+        yield ctx.schedule_activity("wait_for_approval", [])
+        return {"approved": self.approved, "approvals": list(self.approvals)}
+
+
+@workflow.defn(name="typed-signal-workflow")
+class TypedSignalWorkflow:
+    def __init__(self) -> None:
+        self.count: int = 0
+
+    @workflow.signal("increment")
+    def on_increment(self, amount: int) -> None:
+        self.count += amount
+
+    def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
+        yield ctx.schedule_activity("tick", [])
+        yield ctx.schedule_activity("tick", [])
+        return self.count
+
+
+class TestSignalDecoratorRegistry:
+    def test_defn_collects_signal_methods_into_registry(self) -> None:
+        signals = ApprovalWorkflow.__workflow_signals__  # type: ignore[attr-defined]
+
+        assert signals == {"approve": "on_approve"}
+
+    def test_signal_decorator_sets_signal_name_marker(self) -> None:
+        assert ApprovalWorkflow.on_approve.__signal_name__ == "approve"  # type: ignore[attr-defined]
+
+    def test_workflow_without_signal_methods_has_empty_registry(self) -> None:
+        @workflow.defn(name="no-signals")
+        class PlainWorkflow:
+            def run(self, ctx: WorkflowContext) -> str:
+                return "done"
+
+        assert PlainWorkflow.__workflow_signals__ == {}  # type: ignore[attr-defined]
+
+
+def _activity_completed_event(result: Any) -> dict[str, Any]:
+    return {
+        "event_type": "ActivityCompleted",
+        "payload": {"result": serializer.envelope(result), "payload_codec": serializer.AVRO_CODEC},
+    }
+
+
+def _signal_received_event(name: str, args: list[Any]) -> dict[str, Any]:
+    return {
+        "event_type": "SignalReceived",
+        "payload": {
+            "signal_name": name,
+            "value": serializer.envelope(args),
+            "payload_codec": serializer.AVRO_CODEC,
+        },
+    }
+
+
+class TestSignalDispatchDuringReplay:
+    def test_signal_mutates_workflow_state_before_completion(self) -> None:
+        events = [
+            _signal_received_event("approve", ["alice"]),
+            _activity_completed_event("done"),
+        ]
+
+        outcome = replay(ApprovalWorkflow, events, [])
+
+        # Workflow returns — CompleteWorkflow emitted with mutated state.
+        assert len(outcome.commands) == 1
+        assert isinstance(outcome.commands[0], CompleteWorkflow)
+        assert outcome.commands[0].result == {"approved": True, "approvals": ["alice"]}
+
+    def test_multiple_signals_apply_in_history_order(self) -> None:
+        events = [
+            _signal_received_event("increment", [3]),
+            _activity_completed_event(None),
+            _signal_received_event("increment", [5]),
+            _activity_completed_event(None),
+        ]
+
+        outcome = replay(TypedSignalWorkflow, events, [])
+
+        assert len(outcome.commands) == 1
+        assert isinstance(outcome.commands[0], CompleteWorkflow)
+        assert outcome.commands[0].result == 8
+
+    def test_signal_before_any_activity_still_applies(self) -> None:
+        events = [
+            _signal_received_event("approve", ["bob"]),
+        ]
+
+        outcome = replay(ApprovalWorkflow, events, [])
+
+        # Activity has not completed yet — workflow yields ScheduleActivity and
+        # the signal has already been applied to instance state.
+        assert len(outcome.commands) == 1
+        assert isinstance(outcome.commands[0], ScheduleActivity)
+
+    def test_unknown_signal_is_silently_dropped(self) -> None:
+        events = [
+            _signal_received_event("not_a_real_signal", ["payload"]),
+            _activity_completed_event("done"),
+        ]
+
+        outcome = replay(ApprovalWorkflow, events, [])
+
+        assert len(outcome.commands) == 1
+        assert isinstance(outcome.commands[0], CompleteWorkflow)
+        # Handler for "approve" was never invoked because no matching signal
+        # arrived — state remains the default.
+        assert outcome.commands[0].result == {"approved": False, "approvals": []}
+
+    def test_signal_with_no_args_decodes_to_empty_handler_call(self) -> None:
+        @workflow.defn(name="ping-workflow")
+        class PingWorkflow:
+            def __init__(self) -> None:
+                self.pings: int = 0
+
+            @workflow.signal("ping")
+            def on_ping(self) -> None:
+                self.pings += 1
+
+            def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
+                yield ctx.schedule_activity("step", [])
+                return self.pings
+
+        events = [
+            _signal_received_event("ping", []),
+            _signal_received_event("ping", []),
+            _activity_completed_event(None),
+        ]
+
+        outcome = replay(PingWorkflow, events, [])
+
+        assert isinstance(outcome.commands[0], CompleteWorkflow)
+        assert outcome.commands[0].result == 2