Preserve legacy parallel timing and prepare explicit propagation

CHANGELOG.md

@@ -6,6 +6,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Changed
+
+- Replay refuses explicit `parallel_group_policy` histories before constructing application state until a qualified policy consumer is enabled. Absent-policy histories retain their wait, failure-selection and clock behavior. Explicit propagation is being prepared behind disabled worker admission. (#307)
+
 ### Added
 
 - Timeout cleanup interruption (workflow#863, workflow#1129). `CancellationContext.from_dict()` reads the context's optional top-level `cause`: absent or `"cancellation"` is an ordinary cancellation, `"timeout"` marks a timeout-cleanup request (the run passed its work deadline and is now in cleanup), and any other value raises `ValueError`. The context carries no other timeout fields: its existing `cleanup_deadline_at` is the cleanup deadline, and the timeout kind and work deadline stay on the `WorkflowTimeoutCleanupStarted` event. `cause` and `is_timeout` are readable on the context, and `to_dict()` writes `cause` back exactly as received, including an explicit `"cancellation"`, leaving ordinary contexts byte-identical. A timeout request is raised at the authored call as the new experimental `WorkflowTimeoutCleanupRequested`, a subclass of `WorkflowCancelled`, so existing `except WorkflowCancelled:` handlers, such as saga compensation, run on timeout too. It stays `BaseException`-rooted and is not an `asyncio.CancelledError`. Test `cause`, the cross-SDK discriminator, to tell the two apart. `cause` is never inherited: a child cancelled under a timeout request, scoped or not, receives an ordinary cancellation without `cause`.

src/durable_workflow/replay_clock.py

@@ -343,6 +343,19 @@ def observe_activity_group(
             events.append(event)
         self._pending = (None, _activity_group_time(events, failed_member))
 
+    def observe_propagated_activity_failure(self, member: _GroupMember, *, reconstruction: bool) -> None:
+        """Use only the qualified first-recorded failure of an explicit group."""
+        self._pending = (None, None)
+        if reconstruction:
+            return
+        event = self._activity_group_boundary(member)
+        if event is None or _kind(event) == "ActivityCompleted":
+            return
+        time = _group_event_time(event)
+        milliseconds = canonical_milliseconds(time) if time is not None else None
+        if milliseconds is not None:
+            self._pending = (member[0], from_canonical_milliseconds(milliseconds))
+
     def _activity_group_boundary(self, member: _GroupMember) -> Mapping[str, Any] | None:
         index, sequence, activity_type, path = member
         indices = self._by_sequence.get(sequence, [])

tests/fixtures/replay_regressions/explicit-parallel-policy-requires-consumer.json

@@ -0,0 +1,108 @@
+{
+  "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json",
+  "fixture_schema": "durable-workflow.replay-regression/v1",
+  "id": "explicit-parallel-policy-requires-consumer",
+  "protocol_version": "1.20",
+  "bindings": [
+    "python"
+  ],
+  "workflow": {
+    "type": "tests.replay.parallel-failure-policy",
+    "input": [],
+    "payload_codec": "avro"
+  },
+  "history": [
+    {
+      "event_type": "WorkflowStarted",
+      "recorded_at": "2026-01-01T00:00:00Z",
+      "payload": {},
+      "sequence": 1
+    },
+    {
+      "event_type": "ActivityScheduled",
+      "recorded_at": "2026-01-01T00:00:00Z",
+      "payload": {
+        "sequence": 1,
+        "activity_execution_id": "execution-0",
+        "activity_type": "slow-first",
+        "parallel_group_id": "parallel-activities:1:2",
+        "parallel_group_kind": "activity",
+        "parallel_group_base_sequence": 1,
+        "parallel_group_size": 2,
+        "parallel_group_index": 0,
+        "parallel_group_path": [
+          {
+            "parallel_group_id": "parallel-activities:1:2",
+            "parallel_group_kind": "activity",
+            "parallel_group_base_sequence": 1,
+            "parallel_group_size": 2,
+            "parallel_group_index": 0,
+            "parallel_group_policy": "propagate"
+          }
+        ],
+        "parallel_group_policy": "propagate"
+      },
+      "sequence": 2
+    },
+    {
+      "event_type": "ActivityScheduled",
+      "recorded_at": "2026-01-01T00:00:00Z",
+      "payload": {
+        "sequence": 2,
+        "activity_execution_id": "execution-1",
+        "activity_type": "fast-second",
+        "parallel_group_id": "parallel-activities:1:2",
+        "parallel_group_kind": "activity",
+        "parallel_group_base_sequence": 1,
+        "parallel_group_size": 2,
+        "parallel_group_index": 1,
+        "parallel_group_path": [
+          {
+            "parallel_group_id": "parallel-activities:1:2",
+            "parallel_group_kind": "activity",
+            "parallel_group_base_sequence": 1,
+            "parallel_group_size": 2,
+            "parallel_group_index": 1,
+            "parallel_group_policy": "propagate"
+          }
+        ],
+        "parallel_group_policy": "propagate"
+      },
+      "sequence": 3
+    },
+    {
+      "event_type": "ActivityFailed",
+      "recorded_at": "2026-01-01T00:00:01Z",
+      "payload": {
+        "sequence": 2,
+        "activity_execution_id": "execution-1",
+        "activity_type": "fast-second",
+        "parallel_group_id": "parallel-activities:1:2",
+        "parallel_group_kind": "activity",
+        "parallel_group_base_sequence": 1,
+        "parallel_group_size": 2,
+        "parallel_group_index": 1,
+        "parallel_group_path": [
+          {
+            "parallel_group_id": "parallel-activities:1:2",
+            "parallel_group_kind": "activity",
+            "parallel_group_base_sequence": 1,
+            "parallel_group_size": 2,
+            "parallel_group_index": 1,
+            "parallel_group_policy": "propagate"
+          }
+        ],
+        "message": "higher position failed first",
+        "parallel_group_policy": "propagate"
+      },
+      "sequence": 4
+    }
+  ],
+  "expected": {
+    "command_sequence": []
+  },
+  "expected_replay_error": {
+    "type": "LocalActivityExecutionAborted",
+    "message_contains": "parallel_group_policy_not_supported"
+  }
+}

tests/fixtures/replay_regressions/legacy-parallel-failure-remains-pending.json

@@ -0,0 +1,101 @@
+{
+  "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json",
+  "fixture_schema": "durable-workflow.replay-regression/v1",
+  "id": "legacy-parallel-failure-remains-pending",
+  "protocol_version": "1.20",
+  "bindings": [
+    "python"
+  ],
+  "workflow": {
+    "type": "tests.replay.parallel-failure-policy",
+    "input": [],
+    "payload_codec": "avro"
+  },
+  "history": [
+    {
+      "event_type": "WorkflowStarted",
+      "recorded_at": "2026-01-01T00:00:00Z",
+      "payload": {},
+      "sequence": 1
+    },
+    {
+      "event_type": "ActivityScheduled",
+      "recorded_at": "2026-01-01T00:00:00Z",
+      "payload": {
+        "sequence": 1,
+        "activity_execution_id": "execution-0",
+        "activity_type": "slow-first",
+        "parallel_group_id": "parallel-activities:1:2",
+        "parallel_group_kind": "activity",
+        "parallel_group_base_sequence": 1,
+        "parallel_group_size": 2,
+        "parallel_group_index": 0,
+        "parallel_group_path": [
+          {
+            "parallel_group_id": "parallel-activities:1:2",
+            "parallel_group_kind": "activity",
+            "parallel_group_base_sequence": 1,
+            "parallel_group_size": 2,
+            "parallel_group_index": 0
+          }
+        ]
+      },
+      "sequence": 2
+    },
+    {
+      "event_type": "ActivityScheduled",
+      "recorded_at": "2026-01-01T00:00:00Z",
+      "payload": {
+        "sequence": 2,
+        "activity_execution_id": "execution-1",
+        "activity_type": "fast-second",
+        "parallel_group_id": "parallel-activities:1:2",
+        "parallel_group_kind": "activity",
+        "parallel_group_base_sequence": 1,
+        "parallel_group_size": 2,
+        "parallel_group_index": 1,
+        "parallel_group_path": [
+          {
+            "parallel_group_id": "parallel-activities:1:2",
+            "parallel_group_kind": "activity",
+            "parallel_group_base_sequence": 1,
+            "parallel_group_size": 2,
+            "parallel_group_index": 1
+          }
+        ]
+      },
+      "sequence": 3
+    },
+    {
+      "event_type": "ActivityFailed",
+      "recorded_at": "2026-01-01T00:00:01Z",
+      "payload": {
+        "sequence": 2,
+        "activity_execution_id": "execution-1",
+        "activity_type": "fast-second",
+        "parallel_group_id": "parallel-activities:1:2",
+        "parallel_group_kind": "activity",
+        "parallel_group_base_sequence": 1,
+        "parallel_group_size": 2,
+        "parallel_group_index": 1,
+        "parallel_group_path": [
+          {
+            "parallel_group_id": "parallel-activities:1:2",
+            "parallel_group_kind": "activity",
+            "parallel_group_base_sequence": 1,
+            "parallel_group_size": 2,
+            "parallel_group_index": 1
+          }
+        ],
+        "message": "higher position failed first"
+      },
+      "sequence": 4
+    }
+  ],
+  "expected": {
+    "command_sequence": [
+      {"type": "schedule_activity", "activity_type": "slow-first"},
+      {"type": "schedule_activity", "activity_type": "fast-second"}
+    ]
+  }
+}

tests/fixtures/replay_regressions/run-close-policy-requires-consumer.json

@@ -0,0 +1,56 @@
+{
+  "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json",
+  "fixture_schema": "durable-workflow.replay-regression/v1",
+  "id": "run-close-policy-requires-consumer",
+  "protocol_version": "1.20",
+  "bindings": [
+    "python"
+  ],
+  "workflow": {
+    "type": "tests.replay.parallel-failure-policy",
+    "input": [],
+    "payload_codec": "avro"
+  },
+  "history": [
+    {
+      "event_type": "WorkflowStarted",
+      "recorded_at": "2026-01-01T00:00:00Z",
+      "payload": {},
+      "sequence": 1
+    },
+    {
+      "event_type": "ActivityCancelled",
+      "recorded_at": "2026-01-01T00:00:09Z",
+      "payload": {
+        "sequence": 1,
+        "activity_execution_id": "execution-0",
+        "activity_type": "slow-first",
+        "parallel_group_id": "parallel-activities:1:2",
+        "parallel_group_kind": "activity",
+        "parallel_group_base_sequence": 1,
+        "parallel_group_size": 2,
+        "parallel_group_index": 0,
+        "parallel_group_path": [
+          {
+            "parallel_group_id": "parallel-activities:1:2",
+            "parallel_group_kind": "activity",
+            "parallel_group_base_sequence": 1,
+            "parallel_group_size": 2,
+            "parallel_group_index": 0,
+            "parallel_group_policy": "propagate"
+          }
+        ],
+        "parallel_group_policy": "propagate",
+        "cancellation_reason": "run_closed"
+      },
+      "sequence": 2
+    }
+  ],
+  "expected": {
+    "command_sequence": []
+  },
+  "expected_replay_error": {
+    "type": "LocalActivityExecutionAborted",
+    "message_contains": "parallel_group_policy_not_supported"
+  }
+}