[cross-repo from server#312] Conformance blocker: expand replay coverage beyond current smoke (#111)

Author: durable-workflow-ops

src/durable_workflow/__init__.py

@@ -58,6 +58,7 @@
     DurableWorkflowError,
     InvalidArgument,
     NamespaceNotFound,
+    NonDeterministicReplayError,
     NonRetryableError,
     QueryFailed,
     ScheduleAlreadyExists,
@@ -270,6 +271,7 @@
     "LocalFilesystemExternalStorage",
     "MetricsRecorder",
     "NamespaceNotFound",
+    "NonDeterministicReplayError",
     "NoopMetrics",
     "QueryFailed",
     "PrometheusMetrics",

src/durable_workflow/errors.py

@@ -198,6 +198,33 @@ def __init__(
         self.exception_type = exception_type
 
 
+class NonDeterministicReplayError(DurableWorkflowError):
+    """Current workflow code yielded a command shape that does not match history."""
+
+    def __init__(
+        self,
+        workflow_sequence: int,
+        expected_shape: str,
+        recorded_event_types: list[str],
+        *,
+        detail: str | None = None,
+    ) -> None:
+        message = (
+            f"workflow history at workflow sequence {workflow_sequence} recorded "
+            f"{recorded_event_types!r}, but the current workflow yielded {expected_shape}."
+        )
+        if detail:
+            message = f"{message} {detail}"
+        message = (
+            f"{message} Keep yielded workflow steps stable across deployments or "
+            "run this workflow on a compatible build."
+        )
+        super().__init__(message)
+        self.workflow_sequence = workflow_sequence
+        self.expected_shape = expected_shape
+        self.recorded_event_types = list(recorded_event_types)
+
+
 class UpdateRejected(DurableWorkflowError):
     """A workflow update was rejected by the workflow's validator."""
 

src/durable_workflow/replay_verify.py

@@ -34,7 +34,7 @@
 from pathlib import Path
 from typing import Any, Callable
 
-from .errors import ChildWorkflowFailed, WorkflowFailed
+from .errors import ChildWorkflowFailed, NonDeterministicReplayError, WorkflowFailed
 from .workflow import (
     CompleteWorkflow,
     ReplayOutcome,
@@ -172,6 +172,7 @@ class CaseReport:
     workflow_type: str | None
     family: str | None
     source: Mapping[str, Any] | None = None
+    divergence: Mapping[str, Any] | None = None
     expected: Mapping[str, Any] | None = None
     observed: Mapping[str, Any] | None = None
     error: Mapping[str, Any] | None = None
@@ -196,6 +197,7 @@ def to_dict(self) -> dict[str, Any]:
             "workflow_type": self.workflow_type,
             "family": self.family,
             "source": dict(self.source) if self.source is not None else None,
+            "divergence": dict(self.divergence) if self.divergence is not None else None,
             "expected": dict(self.expected) if self.expected is not None else None,
             "observed": dict(self.observed) if self.observed is not None else None,
             "error": dict(self.error) if self.error is not None else None,
@@ -550,6 +552,30 @@ def _replay_case(replayer: Replayer, case: Mapping[str, Any]) -> CaseReport:
             case.get("start_input") or [],
             workflow_type=workflow_type,
         )
+    except NonDeterministicReplayError as exc:
+        divergence = {
+            "workflow_sequence": exc.workflow_sequence,
+            "expected_shape": exc.expected_shape,
+            "recorded_event_types": list(exc.recorded_event_types),
+            "message": str(exc),
+        }
+        return CaseReport(
+            id=case_id,
+            status=STATUS_DRIFTED,
+            reason=REASON_SHAPE_MISMATCH,
+            workflow_type=workflow_type,
+            family=family,
+            source=source,
+            divergence=divergence,
+            expected=case.get("expected"),
+            error={
+                "class": type(exc).__name__,
+                "message": str(exc),
+                "workflow_sequence": exc.workflow_sequence,
+                "expected_shape": exc.expected_shape,
+                "recorded_event_types": list(exc.recorded_event_types),
+            },
+        )
     except (TypeError, ValueError, ChildWorkflowFailed, WorkflowFailed) as exc:
         return CaseReport(
             id=case_id,