Cancel the live activity task after a retry

CHANGELOG.md

@@ -1,5 +1,12 @@
 # Changelog
 
+## 2.13.1 - 2026-10-24
+
+- Cancel the live task when a retried activity receives cooperative cancellation.
+  Preserve completed tasks and snapshot the cancelled attempt's matching task so
+  its worker can acknowledge the stop and repeat that acknowledgement safely.
+- Apply the same task selection when waiting for cancellation completion.
+
 ## 2.13.0 - 2026-10-24
 
 - Recognize bounded `ActivityOutcomeRejected` diagnostics in history pages,

composer.json

@@ -87,7 +87,7 @@
             "dev-main": "2.0.x-dev"
         },
         "durable-workflow": {
-            "product-train": "2.13.0",
+            "product-train": "2.13.1",
             "laravel-embedded-upgrade-contract": "resources/laravel-embedded-upgrade-contract.json",
             "laravel-dependency-security-policy": "resources/laravel-dependency-security-policy.json"
         },

src/V2/Support/ActivityCancellationWait.php

@@ -46,7 +46,8 @@ public static function prepare(WorkflowRun $run, WorkflowTask $workflowTask, Act
         if (in_array($execution->status, [ActivityStatus::Pending, ActivityStatus::Running], true)) {
             $activityTask = $run->tasks->first(static fn (WorkflowTask $task): bool =>
                 $task->task_type === TaskType::Activity
-                && ($task->payload['activity_execution_id'] ?? null) === $execution->id);
+                && ($task->payload['activity_execution_id'] ?? null) === $execution->id
+                && in_array($task->status, [TaskStatus::Ready, TaskStatus::Leased], true));
             $event = ActivityCancellation::record($run, $execution, $activityTask, $context->requestId);
             if ($event !== null) {
                 $run->historyEvents->push($event);

src/V2/Support/WorkflowExecutor.php

@@ -5831,7 +5831,8 @@ private function cancelOpenWaitAtSequence(
                 /** @var WorkflowTask|null $activityTask */
                 $activityTask = $run->tasks->first(
                     static fn (WorkflowTask $candidate): bool => $candidate->task_type === TaskType::Activity
-                        && ($candidate->payload['activity_execution_id'] ?? null) === $execution->id,
+                        && ($candidate->payload['activity_execution_id'] ?? null) === $execution->id
+                        && in_array($candidate->status, [TaskStatus::Ready, TaskStatus::Leased], true),
                 );
                 ActivityCancellation::record(
                     $run,

tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php

@@ -24,8 +24,10 @@
 use Throwable;
 use Workflow\Serializers\Serializer;
 use Workflow\V2\Activity;
+use Workflow\V2\Contracts\ActivityTaskBridge;
 use Workflow\V2\Contracts\HistoryProjectionRole;
 use Workflow\V2\Contracts\WorkflowTaskBridge;
+use Workflow\V2\Enums\CancellationPolicy;
 use Workflow\V2\Enums\HistoryEventType;
 use Workflow\V2\Enums\RunStatus;
 use Workflow\V2\Enums\TaskStatus;
@@ -52,6 +54,7 @@
 use Workflow\V2\Models\WorkflowTask;
 use Workflow\V2\Models\WorkflowTimelineEntry;
 use Workflow\V2\StartOptions;
+use Workflow\V2\Support\ActivityCancellationAcknowledgement;
 use Workflow\V2\Support\CancellationScopeDelivery;
 use Workflow\V2\Support\CancellationScopeHistory;
 use Workflow\V2\Support\CancellationScopeRequests;
@@ -64,6 +67,7 @@
 use Workflow\V2\Support\ReplayDiff;
 use Workflow\V2\Support\RunActivityView;
 use Workflow\V2\Support\RunTimelineProjector;
+use Workflow\V2\Support\WorkflowCommandNormalizer;
 use Workflow\V2\Support\WorkflowFiberRunner;
 use Workflow\V2\Support\WorkflowReplayer;
 use Workflow\V2\Support\WorkflowStep;
@@ -81,6 +85,124 @@ final class V2EmbeddedReplayRegressionCorpusTest extends TestCase
 
     private int $workflowNumber = 0;
 
+    public function testRetriedActivityCancellationCorpusKeepsTheCompletedTaskAndAcknowledgesTheLiveTask(): void
+    {
+        $path = self::FIXTURE_DIR . '/retried-activity-cancellation-live-task.json';
+        if (! is_file($path)) {
+            $this->markTestSkipped('The retained retried-activity cancellation evidence is absent.');
+        }
+        $fixture = json_decode(file_get_contents($path), true, flags: JSON_THROW_ON_ERROR);
+        self::stopWorkers();
+        Queue::fake();
+        config([
+            'queue.default' => 'database',
+            'workflows.v2.compatibility.current' => 'build-a',
+            'workflows.v2.compatibility.supported' => ['build-a'],
+        ]);
+        $this->executeColdReplayFixture($fixture);
+        $scenario = $fixture['retry_cancellation'];
+        $scheduled = collect($fixture['history'])->firstWhere('event_type', 'ActivityScheduled')['payload'];
+        $recordedCancellation = collect($fixture['history'])->firstWhere('event_type', 'ActivityCancelled')['payload'];
+        $bridge = $this->app->make(DefaultWorkflowTaskBridge::class);
+        $activities = $this->app->make(ActivityTaskBridge::class);
+        $lease = static function (WorkflowTask $task): WorkflowTask {
+            $task->forceFill([
+                'status' => TaskStatus::Leased,
+                'lease_owner' => 'portable-worker',
+                'lease_expires_at' => now()
+                    ->addMinutes(5),
+            ])->save();
+            return $task;
+        };
+        foreach ($scenario['policies'] as $policyValue) {
+            $this->clearWorkflowState();
+            $policy = CancellationPolicy::from($policyValue);
+            $stub = WorkflowStub::make(
+                $fixture['workflow']['type'],
+                'corpus-retried-cancellation-' . ++$this->workflowNumber
+            );
+            $stub->start(...$fixture['workflow']['arguments']);
+            $run = WorkflowRun::query()->findOrFail($stub->runId());
+            $initial = $lease($run->tasks()->where('task_type', TaskType::Workflow)->sole());
+            $commands = WorkflowCommandNormalizer::normalize([[
+                'type' => 'schedule_activity',
+                'activity_type' => $scheduled['activity_type'],
+                'arguments' => Serializer::serialize($scenario['activity_arguments']),
+                'cancellation_policy' => $policy->value,
+                'retry_policy' => $scenario['retry_policy'],
+            ]], $fixture['protocol_version']);
+            $this->assertTrue($bridge->complete($initial->id, $commands)['completed']);
+            $execution = $run->activityExecutions()
+                ->sole();
+            $firstTask = $run->tasks()
+                ->where('task_type', TaskType::Activity)->sole();
+            $this->assertTrue($activities->claimStatus($firstTask->id, 'first-owner')['claimed']);
+            $firstAttempt = $execution->attempts()
+                ->sole();
+            $retry = $activities->fail($firstAttempt->id, [
+                'class' => \RuntimeException::class,
+                'message' => 'temporary failure',
+            ]);
+            $this->assertTrue($retry['recorded']);
+            $liveTask = WorkflowTask::query()->findOrFail($retry['next_task_id']);
+            $this->assertTrue($activities->claimStatus($liveTask->id, 'replacement-owner')['claimed']);
+            $liveAttempt = $execution->attempts()
+                ->where('attempt_number', $recordedCancellation['attempt_number'])->sole();
+            $this->assertSame($scenario['completed_task_status'], $firstTask->refresh()->status->value);
+            $completedBefore = $firstTask->getRawOriginal();
+            $this->assertTrue(
+                $stub->requestCancellation('maintenance', $scenario['cleanup_timeout_seconds'])->accepted()
+            );
+            $run->refresh();
+            $deadline = $run->cancellation_deadline_at->toISOString();
+            $ready = static fn (): WorkflowTask => $run->tasks()
+                ->where('task_type', TaskType::Workflow)->where('status', TaskStatus::Ready)->sole();
+            $deliver = static fn (WorkflowTask $task): array => $bridge->deliverCancellation(
+                $task->id,
+                $run->cancellation_request_command_id,
+                $scheduled['sequence'],
+                'activity'
+            );
+            $delivery = $deliver($lease($ready()));
+            $this->assertSame($policy === CancellationPolicy::TryCancel, $delivery['delivered']);
+            if ($policy === CancellationPolicy::WaitCancellationCompleted) {
+                $this->assertTrue($delivery['claim_released']);
+                $this->assertSame('cancellation_waiting_for_activity', $delivery['reason']);
+            }
+            $this->assertSame($completedBefore, $firstTask->refresh()->getRawOriginal());
+            $this->assertSame($scenario['live_task_status'], $liveTask->refresh()->status->value);
+            $this->assertNull($liveTask->lease_expires_at);
+            $cancelled = $run->historyEvents()
+                ->where('event_type', HistoryEventType::ActivityCancelled)->sole();
+            $this->assertSame($liveTask->id, $cancelled->payload['task']['id']);
+            $this->assertSame($liveAttempt->id, $cancelled->payload['activity_attempt']['id']);
+            $accepted = ActivityCancellationAcknowledgement::recordStopped(
+                $liveAttempt->id,
+                'replacement-owner',
+                $run->cancellation_request_command_id
+            );
+            $this->assertTrue($accepted['acknowledged']);
+            $this->assertFalse($accepted['duplicate']);
+            $duplicate = ActivityCancellationAcknowledgement::recordStopped(
+                $liveAttempt->id,
+                'replacement-owner',
+                $run->cancellation_request_command_id
+            );
+            $this->assertTrue($duplicate['acknowledged']);
+            $this->assertTrue($duplicate['duplicate']);
+            $this->assertSame($accepted['history_event_id'], $duplicate['history_event_id']);
+            $this->assertSame(
+                1,
+                $run->historyEvents()
+                    ->where('event_type', HistoryEventType::ActivityCancellationAcknowledged)->count()
+            );
+            if ($policy === CancellationPolicy::WaitCancellationCompleted) {
+                $this->assertTrue($deliver($lease($ready()))['delivered']);
+            }
+            $this->assertSame($deadline, $run->refresh()->cancellation_deadline_at->toISOString());
+        }
+    }
+
     #[RunInSeparateProcess]
     #[PreserveGlobalState(false)]
     public function testCommittedTimerContinuationUsesOneWorkflowConstruction(): void
@@ -187,6 +309,11 @@ public function testFixturesExecuteThroughDeclaredReplayConsumers(): void
         foreach ($this->fixtures() as $fixture) {
             $this->executeColdReplayFixture($fixture);
 
+            if ($fixture['id'] === 'retried-activity-cancellation-live-task') {
+                // Its live retry/delivery proof is exercised by the dedicated corpus test.
+                continue;
+            }
+
             if ($fixture['id'] === 'embedded-committed-timer-continuation') {
                 // Its positive executor path runs in the isolated queue test above.
                 continue;

tests/Fixtures/V2/ReplayRegression/retried-activity-cancellation-live-task.json

@@ -0,0 +1,135 @@
+{
+  "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json",
+  "fixture_schema": "durable-workflow.replay-regression/v1",
+  "id": "retried-activity-cancellation-live-task",
+  "protocol_version": "1.20",
+  "bindings": [
+    "php"
+  ],
+  "consumers": [
+    "workflow-fiber-runner",
+    "workflow-executor"
+  ],
+  "workflow": {
+    "type": "Tests\\Fixtures\\V2\\TestCooperativeActivityCleanupWorkflow",
+    "arguments": [],
+    "payload_codec": "avro"
+  },
+  "history": [
+    {
+      "sequence": 1,
+      "event_type": "WorkflowStarted",
+      "payload": {},
+      "recorded_at": "2026-10-24T01:00:00Z"
+    },
+    {
+      "sequence": 2,
+      "event_type": "ActivityScheduled",
+      "payload": {
+        "sequence": 1,
+        "activity_execution_id": "corpus-activity",
+        "activity_class": "Tests\\Fixtures\\V2\\TestFinallyWorkActivity",
+        "activity_type": "Tests\\Fixtures\\V2\\TestFinallyWorkActivity"
+      },
+      "recorded_at": "2026-10-24T01:00:01Z"
+    },
+    {
+      "sequence": 3,
+      "event_type": "ActivityStarted",
+      "payload": {
+        "sequence": 1,
+        "activity_execution_id": "corpus-activity",
+        "activity_class": "Tests\\Fixtures\\V2\\TestFinallyWorkActivity",
+        "activity_type": "Tests\\Fixtures\\V2\\TestFinallyWorkActivity",
+        "activity_attempt_id": "corpus-attempt-1",
+        "attempt_number": 1
+      },
+      "recorded_at": "2026-10-24T01:00:02Z"
+    },
+    {
+      "sequence": 4,
+      "event_type": "ActivityRetryScheduled",
+      "payload": {
+        "sequence": 1,
+        "activity_execution_id": "corpus-activity",
+        "activity_class": "Tests\\Fixtures\\V2\\TestFinallyWorkActivity",
+        "activity_type": "Tests\\Fixtures\\V2\\TestFinallyWorkActivity",
+        "activity_attempt_id": "corpus-attempt-1"
+      },
+      "recorded_at": "2026-10-24T01:00:03Z"
+    },
+    {
+      "sequence": 5,
+      "event_type": "ActivityStarted",
+      "payload": {
+        "sequence": 1,
+        "activity_execution_id": "corpus-activity",
+        "activity_class": "Tests\\Fixtures\\V2\\TestFinallyWorkActivity",
+        "activity_type": "Tests\\Fixtures\\V2\\TestFinallyWorkActivity",
+        "activity_attempt_id": "corpus-attempt-2",
+        "attempt_number": 2
+      },
+      "recorded_at": "2026-10-24T01:00:04Z"
+    },
+    {
+      "sequence": 6,
+      "event_type": "CooperativeCancellationRequested",
+      "payload": {
+        "workflow_command_id": "corpus-cancel",
+        "command_type": "request_cancellation"
+      },
+      "recorded_at": "2026-10-24T01:00:05Z"
+    },
+    {
+      "sequence": 7,
+      "event_type": "ActivityCancelled",
+      "payload": {
+        "sequence": 1,
+        "activity_execution_id": "corpus-activity",
+        "activity_class": "Tests\\Fixtures\\V2\\TestFinallyWorkActivity",
+        "activity_type": "Tests\\Fixtures\\V2\\TestFinallyWorkActivity",
+        "activity_attempt_id": "corpus-attempt-2",
+        "attempt_number": 2
+      },
+      "recorded_at": "2026-10-24T01:00:06Z"
+    },
+    {
+      "sequence": 8,
+      "event_type": "CooperativeCancellationDelivered",
+      "payload": {
+        "workflow_command_id": "corpus-cancel",
+        "sequence": 1,
+        "call_kind": "activity"
+      },
+      "recorded_at": "2026-10-24T01:00:07Z"
+    }
+  ],
+  "expected": {
+    "completed": false,
+    "result": null,
+    "commands": [
+      {
+        "type": "schedule_activity",
+        "activity_type": "Tests\\Fixtures\\V2\\TestGreetingActivity"
+      }
+    ]
+  },
+  "retry_cancellation": {
+    "policies": [
+      "try_cancel",
+      "wait_cancellation_completed"
+    ],
+    "activity_arguments": [
+      false
+    ],
+    "retry_policy": {
+      "max_attempts": 2,
+      "backoff_seconds": [
+        0
+      ]
+    },
+    "cleanup_timeout_seconds": 60,
+    "completed_task_status": "completed",
+    "live_task_status": "cancelled"
+  }
+}