[cross-repo from workflow#631] Conformance blocker: expand replay coverage beyond current smoke (#113)

Author: durable-workflow-ops

CONFORMANCE.md

@@ -26,7 +26,7 @@ The Python SDK claims two targets from the suite's matrix:
 | `control_plane_request_response` | `tests/fixtures/control-plane/` | stable, parity-shared with `cli` |
 | `signal_query_runtime_contract` | `tests/test_signals.py`, `tests/test_queries.py`, `tests/test_worker.py` | stable, parity-shared with PHP worker, CLI, and server routes |
 | `worker_task_lifecycle` | `tests/fixtures/external-task-input/`, `tests/fixtures/external-task-result/` | stable |
-| `history_replay_bundles` | `tests/fixtures/golden_history/` | stable, parity-shared with `workflow` golden bundles |
+| `history_replay_bundles` | `tests/fixtures/golden_history/` and the public replay scenario manifest at <https://durable-workflow.github.io/platform-conformance/replay-runtime-scenarios.json> | stable, parity-shared with `workflow` golden bundles and the full runtime replay scenario matrix |
 
 The fixtures in this repo are exercised today by:
 
@@ -51,7 +51,7 @@ result document before tag, with the conformance level at `full` or
 | Field | Value |
 | --- | --- |
 | Required claimed targets | `official_sdk`, `worker_protocol_implementation` |
-| Required suite version | `PlatformConformanceSuite::VERSION` (currently `2`, mirrored at `/platform-conformance-contract.json`) |
+| Required suite version | `PlatformConformanceSuite::VERSION` (currently `3`, mirrored at `/platform-conformance-contract.json`) |
 | CI job | `platform-conformance` (lands when the harness reference implementation publishes; until then `cli-parity` and `test_history_event_contract.py` cover the same ground) |
 | Block on `nonconforming` | yes |
 | Artifact attached to release | harness result document, schema `durable-workflow.v2.platform-conformance.result` |

src/durable_workflow/workflow.py

@@ -2190,30 +2190,27 @@ def _assert_next_step_matches(command: Any, offset: int = 0) -> None:
             return
         _assert_step_matches(command, recorded_steps[step_index])
 
-    def _next_unconsumed_recorded_step() -> _RecordedStep | None:
+    def _unconsumed_recorded_steps() -> list[_RecordedStep]:
         candidates: list[_RecordedStep] = []
         if result_cursor < len(recorded_steps):
-            candidates.append(recorded_steps[result_cursor])
+            candidates.extend(recorded_steps[result_cursor:])
         if wait_yield_count < len(recorded_wait_steps):
-            candidates.append(recorded_wait_steps[wait_yield_count])
+            candidates.extend(recorded_wait_steps[wait_yield_count:])
         if pending_step_cursor < len(recorded_pending_steps):
-            candidates.append(recorded_pending_steps[pending_step_cursor])
+            candidates.extend(recorded_pending_steps[pending_step_cursor:])
+        return sorted(candidates, key=lambda step: step.workflow_sequence)
+
+    def _next_unconsumed_recorded_step() -> _RecordedStep | None:
+        candidates = _unconsumed_recorded_steps()
         if not candidates:
             return None
-        return min(candidates, key=lambda step: step.workflow_sequence)
+        return candidates[0]
 
     def _assert_pending_step_matches(command: Any, offset: int = 0) -> None:
-        if offset == 0:
-            step = _next_unconsumed_recorded_step()
-            if step is None:
-                return
-            _assert_step_matches(command, step)
-            return
-
-        step_index = pending_step_cursor + offset
-        if step_index >= len(recorded_pending_steps):
+        candidates = _unconsumed_recorded_steps()
+        if offset >= len(candidates):
             return
-        _assert_step_matches(command, recorded_pending_steps[step_index])
+        _assert_step_matches(command, candidates[offset])
 
     def _assert_no_unconsumed_history(terminal_shape: str) -> None:
         step = _next_unconsumed_recorded_step()

tests/test_replay.py

@@ -6,7 +6,7 @@
 import pytest
 
 from durable_workflow import Replayer, ReplayOutcome, serializer, workflow
-from durable_workflow.errors import ActivityFailed, ChildWorkflowFailed
+from durable_workflow.errors import ActivityFailed, ChildWorkflowFailed, NonDeterministicReplayError
 from durable_workflow.workflow import (
     ActivityRetryPolicy,
     ChildWorkflowRetryPolicy,
@@ -165,6 +165,21 @@ def test_first_replay_schedules(self) -> None:
         assert cmd.activity_type == "greet"
         assert cmd.arguments == ["world"]
 
+    def test_pending_timer_history_rejects_single_activity_drift(self) -> None:
+        history = [
+            {
+                "event_type": "TimerScheduled",
+                "payload": {"sequence": 1, "timer_kind": "durable_timer"},
+            },
+        ]
+
+        with pytest.raises(NonDeterministicReplayError) as exc_info:
+            replay(OneActivity, history, ["world"])
+
+        assert exc_info.value.workflow_sequence == 1
+        assert exc_info.value.expected_shape == "activity"
+        assert exc_info.value.recorded_event_types == ["TimerScheduled"]
+
     def test_completed_activity_triggers_completion(self) -> None:
         history = [{"event_type": "ActivityCompleted", "payload": {"result": '"hello, world"'}}]
         outcome = replay(OneActivity, history, ["world"])
@@ -1229,6 +1244,13 @@ def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
         return result
 
 
+@workflow.defn(name="fan-out-activity-timer-wf")
+class FanOutActivityTimerWorkflow:
+    def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
+        yield [ctx.schedule_activity("fetch", []), ctx.start_timer(5)]
+        return "done"
+
+
 @workflow.defn(name="fan-out-then-sequential-wf")
 class FanOutThenSequentialWorkflow:
     def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
@@ -1288,6 +1310,24 @@ def test_timers_fired_then_activity(self) -> None:
         assert isinstance(outcome.commands[0], ScheduleActivity)
         assert outcome.commands[0].activity_type == "after-timers"
 
+    def test_pending_batch_history_matches_by_workflow_sequence(self) -> None:
+        history = [
+            {
+                "event_type": "TimerScheduled",
+                "payload": {"sequence": 2, "timer_kind": "durable_timer"},
+            },
+            {
+                "event_type": "ActivityScheduled",
+                "payload": {"sequence": 1, "activity_type": "fetch"},
+            },
+        ]
+
+        outcome = replay(FanOutActivityTimerWorkflow, history, [])
+
+        assert len(outcome.commands) == 2
+        assert isinstance(outcome.commands[0], ScheduleActivity)
+        assert isinstance(outcome.commands[1], StartTimer)
+
     def test_fan_out_then_sequential(self) -> None:
         history = [
             {"event_type": "ActivityCompleted", "payload": {"result": '"r1"'}},

tests/test_replay_verify.py

@@ -149,6 +149,31 @@ def test_verify_replay_refuses_in_flight_scheduled_command_drift() -> None:
     assert report.error["recorded_event_types"] == ["ActivityScheduled"]
 
 
+def test_verify_replay_refuses_single_command_replacing_pending_timer() -> None:
+    history = [
+        {
+            "event_type": "TimerScheduled",
+            "payload": {
+                "sequence": 1,
+                "timer_kind": "durable_timer",
+            },
+        }
+    ]
+
+    report = verify_replay(
+        ActivityDivergenceWorkflow,
+        history,
+        case_id="activity-replaced-timer",
+    )
+
+    assert report.status == STATUS_DRIFTED
+    assert report.reason == REASON_SHAPE_MISMATCH
+    assert report.error is not None
+    assert report.error["workflow_sequence"] == 1
+    assert report.error["expected_shape"] == "activity"
+    assert report.error["recorded_event_types"] == ["TimerScheduled"]
+
+
 def test_verify_replay_refuses_same_shape_activity_name_drift() -> None:
     history = [
         {