Bind the redrive successor cancellation to Core's full shape

CHANGELOG.md

@@ -13,6 +13,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Corrected
 
 - **2.13.1 claim withdrawn (#237).** The 2.13.1 entry said "The cancellation clock prefers a reused step's `reused_recorded_at`." As an observable result, that is incorrect. The test behind it was not a history Workflow 2.19 can write: it had no redrive `WorkflowStarted` and delivered the cancellation before the reused step. Redrive writes the successor's start and its reused prefix together, so a successor cancellation always follows the prefix and `remaining()` is computed from that delivery. A reused event's `reused_recorded_at` therefore cannot reach the budget, and the preference changes no observable result. Replay behavior is unchanged. The 2.13.1 start-clock change (`ctx.now()` from `replayed_started_at`) is unaffected. Valid successor fixtures replace the evidence, and the 2.13.1 fixture `redrive-start-replayed-clock`, which has only a partial redrive envelope, is registered as non-qualifying.
+- **Redrive successor evidence is checked against Core's full cancellation shape (#239).** The test check for a successor's own cancellation now requires what Workflow 28893f97 writes: a root `durable-workflow.cancellation-context/v1` context whose lineage is exactly the request's own address, a null `parent_request_id`, `requested_at` no later than the request, `command_type` `request_cancellation` on the request, and a single-call `call_kind` on the delivery that matches the recorded call. A delivery to a call resolved before the request is refused, and so is any event recorded at an unscheduled step the delivery consumed. The unreleased #236 fixture `redrive-reused-upserts-cancellation-clock` has no `command_type`, so it is registered as non-qualifying and `redrive-successor-reused-upserts-cancellation-clock` replaces it. Tests and fixtures only; replay behavior is unchanged.
 
 ### Added
 

tests/fixtures/replay_regressions/redrive-successor-reused-upserts-cancellation-clock.json

@@ -0,0 +1,155 @@
+{
+  "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json",
+  "fixture_schema": "durable-workflow.replay-regression/v1",
+  "id": "redrive-successor-reused-upserts-cancellation-clock",
+  "protocol_version": "1.20",
+  "bindings": [
+    "python"
+  ],
+  "workflow": {
+    "type": "tests.replay.redrive-visibility-cleanup",
+    "input": [
+      "Taylor"
+    ],
+    "payload_codec": "avro",
+    "workflow_id": "regression-corpus-redrive-visibility",
+    "run_id": "01JBREDRIVESUCCESSORRUN0000"
+  },
+  "history": [
+    {
+      "sequence": 1,
+      "event_type": "WorkflowStarted",
+      "recorded_at": "2026-11-03T00:05:00+00:00",
+      "payload": {
+        "workflow_type": "tests.replay.redrive-visibility-cleanup",
+        "workflow_instance_id": "regression-corpus-redrive-visibility",
+        "workflow_run_id": "01JBREDRIVESUCCESSORRUN0000",
+        "workflow_command_id": "01JBREDRIVECOMMAND00000000",
+        "continued_from_run_id": "01JBREDRIVESOURCERUN000000",
+        "resume_step_sequence": 3,
+        "recovery_kind": "redrive",
+        "execution_deadline_at": null,
+        "run_deadline_at": null,
+        "replayed_started_at": "2026-11-03T00:00:00+00:00",
+        "workflow_link_id": "01JBREDRIVELINK00000000000"
+      }
+    },
+    {
+      "sequence": 2,
+      "event_type": "MemoUpserted",
+      "recorded_at": "2026-11-03T00:05:00+00:00",
+      "payload": {
+        "sequence": 1,
+        "entries": {
+          "codec": "avro",
+          "blob": "wwHioz3/VYAiNw4EEGN1c3RvbWVyCgxUYXlsb3IKc3RhZ2UKEGdyZWV0aW5nAA=="
+        },
+        "merged": {
+          "codec": "avro",
+          "blob": "wwHioz3/VYAiNw4EEGN1c3RvbWVyCgxUYXlsb3IKc3RhZ2UKEGdyZWV0aW5nAA=="
+        },
+        "reused_from_run_id": "01JBREDRIVESOURCERUN000000",
+        "reused_recorded_at": "2026-11-03T00:00:00+00:00"
+      }
+    },
+    {
+      "sequence": 3,
+      "event_type": "SearchAttributesUpserted",
+      "recorded_at": "2026-11-03T00:05:00+00:00",
+      "payload": {
+        "sequence": 2,
+        "attributes": {
+          "Stage": "greeting"
+        },
+        "attribute_types": {
+          "Stage": "keyword"
+        },
+        "merged": {
+          "Stage": "greeting"
+        },
+        "reused_from_run_id": "01JBREDRIVESOURCERUN000000",
+        "reused_recorded_at": "2026-11-03T00:00:00+00:00"
+      }
+    },
+    {
+      "sequence": 4,
+      "event_type": "CooperativeCancellationRequested",
+      "recorded_at": "2026-11-03T00:05:02+00:00",
+      "payload": {
+        "workflow_command_id": "redrive-cancel-1",
+        "workflow_instance_id": "regression-corpus-redrive-visibility",
+        "workflow_run_id": "01JBREDRIVESUCCESSORRUN0000",
+        "command_type": "request_cancellation",
+        "reason": "maintenance",
+        "cleanup_deadline_at": "2026-11-03T00:05:30.000000Z",
+        "cancellation": {
+          "schema": "durable-workflow.cancellation-context/v1",
+          "request_id": "redrive-cancel-1",
+          "root_request_id": "redrive-cancel-1",
+          "root_workflow_instance_id": "regression-corpus-redrive-visibility",
+          "root_workflow_run_id": "01JBREDRIVESUCCESSORRUN0000",
+          "parent_request_id": null,
+          "reason": "maintenance",
+          "requester": {
+            "type": "operator",
+            "id": "operator-1",
+            "label": "Maintainer"
+          },
+          "source": "control_plane",
+          "requested_at": "2026-11-03T00:05:02.000000Z",
+          "cleanup_deadline_at": "2026-11-03T00:05:30.000000Z",
+          "lineage": [
+            {
+              "request_id": "redrive-cancel-1",
+              "workflow_instance_id": "regression-corpus-redrive-visibility",
+              "workflow_run_id": "01JBREDRIVESUCCESSORRUN0000"
+            }
+          ]
+        }
+      }
+    },
+    {
+      "sequence": 5,
+      "event_type": "CooperativeCancellationDelivered",
+      "recorded_at": "2026-11-03T00:05:04+00:00",
+      "payload": {
+        "workflow_command_id": "redrive-cancel-1",
+        "workflow_run_id": "01JBREDRIVESUCCESSORRUN0000",
+        "sequence": 3,
+        "call_kind": "activity",
+        "cancellation": {
+          "schema": "durable-workflow.cancellation-context/v1",
+          "request_id": "redrive-cancel-1",
+          "root_request_id": "redrive-cancel-1",
+          "root_workflow_instance_id": "regression-corpus-redrive-visibility",
+          "root_workflow_run_id": "01JBREDRIVESUCCESSORRUN0000",
+          "parent_request_id": null,
+          "reason": "maintenance",
+          "requester": {
+            "type": "operator",
+            "id": "operator-1",
+            "label": "Maintainer"
+          },
+          "source": "control_plane",
+          "requested_at": "2026-11-03T00:05:02.000000Z",
+          "cleanup_deadline_at": "2026-11-03T00:05:30.000000Z",
+          "lineage": [
+            {
+              "request_id": "redrive-cancel-1",
+              "workflow_instance_id": "regression-corpus-redrive-visibility",
+              "workflow_run_id": "01JBREDRIVESUCCESSORRUN0000"
+            }
+          ]
+        }
+      }
+    }
+  ],
+  "expected": {
+    "type": "complete_workflow",
+    "result": {
+      "after_memo": "2026-11-03T00:00:00+00:00",
+      "after_search_attributes": "2026-11-03T00:00:00+00:00",
+      "remaining": 26.0
+    }
+  }
+}

tests/redrive_fixture_registry.py

@@ -28,14 +28,16 @@ class NonQualifying:
     # Added by #236 (unreleased at the time of #237).
     "redrive-reused-upserts-clock": 3,
     "redrive-reused-signal-wait-clock": 2,
-    "redrive-reused-upserts-cancellation-clock": 3,
     "redrive-reused-search-attribute-value-mismatch": 3,
     "redrive-reused-search-attribute-type-mismatch": 3,
     "redrive-reused-search-attribute-update-missing": 3,
     # Added by #237, built to Workflow 28893f97 FailedRunRedrivePlan::successorEvents.
     "redrive-successor-reused-markers": 4,
     "redrive-successor-reused-timer": 3,
     "redrive-successor-reused-cancellation-clock": 3,
+    # Added by #239: redrive-reused-upserts-cancellation-clock with the
+    # request's command_type, which Core WorkflowStub::recordCancellationRequest always writes.
+    "redrive-successor-reused-upserts-cancellation-clock": 3,
 }
 
 NON_QUALIFYING: dict[str, NonQualifying] = {
@@ -51,6 +53,17 @@ class NonQualifying:
             "start clock on a valid successor."
         ),
     ),
+    "redrive-reused-upserts-cancellation-clock": NonQualifying(
+        sha256="f8c40d18fa71a0656089290fd5be44e7df318fafd12721d7a6680e1113ec4856",
+        error="the request's command_type is request_cancellation",
+        replacement="redrive-successor-reused-upserts-cancellation-clock",
+        release="unreleased (#236, after 2.13.1)",
+        reason=(
+            "Its CooperativeCancellationRequested has no command_type. Core WorkflowStub::recordCancellationRequest "
+            "(workflow 28893f97) always writes command_type request_cancellation, so Workflow cannot "
+            "write this request. Its replay result is correct; the replacement adds only that field."
+        ),
+    ),
 }
 
 # Fixtures for a run that redrive starts from. None in this corpus.

tests/test_redrive_successor_evidence.py

@@ -355,7 +355,7 @@
 # The successor cancellation is exactly one request, then exactly one delivery
 # of it, carrying one context bound to the successor, delivered to a call the
 # successor reached. Each mutation below was accepted by the bd41be60 check.
-CANCELLED_SUCCESSORS = [CANCELLATION, "redrive-reused-upserts-cancellation-clock"]
+CANCELLED_SUCCESSORS = [CANCELLATION, "redrive-successor-reused-upserts-cancellation-clock"]
 
 
 def _duplicate(event_type: str) -> Any:
@@ -460,3 +460,154 @@
 def test_a_plain_copied_schedule_may_name_the_source_run(name: str, change: Any) -> None:
     # Core copies the source run's schedule payload; a null key is plain.
     check(name, changed(name, change))
+
+
+# #239: the successor cancellation as Core 28893f97 writes it. Each mutation
+# below was accepted by the 262eb6ef check and is refused here.
+def _step_three_activity(history: list[dict[str, Any]]) -> dict[str, Any]:
+    """The successor's step-3 ActivityScheduled, inserted before the request if absent."""
+    scheduled = [item for item in history if item["event_type"] == "ActivityScheduled"]
+    if scheduled:
+        return scheduled[0]
+    item = {"event_type": "ActivityScheduled", "recorded_at": _request(history)["recorded_at"],
+            "payload": {"activity_execution_id": "01JBREDRIVESUCCESSORACT001",
+                        "activity_type": "tests.redrive.second", "sequence": 3}}
+    history.insert(history.index(_request(history)), item)
+    _renumber(history)
+    return item
+
+
+def _completed_before_request(history: list[dict[str, Any]]) -> None:
+    scheduled = _step_three_activity(history)
+    completed = {"event_type": "ActivityCompleted", "recorded_at": _request(history)["recorded_at"],
+                 "payload": {**scheduled["payload"], "result": "wwHioz3/VYAiNwoEb2s=", "payload_codec": "avro"}}
+    history.insert(history.index(scheduled) + 1, completed)
+    _renumber(history)
+
+
+def _scheduled_after_delivery(history: list[dict[str, Any]]) -> None:
+    scheduled = _step_three_activity(history)
+    history.remove(scheduled)
+    scheduled["recorded_at"] = _delivery(history)["recorded_at"]
+    history.insert(history.index(_delivery(history)) + 1, scheduled)
+    _renumber(history)
+
+
+def _source_run_prepended(context: dict[str, Any]) -> None:
+    context["lineage"].insert(0, {"request_id": "source-request", "workflow_instance_id":
+                                  context["root_workflow_instance_id"], "workflow_run_id": SOURCE_RUN})
+
+
+CORE_CANCELLATION_MUTATIONS = {
+    # 1. Root binding: a parent no longer skips the root check.
+    "parent-with-another-root": (_context("both", lambda context: context.update(
+        parent_request_id="parent-request", root_request_id="parent-request")), "its own root"),
+    "parent-request-id-missing": (_context("both", lambda context: context.pop("parent_request_id")), "its own root"),
+    # 2. Lineage: the whole lineage is the request's own address.
+    "lineage-source-run-prepended": (_context("both", _source_run_prepended), "rooted at the successor run"),
+    "lineage-entry-repeated": (_context("both", lambda context: context["lineage"].append(
+        dict(context["lineage"][0]))), "rooted at the successor run"),
+    # 3. Delivery step.
+    "delivery-to-a-call-completed-before-the-request": (_completed_before_request, "resolved before its request"),
+    "schedule-after-a-delivery-to-the-resume-step": (_scheduled_after_delivery, "the successor reached"),
+    # 4. Fields Core always writes.
+    "request-command-type-missing": (lambda history: _request(history)["payload"].pop("command_type"),
+                                     "command_type"),
+    "request-command-type-other": (lambda history: _request(history)["payload"].__setitem__(
+        "command_type", "request_termination"), "command_type"),
+    "delivery-call-kind-missing": (lambda history: _delivery(history)["payload"].pop("call_kind"), "call_kind"),
+    "delivery-call-kind-unknown": (lambda history: _delivery(history)["payload"].__setitem__(
+        "call_kind", "workflow"), "call_kind"),
+    "delivery-call-kind-parallel": (lambda history: _delivery(history)["payload"].__setitem__(
+        "call_kind", "parallel"), "call_kind"),
+    "delivery-sequence-span": (lambda history: _delivery(history)["payload"].__setitem__("sequence_span", 2),
+                               "one single call"),
+    "context-schema-missing": (_context("both", lambda context: context.pop("schema")), "schema"),
+    "context-schema-unknown": (_context("both", lambda context: context.__setitem__(
+        "schema", "durable-workflow.cancellation-context/v3")), "schema"),
+    "context-schema-scoped": (_context("both", lambda context: context.__setitem__(
+        "schema", "durable-workflow.cancellation-context/v2")), "schema"),
+    "context-scope-origin": (_context("both", lambda context: context.__setitem__("scope_origin", {})), "schema"),
+    "context-requested-at-missing": (_context("both", lambda context: context.pop("requested_at")),
+                                     "history time must be a string"),
+    "context-requested-after-the-request": (_context("both", lambda context: context.__setitem__(
+        "requested_at", "2026-11-03T00:05:05.000000Z")), "requested_at"),
+    "context-requested-before-the-start": (_context("both", lambda context: context.__setitem__(
+        "requested_at", "2026-11-03T00:04:59.000000Z")), "requested_at"),
+    "context-deadline-at-the-request": (lambda history: [
+        target.__setitem__("cleanup_deadline_at", _request(history)["payload"]["cancellation"]["requested_at"])
+        for target in (_request(history)["payload"], _request(history)["payload"]["cancellation"],
+                       _delivery(history)["payload"]["cancellation"])], "cleanup deadline"),
+    "context-requester-empty": (_context("both", lambda context: context.__setitem__("requester", {})),
+                                "requester"),
+    "context-requester-unknown-key": (_context("both", lambda context: context["requester"].__setitem__(
+        "email", "operator@example.test")), "requester"),
+    "context-source-blank": (_context("both", lambda context: context.__setitem__("source", " ")), "source"),
+    "request-reason-differs": (lambda history: _request(history)["payload"].__setitem__("reason", "other"),
+                               "reason"),
+}
+
+
+@pytest.mark.parametrize("name", CANCELLED_SUCCESSORS)
+@pytest.mark.parametrize("mutation", sorted(CORE_CANCELLATION_MUTATIONS))
+def test_the_successor_cancellation_has_core_shape(name: str, mutation: str) -> None:
+    change, message = CORE_CANCELLATION_MUTATIONS[mutation]
+    with pytest.raises(AssertionError, match=message):
+        check(name, changed(name, change))
+
+
+def test_a_delivery_to_a_recorded_call_names_its_kind() -> None:
+    with pytest.raises(AssertionError, match="the recorded call's kind"):
+        check(CANCELLATION, changed(CANCELLATION, lambda history: _delivery(history)["payload"].__setitem__(
+            "call_kind", "timer")))
+
+
+def _local_activity(call_kind: str) -> Any:
+    def change(history: list[dict[str, Any]]) -> None:
+        event(history, "ActivityScheduled")["payload"]["local_activity"] = True
+        _delivery(history)["payload"]["call_kind"] = call_kind
+
+    return change
+
+
+def test_a_delivery_to_a_recorded_local_activity_names_local_activity() -> None:
+    # Core callKindAt: an ActivityScheduled with local_activity true is a
+    # local_activity call. b9210162 refused this valid delivery.
+    check(CANCELLATION, changed(CANCELLATION, _local_activity("local_activity")))
+
+
+def test_an_activity_delivery_to_a_recorded_local_activity_is_refused() -> None:
+    # b9210162 accepted this delivery, which Core refuses.
+    with pytest.raises(AssertionError, match="the recorded call's kind"):
+        check(CANCELLATION, changed(CANCELLATION, _local_activity("activity")))
+
+
+def test_a_local_activity_delivery_to_a_recorded_activity_is_refused() -> None:
+    with pytest.raises(AssertionError, match="the recorded call's kind"):
+        check(CANCELLATION, changed(CANCELLATION, lambda history: _delivery(history)["payload"].__setitem__(
+            "call_kind", "local_activity")))
+
+
+@pytest.mark.parametrize("name", CANCELLED_SUCCESSORS)
+def test_a_whitespace_requester_value_is_accepted_as_core_does(name: str) -> None:
+    # Core CancellationContext::fromArray refuses only an empty string.
+    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.
+    name = "redrive-successor-reused-upserts-cancellation-clock"
+    check(name, changed(name, lambda history: _delivery(history)["payload"].__setitem__("call_kind", call_kind)))
+
+
+def test_a_call_completed_after_the_request_is_still_deliverable() -> None:
+    # Core resolvedBeforeRequest: only a resolution before the request refuses it.
+    def completed_between(history: list[dict[str, Any]]) -> None:
+        scheduled = _step_three_activity(history)
+        completed = {"event_type": "ActivityCompleted", "recorded_at": _delivery(history)["recorded_at"],
+                     "payload": {**scheduled["payload"], "result": "wwHioz3/VYAiNwoEb2s=", "payload_codec": "avro"}}
+        history.insert(history.index(_delivery(history)), completed)
+        _renumber(history)
+
+    check(CANCELLATION, changed(CANCELLATION, completed_between))