Add NonDeterminism Tracking

Author: natemort

cadence/_internal/workflow/context.py

@@ -4,7 +4,6 @@
 from typing import Iterator, Optional, Any, Unpack, Type, cast
 
 from cadence._internal.workflow.statemachine.decision_manager import DecisionManager
-from cadence._internal.workflow.decisions_helper import DecisionsHelper
 from cadence.api.v1.common_pb2 import ActivityType
 from cadence.api.v1.decision_pb2 import ScheduleActivityTaskDecisionAttributes
 from cadence.api.v1.tasklist_pb2 import TaskList, TaskListKind
@@ -21,7 +20,6 @@ def __init__(
         self._info = info
         self._replay_mode = True
         self._replay_current_time_milliseconds: Optional[int] = None
-        self._decision_helper = DecisionsHelper()
         self._decision_manager = decision_manager
 
     def info(self) -> WorkflowInfo:
@@ -70,9 +68,7 @@ async def execute_activity(
         )
 
         activity_input = self.data_converter().to_data(list(args))
-        activity_id = self._decision_helper.generate_activity_id(activity)
         schedule_attributes = ScheduleActivityTaskDecisionAttributes(
-            activity_id=activity_id,
             activity_type=ActivityType(name=activity),
             domain=self.info().workflow_domain,
             task_list=TaskList(kind=TaskListKind.TASK_LIST_KIND_NORMAL, name=task_list),

cadence/_internal/workflow/decisions_helper.py

@@ -1,3 +1,4 @@
+import traceback
 from asyncio import AbstractEventLoop, Handle, TimerHandle, futures, tasks, Future, Task
 from contextvars import Context
 import logging
@@ -9,6 +10,12 @@
 
 logger = logging.getLogger(__name__)
 
+
+class FatalDecisionError(Exception):
+    def __init__(self, *args) -> None:
+        super().__init__(*args)
+
+
 _Ts = TypeVarTuple("_Ts")
 _T = TypeVar("_T")
 
@@ -455,9 +462,34 @@ def default_exception_handler(self, context: dict[str, Any]) -> None:
         )
 
     def call_exception_handler(self, context: dict[str, Any]) -> None:
-        raise NotImplementedError(
-            "Custom exception handlers not supported in deterministic event loop"
-        )
+        message = context.get("message")
+        if not message:
+            message = "Unhandled exception in event loop"
+
+        exception = context.get("exception")
+        if isinstance(exception, BaseException):
+            exc_info = exception
+        else:
+            exc_info = None
+
+        log_lines = [message]
+        for key in sorted(context):
+            if key in {"message", "exception"}:
+                continue
+            value = context[key]
+            if key == "source_traceback":
+                tb = "".join(traceback.format_list(value))
+                value = "Object created at (most recent call last):\n"
+                value += tb.rstrip()
+            elif key == "handle_traceback":
+                tb = "".join(traceback.format_list(value))
+                value = "Handle created at (most recent call last):\n"
+                value += tb.rstrip()
+            else:
+                value = repr(value)
+            log_lines.append(f"{key}: {value}")
+
+        logger.error("\n".join(log_lines), exc_info=exc_info)
 
     # Task factory
     def set_task_factory(  # type: ignore[override]