feat: Add NonDeterminism Tracking (cadence-workflow#65)

<!-- Describe what has changed in this PR -->
**What changed?**
- Remove DecisionsHelper. The mechanism it used for generating IDs is
inconsistent with the other clients, where they all share an
incrementing integer. The other functionality it provided was unused.
- Add NonDeterminism Tracking to provide specific errors when
nondeterminism occurs.
- We look at the outcome decisions for the batch and create a set of
expectations that must be met so that we can immediately flag
non-determinism, rather than waiting until after the user code finishing
execution.
  - We add markers any time there's immediate cancellation.
- Add a state machine for Workflow completion. We need to route the
Decision attributes through the DecisionManager so that it can check for
nondeterminism.
- Restructure workflow exception handling a bit to handle
ExceptionGroup. This still needs more work.

<!-- Tell your future self why have you made these changes -->
**Why?**
- Feature parity with other clients
- Trying out new non-determinism detection algorithm to see if it would
work for other clients.

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

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

<!-- 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**

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,36 @@ 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"
-        )
+        # This is called if a task has an unhandled exception. Short term, it's helpful to log these for debugging.
+        # Long term, we need some combination of failing decision tasks or workflows based on these errors.
+        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]