From 6d175d084a36a366f48d3b5823bf30588b3ea070 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Thu, 5 Nov 2026 02:40:07 +0000 Subject: [PATCH 1/3] Bound the redrive successor call ledger and match terminal identities --- tests/test_redrive_call_ledger.py | 223 +++++++++++++++++++++++ tests/test_redrive_successor_evidence.py | 11 +- tests/test_replay_clock.py | 93 +++++++--- 3 files changed, 298 insertions(+), 29 deletions(-) create mode 100644 tests/test_redrive_call_ledger.py diff --git a/tests/test_redrive_call_ledger.py b/tests/test_redrive_call_ledger.py new file mode 100644 index 0000000..1cd2dce --- /dev/null +++ b/tests/test_redrive_call_ledger.py @@ -0,0 +1,223 @@ +"""Source-only successor evidence controls; these are not producer/runtime proof. + +The ledger deliberately covers remote/local activity, plain timer and untimed +signal calls without internal-timeout rows. Core supports additional shapes +which this fixture model refuses. +The new positive failure/cancellation histories also have Python replay witnesses. +""" + +from __future__ import annotations + +from contextlib import suppress +from copy import deepcopy +from typing import Any + +import pytest + +from durable_workflow import workflow +from durable_workflow.errors import ActivityFailed, WorkflowCancelled +from durable_workflow.workflow import CompleteWorkflow, WorkflowContext, replay +from tests.test_redrive_successor_evidence import ( + CANCELLATION, + _delivery, + _renumber, + _request, + _step_three_activity, +) +from tests.test_replay_clock import check_redrive_successor, redrive_fixture + + +@workflow.defn(name="tests.replay.redrive-activity-recovery-witness") +class ActivityRecoveryWitness: + def run(self, ctx: WorkflowContext, name: str, seconds: int): # type: ignore[no-untyped-def] + greeting = yield ctx.schedule_activity("tests.redrive.first", [name]) + yield ctx.start_timer(seconds) + try: + with suppress(ActivityFailed): + yield ctx.schedule_activity("tests.redrive.second", [greeting]) + yield ctx.schedule_activity("tests.redrive.recovery", []) + except WorkflowCancelled as error: + assert error.context is not None + return error.context.remaining() + + +@workflow.defn(name="tests.replay.redrive-timer-cancellation-witness") +class TimerCancellationWitness: + def run(self, ctx: WorkflowContext, name: str, seconds: int): # type: ignore[no-untyped-def] + yield ctx.schedule_activity("tests.redrive.first", [name]) + yield ctx.start_timer(seconds) + try: + yield ctx.start_timer(60) + except WorkflowCancelled as error: + assert error.context is not None + return error.context.remaining() + + +def document() -> dict[str, Any]: + return deepcopy(redrive_fixture(CANCELLATION)) + + +def terminal( + document: dict[str, Any], kind: str, *, before_request: bool = False, opened: dict[str, Any] | None = None, +) -> dict[str, Any]: + history = document["history"] + if opened is None: + opened = _step_three_activity(history) + boundary = _request(history) if before_request else _delivery(history) + row = {"event_type": kind, "recorded_at": boundary["recorded_at"], + "payload": {**opened["payload"], "message": "fixture terminal failure"}} + history.insert(history.index(boundary), row) + _renumber(history) + return row + + +def bind_witness(document: dict[str, Any], name: str) -> None: + document["workflow"]["type"] = name + document["history"][0]["payload"]["workflow_type"] = name + + +def activity_recovery(kind: str, before_request: bool) -> dict[str, Any]: + doc = document() + bind_witness(doc, "tests.replay.redrive-activity-recovery-witness") + terminal(doc, kind, before_request=before_request) + _delivery(doc["history"])["payload"]["sequence"] = 4 + return doc + + +def cancelled_timer() -> dict[str, Any]: + doc = document() + bind_witness(doc, "tests.replay.redrive-timer-cancellation-witness") + opened = _step_three_activity(doc["history"]) + opened.update(event_type="TimerScheduled", payload={"sequence": 3, "timer_id": "successor-timer", + "delay_seconds": 60}) + history = doc["history"] + row = terminal(doc, "TimerCancelled", opened=opened) + row["payload"] = {**opened["payload"], "cancelled_at": row["recorded_at"]} + _delivery(history)["payload"]["call_kind"] = "timer" + _renumber(history) + return doc + + +def assert_replays(doc: dict[str, Any], witness: type) -> None: + task = doc["workflow"] + outcome = replay(witness, doc["history"], task["input"], run_id=task["run_id"], + workflow_id=task["workflow_id"], payload_codec="avro") + assert list(outcome.commands) == [CompleteWorkflow(29.0)] + + +@pytest.mark.parametrize("kind", ["ActivityFailed", "ActivityTimedOut"]) +@pytest.mark.parametrize("before_request", [False, True]) +def test_failed_activity_positive_has_a_reachable_recovery_workflow(kind: str, before_request: bool) -> None: + doc = activity_recovery(kind, before_request) + assert_replays(doc, ActivityRecoveryWitness) + check_redrive_successor(doc["history"], doc["workflow"]) + + +def test_cancelled_timer_positive_has_a_reachable_cancellation_workflow() -> None: + doc = cancelled_timer() + assert_replays(doc, TimerCancellationWitness) + check_redrive_successor(doc["history"], doc["workflow"]) + + +@pytest.mark.parametrize("kind", ["condition", "child", "service_operation"]) +def test_unmodeled_unscheduled_call_families_are_explicitly_refused(kind: str) -> None: + doc = document() + history = doc["history"] + history.remove(_step_three_activity(history)) + _delivery(history)["payload"]["call_kind"] = kind + _renumber(history) + with pytest.raises(AssertionError, match="single call kind"): + check_redrive_successor(history, doc["workflow"]) + + +@pytest.mark.parametrize("kind", ["condition_timeout", "signal_timeout"]) +@pytest.mark.parametrize("event_kind", ["TimerScheduled", "TimerFired", "TimerCancelled"]) +def test_internal_timeout_rows_are_outside_the_plain_timer_model(kind: str, event_kind: str) -> None: + doc = cancelled_timer() + history = doc["history"] + if event_kind == "TimerFired": + next(row for row in history if row["event_type"] == "TimerCancelled")["event_type"] = "TimerFired" + target = next(row for row in history if row["event_type"] == event_kind and row["payload"]["sequence"] == 3) + target["payload"].update(timer_kind=kind, condition_wait_id="wait", signal_wait_id="wait") + with pytest.raises(AssertionError): + check_redrive_successor(history, doc["workflow"]) + + +@pytest.mark.parametrize("kind", ["ActivityCompleted", "ActivityCancelled", "ActivityFailed", "ActivityTimedOut"]) +@pytest.mark.parametrize("mutation", ["resource", "step", "mode", "before-opener"]) +def test_activity_terminals_cannot_resolve_a_different_call(kind: str, mutation: str) -> None: + doc = document() + history = doc["history"] + row = terminal(doc, kind) + if mutation == "resource": + row["payload"]["activity_execution_id"] = "foreign" + elif mutation == "step": + row["payload"]["sequence"] = 4 + elif mutation == "mode": + row["payload"]["execution_mode"] = "local" + else: + history.remove(row) + history.insert(history.index(_step_three_activity(history)), row) + _renumber(history) + with pytest.raises(AssertionError): + check_redrive_successor(history, doc["workflow"]) + + +@pytest.mark.parametrize("kind", ["ActivityCompleted", "ActivityCancelled", "ActivityFailed", "ActivityTimedOut"]) +@pytest.mark.parametrize("local", [False, True]) +@pytest.mark.parametrize("before_request", [False, True]) +def test_activity_terminal_families_gate_the_original_call_at_the_request( + kind: str, local: bool, before_request: bool, +) -> None: + doc = document() + opened = _step_three_activity(doc["history"]) + if local: + opened["payload"]["local_activity"] = True + _delivery(doc["history"])["payload"]["call_kind"] = "local_activity" + terminal(doc, kind, before_request=before_request) + if before_request: + with pytest.raises(AssertionError, match="resolved before its request"): + check_redrive_successor(doc["history"], doc["workflow"]) + else: + check_redrive_successor(doc["history"], doc["workflow"]) + + +@pytest.mark.parametrize("kind", ["TimerFired", "TimerCancelled"]) +@pytest.mark.parametrize("mutation", ["resource", "step", "before-opener", "before-request"]) +def test_timer_terminals_are_bound_to_the_original_timer(kind: str, mutation: str) -> None: + doc = cancelled_timer() + history = doc["history"] + row = next(row for row in history if row["event_type"] == "TimerCancelled") + row["event_type"] = kind + if mutation == "resource": + row["payload"]["timer_id"] = "foreign" + elif mutation == "step": + row["payload"]["sequence"] = 4 + else: + history.remove(row) + boundary = _request(history) if mutation == "before-request" else next( + event for event in history if event["event_type"] == "TimerScheduled" and event["payload"]["sequence"] == 3) + history.insert(history.index(boundary), row) + _renumber(history) + with pytest.raises(AssertionError): + check_redrive_successor(history, doc["workflow"]) + + +@pytest.mark.parametrize("local_field", ["local_activity", "execution_mode"]) +def test_local_activity_shape_and_delivery_must_agree(local_field: str) -> None: + doc = document() + history = doc["history"] + opened = _step_three_activity(history) + opened["payload"][local_field] = True if local_field == "local_activity" else "local" + with pytest.raises(AssertionError, match="recorded call's kind"): + check_redrive_successor(history, doc["workflow"]) + _delivery(history)["payload"]["call_kind"] = "local_activity" + check_redrive_successor(history, doc["workflow"]) + + +@pytest.mark.parametrize("field", ["parallel_group_id", "selection_group_id", "cancellation_scope_id"]) +def test_grouped_and_scoped_rows_are_outside_the_model(field: str) -> None: + doc = document() + _step_three_activity(doc["history"])["payload"][field] = "outside-scope" + with pytest.raises(AssertionError): + check_redrive_successor(doc["history"], doc["workflow"]) diff --git a/tests/test_redrive_successor_evidence.py b/tests/test_redrive_successor_evidence.py index e404400..c5787dc 100644 --- a/tests/test_redrive_successor_evidence.py +++ b/tests/test_redrive_successor_evidence.py @@ -594,9 +594,10 @@ def test_a_whitespace_requester_value_is_accepted_as_core_does(name: str) -> Non check(name, changed(name, _context("both", lambda context: context["requester"].__setitem__("label", " ")))) -@pytest.mark.parametrize("call_kind", ["activity", "timer", "child", "service_operation"]) -def test_a_delivery_to_the_unscheduled_resume_step_may_name_any_single_call(call_kind: str) -> None: - # Core lets a delivery consume a call not yet scheduled; its shape is open. +@pytest.mark.parametrize("call_kind", ["activity", "local_activity", "timer", "signal"]) +def test_an_unscheduled_delivery_may_name_an_admitted_single_call(call_kind: str) -> None: + # Core also supports condition/child/service calls, but this deliberately + # narrower model cannot qualify their event/resource/terminal ledgers. name = "redrive-successor-reused-upserts-cancellation-clock" check(name, changed(name, lambda history: _delivery(history)["payload"].__setitem__("call_kind", call_kind))) @@ -748,7 +749,7 @@ def mutate(history: list[dict[str, Any]]) -> None: if state == "resolved": check(CANCELLATION, changed(CANCELLATION, mutate)) else: - with pytest.raises(AssertionError, match="the successor reached|no opened wait"): + with pytest.raises(AssertionError, match="the successor reached|no opened wait|terminal matches"): check(CANCELLATION, changed(CANCELLATION, mutate)) @@ -772,7 +773,7 @@ def mutate(history: list[dict[str, Any]]) -> None: _delivery(history)["payload"]["sequence"] = 5 _renumber(history) - with pytest.raises(AssertionError, match="the successor reached"): + with pytest.raises(AssertionError, match="the successor reached|terminal matches"): check(CANCELLATION, changed(CANCELLATION, mutate)) diff --git a/tests/test_replay_clock.py b/tests/test_replay_clock.py index a84b752..258ca66 100644 --- a/tests/test_replay_clock.py +++ b/tests/test_replay_clock.py @@ -497,9 +497,12 @@ def test_reused_signal_wait_events_do_not_move_the_cancellation_budget() -> None # followed by the reused TimerFired. "ActivityScheduled": ("sequence", "activity_type", "activity_execution_id"), "ActivityCompleted": ("sequence", "activity_type", "activity_execution_id"), + "ActivityFailed": ("sequence", "activity_type", "activity_execution_id"), + "ActivityTimedOut": ("sequence", "activity_type", "activity_execution_id"), "ActivityCancelled": ("sequence", "activity_type", "activity_execution_id"), "TimerScheduled": ("sequence", "timer_id"), "TimerFired": ("sequence", "timer_id"), + "TimerCancelled": ("sequence", "timer_id"), "SideEffectRecorded": ("sequence",), "VersionMarkerRecorded": ("sequence", "change_id"), } @@ -513,7 +516,8 @@ def test_reused_signal_wait_events_do_not_move_the_cancellation_budget() -> None # Successor events that occupy a workflow step of their own. REDRIVE_SUCCESSOR_STEP_EVENTS = frozenset({ "MemoUpserted", "SearchAttributesUpserted", "SignalWaitOpened", "SignalApplied", "ActivityScheduled", - "ActivityCompleted", "ActivityCancelled", "TimerScheduled", "TimerFired", "SideEffectRecorded", + "ActivityCompleted", "ActivityFailed", "ActivityTimedOut", "ActivityCancelled", "TimerScheduled", "TimerFired", + "TimerCancelled", "SideEffectRecorded", "VersionMarkerRecorded", "CooperativeCancellationDelivered", }) @@ -566,29 +570,44 @@ def redrive_time(value: Any) -> datetime: return parsed -# Events a successor cancellation can be delivered to: a call the successor -# recorded itself and that waits. -REDRIVE_CANCELLABLE_EVENTS = frozenset({"ActivityScheduled", "TimerScheduled", "SignalWaitOpened"}) +# This is a deliberately smaller model than Core's CALL_SHAPES. Condition, +# child, service-operation and internal timeout calls need their own identity +# and terminal ledgers before they can qualify here. Grouped/scoped calls are +# also excluded. A supported call has one opener and matching later terminals. +# Core 28893f97: WorkflowStepHistory and CooperativeCancellationDelivery. +REDRIVE_CALL_LEDGER = { + "activity": ("ActivityScheduled", "activity_execution_id", frozenset({ + "ActivityCompleted", "ActivityFailed", "ActivityTimedOut", "ActivityCancelled", + })), + "local_activity": ("ActivityScheduled", "activity_execution_id", frozenset({ + "ActivityCompleted", "ActivityFailed", "ActivityTimedOut", "ActivityCancelled", + })), + "timer": ("TimerScheduled", "timer_id", frozenset({"TimerFired", "TimerCancelled"})), + "signal": ("SignalWaitOpened", "signal_wait_id", frozenset({"SignalApplied"})), +} +REDRIVE_CANCELLABLE_EVENTS = frozenset(shape[0] for shape in REDRIVE_CALL_LEDGER.values()) +REDRIVE_DELIVERY_CALL_KINDS = frozenset(REDRIVE_CALL_LEDGER) +REDRIVE_RESOLUTION_EVENTS = frozenset(kind for shape in REDRIVE_CALL_LEDGER.values() for kind in shape[2]) def _recorded_call_kind(event: dict[str, Any]) -> str: - """The call_kind Core CooperativeCancellationDelivery::callKindAt derives.""" + """The admitted shape, including Core WorkflowStepHistory's local mode.""" payload = event["payload"] - if event["event_type"] == "ActivityScheduled": - return "local_activity" if payload.get("local_activity") is True else "activity" - if event["event_type"] == "TimerScheduled": - return {"condition_timeout": "condition", "signal_timeout": "signal"}.get(payload.get("timer_kind"), "timer") + if event["event_type"].startswith("Activity"): + return "local_activity" if payload.get("local_activity") is True or payload.get("execution_mode") == "local" \ + else "activity" + if event["event_type"].startswith("Timer"): + return "timer" return "signal" -# Core CooperativeCancellationDelivery::CALL_SHAPES: single-call delivery kinds. -REDRIVE_DELIVERY_CALL_KINDS = frozenset({ - "activity", "local_activity", "timer", "condition", "signal", "child", "service_operation", -}) -# Core CooperativeCancellationDelivery::RESOLUTION_TYPES this check can see. -REDRIVE_RESOLUTION_EVENTS = frozenset({"ActivityCompleted", "ActivityCancelled", "TimerFired", "SignalApplied"}) def check_redrive_successor(history: list[dict[str, Any]], workflow: dict[str, Any]) -> None: - """Require a successor Workflow 2.19 can write for this fixture's workflow.""" + """Check the fixture's redrive identity/prefix and admitted single-call ledger. + + This is bounded evidence validation, not a full Core history validator or + proof that arbitrary workflow code can reach every structurally valid row. + The concrete positive replay witnesses remain separate from this model. + """ def require(condition: bool, message: str) -> None: if not condition: @@ -701,7 +720,8 @@ def _check_successor_cancellation( CancellationContext (CancellationContext::fromArray), and CooperativeCancellationDelivery::record and ::validateCallBoundary write and gate the delivery. Parallel and selection_handle deliveries are - deliberately refused: no successor fixture uses those shapes. + deliberately refused, as are condition/child/service and internal-timeout + rows: no complete resource/terminal ledger is modeled for those shapes. """ def require(condition: bool, message: str) -> None: @@ -761,6 +781,36 @@ def text(value: Any) -> bool: require(call_kind in REDRIVE_DELIVERY_CALL_KINDS, "the delivery's call_kind is a single call kind") require(not any(key in delivered for key in ("sequence_span", "operation_sequence", "operation_sequence_span")), "the delivery is one single call") + # Validate the whole admitted successor ledger, not just rows at the + # delivery step. A foreign resource, family, mode or out-of-order terminal + # cannot prove that an earlier sequential call finished. + openers: dict[int, dict[str, Any]] = {} + for event in events: + payload = event["payload"] + require(not any(payload.get(key) is not None for key in ( + "parallel_group_id", "selection_group_id", "cancellation_scope_id", + )), "grouped and scoped rows are outside the single-call model") + require(not (event["event_type"].startswith("Timer") and payload.get("timer_kind") in { + "condition_timeout", "signal_timeout", + }), "internal timeout rows are outside the plain-timer model") + kind = event["event_type"] + if kind not in REDRIVE_CANCELLABLE_EVENTS | REDRIVE_RESOLUTION_EVENTS: + continue + sequence = payload["sequence"] + if kind in REDRIVE_CANCELLABLE_EVENTS: + require(sequence not in openers, "one sequential call occupies each successor step") + openers[sequence] = event + continue + opener = openers.get(sequence) + require(opener is not None, "a terminal follows its recorded opener at the same step") + assert opener is not None + opener_kind = _recorded_call_kind(opener) + _, identity, terminals = REDRIVE_CALL_LEDGER[opener_kind] + require(kind in terminals and _recorded_call_kind(event) == opener_kind + and payload.get(identity) == opener["payload"].get(identity) + and (not kind.startswith("Activity") + or payload["activity_type"] == opener["payload"]["activity_type"]), + "a terminal matches its recorded call's family, resource and mode") step = delivered["sequence"] at_step = [event for event in events if event is not delivery and event["payload"].get("sequence") == step and event["event_type"] in REDRIVE_SUCCESSOR_STEP_EVENTS] @@ -781,15 +831,10 @@ def text(value: Any) -> bool: and event["event_type"] in REDRIVE_SUCCESSOR_STEP_EVENTS] next_step = max((event["payload"]["sequence"] for event in before_delivery), default=resume - 1) + 1 require(step == next_step and not at_step, "the delivery targets a successor step the successor reached") - resolutions = { - "ActivityScheduled": ("activity_execution_id", {"ActivityCompleted", "ActivityCancelled"}), - "TimerScheduled": ("timer_id", {"TimerFired"}), - "SignalWaitOpened": ("signal_wait_id", {"SignalApplied"}), - } for earlier in before_delivery: - if earlier["event_type"] not in resolutions: + if earlier["event_type"] not in REDRIVE_CANCELLABLE_EVENTS: continue - identity, kinds = resolutions[earlier["event_type"]] + _, identity, kinds = REDRIVE_CALL_LEDGER[_recorded_call_kind(earlier)] require(any( terminal["sequence"] > earlier["sequence"] and terminal["event_type"] in kinds and terminal["payload"]["sequence"] == earlier["payload"]["sequence"] From b72ecf56ea45ed53560e30ffb30ab6fe87501250 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Thu, 5 Nov 2026 02:47:54 +0000 Subject: [PATCH 2/3] Prepare negative histories before the asserted ledger check --- tests/test_redrive_successor_evidence.py | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/tests/test_redrive_successor_evidence.py b/tests/test_redrive_successor_evidence.py index c5787dc..4d2617b 100644 --- a/tests/test_redrive_successor_evidence.py +++ b/tests/test_redrive_successor_evidence.py @@ -746,11 +746,12 @@ def mutate(history: list[dict[str, Any]]) -> None: _delivery(history)["payload"]["sequence"] = 4 _renumber(history) + history = changed(CANCELLATION, mutate) if state == "resolved": - check(CANCELLATION, changed(CANCELLATION, mutate)) + check(CANCELLATION, history) else: with pytest.raises(AssertionError, match="the successor reached|no opened wait|terminal matches"): - check(CANCELLATION, changed(CANCELLATION, mutate)) + check(CANCELLATION, history) @pytest.mark.parametrize("mutation", ["open", "resolved-after-delivery", "foreign-resolution", "skipped-step"]) @@ -773,8 +774,9 @@ def mutate(history: list[dict[str, Any]]) -> None: _delivery(history)["payload"]["sequence"] = 5 _renumber(history) + history = changed(CANCELLATION, mutate) with pytest.raises(AssertionError, match="the successor reached|terminal matches"): - check(CANCELLATION, changed(CANCELLATION, mutate)) + check(CANCELLATION, history) def test_selection_handle_delivery_remains_outside_the_successor_model() -> None: From 5ce73591aec07e4a9ae3b5fd185fe2b02996fe5f Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Thu, 5 Nov 2026 02:56:29 +0000 Subject: [PATCH 3/3] Refuse repeated terminal rows for a successor call --- tests/test_redrive_call_ledger.py | 36 +++++++++++++++++++++++++++++++ tests/test_replay_clock.py | 3 +++ 2 files changed, 39 insertions(+) diff --git a/tests/test_redrive_call_ledger.py b/tests/test_redrive_call_ledger.py index 1cd2dce..d37eb93 100644 --- a/tests/test_redrive_call_ledger.py +++ b/tests/test_redrive_call_ledger.py @@ -221,3 +221,39 @@ def test_grouped_and_scoped_rows_are_outside_the_model(field: str) -> None: _step_three_activity(doc["history"])["payload"][field] = "outside-scope" with pytest.raises(AssertionError): check_redrive_successor(doc["history"], doc["workflow"]) + + +@pytest.mark.parametrize(("family", "first", "second"), [ + ("activity", "ActivityFailed", "ActivityCompleted"), + ("activity", "ActivityCompleted", "ActivityFailed"), + ("local_activity", "ActivityTimedOut", "ActivityCancelled"), + ("local_activity", "ActivityCancelled", "ActivityTimedOut"), + ("timer", "TimerCancelled", "TimerFired"), + ("timer", "TimerFired", "TimerCancelled"), + ("signal", "SignalApplied", "SignalApplied"), +]) +@pytest.mark.parametrize("identical", [False, True], ids=["contradictory-or-duplicate", "identical-duplicate"]) +def test_one_opener_cannot_have_two_terminal_rows(family: str, first: str, second: str, identical: bool) -> None: + doc = document() + history = doc["history"] + opened = _step_three_activity(history) + if family == "local_activity": + opened["payload"]["local_activity"] = True + elif family == "timer": + opened.update(event_type="TimerScheduled", payload={"sequence": 3, "timer_id": "successor-timer"}) + elif family == "signal": + opened.update(event_type="SignalWaitOpened", payload={ + "sequence": 3, "signal_wait_id": "successor-signal", "signal_name": "wake", + }) + _delivery(history)["payload"]["call_kind"] = family + row = terminal(doc, first, opened=opened) + if family == "signal": + row["payload"].update(signal_id="synthetic-signal", workflow_command_id="synthetic-signal-command") + check_redrive_successor(history, doc["workflow"]) + duplicate = deepcopy(row) + if not identical: + duplicate["event_type"] = second + history.insert(history.index(row) + 1, duplicate) + _renumber(history) + with pytest.raises(AssertionError, match="one terminal"): + check_redrive_successor(history, doc["workflow"]) diff --git a/tests/test_replay_clock.py b/tests/test_replay_clock.py index 258ca66..406e866 100644 --- a/tests/test_replay_clock.py +++ b/tests/test_replay_clock.py @@ -785,6 +785,7 @@ def text(value: Any) -> bool: # delivery step. A foreign resource, family, mode or out-of-order terminal # cannot prove that an earlier sequential call finished. openers: dict[int, dict[str, Any]] = {} + closed: set[int] = set() for event in events: payload = event["payload"] require(not any(payload.get(key) is not None for key in ( @@ -811,6 +812,8 @@ def text(value: Any) -> bool: and (not kind.startswith("Activity") or payload["activity_type"] == opener["payload"]["activity_type"]), "a terminal matches its recorded call's family, resource and mode") + require(sequence not in closed, "a recorded call has at most one terminal") + closed.add(sequence) step = delivered["sequence"] at_step = [event for event in events if event is not delivery and event["payload"].get("sequence") == step and event["event_type"] in REDRIVE_SUCCESSOR_STEP_EVENTS]