Allow redrive through validated activity retry histories

CHANGELOG.md

@@ -2,6 +2,14 @@
 
 ## Unreleased
 
+- Failed-run redrive accepts a verified linear activity retry chain. A retried
+  completed prefix is reused and only the failed boundary step runs again.
+  Substituted attempt/task identities refuse with
+  `invalid_activity_retry_history`; existing service-worker definition checks
+  still apply. An exactly bound accepted start/adopt receipt is harmless, while
+  rejected or mismatched starts remain unsupported. This core correction is
+  separate from the experimental Saga compensation scope foundation (#1063).
+
 - **Run migrations.** Opt-in deadline-triggered cleanup (#863, experimental,
   embedded). A start can reserve the last part of its hard execution or run
   timeout for cleanup with `StartOptions::withTimeoutCleanup($seconds)` (or

regression-corpus-policy.json

@@ -74,6 +74,12 @@
             "history_sequence",
             "history_payload"
           ]
+        },
+        {
+          "glob": "src/V2/Support/FailedRunRedrivePlan.php"
+        },
+        {
+          "glob": "src/V2/Support/ActivityRetryHistory.php"
         }
       ]
     }

src/V2/Support/ActivityRetryHistory.php

@@ -0,0 +1,140 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Workflow\V2\Support;
+
+use Workflow\V2\Enums\HistoryEventType;
+use Workflow\V2\Models\WorkflowHistoryEvent;
+
+/** @internal Validates recorded retry lineage before reusing or resuming an activity. */
+final class ActivityRetryHistory
+{
+    /**
+     * @param array<WorkflowHistoryEvent> $events
+     */
+    public static function valid(array $events): bool
+    {
+        $hasRetries = false;
+        foreach ($events as $event) {
+            $hasRetries = $hasRetries || $event->event_type === HistoryEventType::ActivityRetryScheduled;
+        }
+        if (! $hasRetries) {
+            return true;
+        }
+        $phase = 'schedule';
+        $execution = null;
+        $attempt = null;
+        $task = null;
+        $number = 0;
+        $nextTask = null;
+        $seenAttempts = [];
+        $seenTasks = [];
+        foreach ($events as $event) {
+            $payload = $event->payload;
+            $id = $payload['activity_execution_id'] ?? null;
+            if (
+                ! is_string($id)
+                || $id === ''
+                || ($execution !== null && $id !== $execution)
+                || $phase === 'terminal'
+            ) {
+                return false;
+            }
+            $execution = $id;
+            if ($event->event_type === HistoryEventType::ActivityScheduled) {
+                if ($phase !== 'schedule') {
+                    return false;
+                }
+                $phase = 'start';
+            } elseif ($event->event_type === HistoryEventType::ActivityStarted) {
+                $currentAttempt = $payload['activity_attempt_id'] ?? null;
+                $currentTask = $event->workflow_task_id ?? ($payload['workflow_task_id'] ?? null);
+                if (! in_array($phase, ['start', 'retry'], true)
+                    || ! is_string($currentAttempt) || $currentAttempt === '' || isset($seenAttempts[$currentAttempt])
+                    || ! is_string($currentTask) || $currentTask === '' || isset($seenTasks[$currentTask])
+                    || ($payload['attempt_number'] ?? null) !== $number + 1
+                    || ($number > 0 && $nextTask !== $currentTask)
+                    || ! self::matchesTask($event, $currentTask)) {
+                    return false;
+                }
+                $attempt = $currentAttempt;
+                $task = $currentTask;
+                $seenAttempts[$attempt] = true;
+                $seenTasks[$task] = true;
+                ++$number;
+                $nextTask = null;
+                $phase = 'active';
+            } elseif ($event->event_type === HistoryEventType::ActivityRetryScheduled) {
+                $retry = $payload['retry_task_id'] ?? null;
+                $maximum = $payload['max_attempts'] ?? null;
+                if ($phase !== 'active'
+                    || ($payload['activity_attempt_id'] ?? null) !== $attempt
+                    || ($payload['retry_after_attempt_id'] ?? null) !== $attempt
+                    || ($payload['retry_after_attempt'] ?? null) !== $number
+                    || ($payload['retry_of_task_id'] ?? null) !== $task
+                    || ! is_string($retry) || $retry === '' || isset($seenTasks[$retry])
+                    || ($maximum !== null && (! is_int($maximum) || $maximum <= $number))
+                    || ! self::matchesTask($event, $task)) {
+                    return false;
+                }
+                $nextTask = $retry;
+                $phase = 'retry';
+            } elseif (in_array($event->event_type, [
+                HistoryEventType::ActivityHeartbeatRecorded,
+                HistoryEventType::ActivityCompleted,
+                HistoryEventType::ActivityFailed,
+            ], true)) {
+                if ($phase !== 'active' || ($payload['activity_attempt_id'] ?? null) !== $attempt
+                    || ($payload['attempt_number'] ?? null) !== $number
+                    || ! self::matchesTask($event, $task)) {
+                    return false;
+                }
+                if ($event->event_type !== HistoryEventType::ActivityHeartbeatRecorded) {
+                    $phase = 'terminal';
+                }
+            } else {
+                return false;
+            }
+            if (in_array($phase, ['active', 'retry', 'terminal'], true)
+                && ! self::matchesAttemptSnapshot($payload, $execution, $attempt, $number, $task)) {
+                return false;
+            }
+        }
+        return $phase === 'terminal';
+    }
+
+    private static function matchesTask(WorkflowHistoryEvent $event, string $task): bool
+    {
+        $payload = $event->payload;
+        return ($event->workflow_task_id === null || $event->workflow_task_id === $task)
+            && (! array_key_exists('workflow_task_id', $payload) || $payload['workflow_task_id'] === $task);
+    }
+
+    private static function matchesAttemptSnapshot(
+        array $payload,
+        string $execution,
+        string $attempt,
+        int $number,
+        string $task,
+    ): bool {
+        if (! array_key_exists('activity_attempt', $payload)) {
+            return true;
+        }
+        $snapshot = $payload['activity_attempt'];
+        if (! is_array($snapshot)) {
+            return false;
+        }
+        foreach ([
+            'id' => $attempt,
+            'activity_execution_id' => $execution,
+            'attempt_number' => $number,
+            'task_id' => $task,
+        ] as $field => $expected) {
+            if (array_key_exists($field, $snapshot) && $snapshot[$field] !== $expected) {
+                return false;
+            }
+        }
+        return true;
+    }
+}

src/V2/Support/FailedRunRedrivePlan.php

@@ -7,6 +7,9 @@
 use Illuminate\Support\Str;
 use Throwable;
 use Workflow\Serializers\Serializer;
+use Workflow\V2\Enums\CommandOutcome;
+use Workflow\V2\Enums\CommandStatus;
+use Workflow\V2\Enums\CommandType;
 use Workflow\V2\Enums\HistoryEventType;
 use Workflow\V2\Enums\RunStatus;
 use Workflow\V2\Models\WorkflowHistoryEvent;
@@ -86,7 +89,7 @@ public static function forRun(WorkflowRun $run): array
             }
 
             if ($event->event_type === HistoryEventType::StartAccepted) {
-                if ($started !== null) {
+                if ($started !== null && ! self::isActiveStartAdoption($run, $event)) {
                     return self::reject('unexpected_start');
                 }
 
@@ -199,6 +202,7 @@ public static function forRun(WorkflowRun $run): array
             if (! in_array($event->event_type, [
                 HistoryEventType::ActivityScheduled,
                 HistoryEventType::ActivityStarted,
+                HistoryEventType::ActivityRetryScheduled,
                 HistoryEventType::ActivityHeartbeatRecorded,
                 HistoryEventType::ActivityCompleted,
                 HistoryEventType::ActivityFailed,
@@ -381,6 +385,10 @@ public static function forRun(WorkflowRun $run): array
                 return self::reject('activity_sequence_gap');
             }
 
+            if (! ActivityRetryHistory::valid($stepEvents)) {
+                return self::reject('invalid_activity_retry_history');
+            }
+
             $activityType = null;
             $completion = null;
             $failureCount = 0;
@@ -612,6 +620,38 @@ public static function successorEvents(array $plan, string $sourceRunId): array
         return $events;
     }
 
+    /**
+     * Native start/adopt records a command receipt, not another workflow start or reusable step.
+     * Only accepted adoption proves the same resolved run and command identity.
+     * A rejected start is not adoption evidence and retains its existing refusal.
+     */
+    private static function isActiveStartAdoption(WorkflowRun $run, WorkflowHistoryEvent $event): bool
+    {
+        $payload = $event->payload;
+        $command = $payload['command'] ?? null;
+        if (! is_array($command) || ($payload['outcome'] ?? null) !== CommandOutcome::ReturnedExistingActive->value
+            || ($payload['rejection_reason'] ?? null) !== null
+            || $event->workflow_run_id !== $run->id) {
+            return false;
+        }
+        foreach (['workflow_instance_id', 'workflow_run_id', 'workflow_class', 'workflow_type'] as $key) {
+            $expected = $key === 'workflow_run_id' ? $run->id : $run->getAttribute($key);
+            if (! is_string($expected) || $expected === '' || ($payload[$key] ?? null) !== $expected) {
+                return false;
+            }
+        }
+        $commandId = $payload['workflow_command_id'] ?? null;
+        return is_string($commandId) && $commandId !== ''
+            && $event->workflow_command_id === $commandId
+            && ($command['id'] ?? null) === $commandId
+            && ($command['type'] ?? null) === CommandType::Start->value
+            && ($command['target_scope'] ?? null) === 'instance'
+            && ($command['resolved_run_id'] ?? null) === $run->id
+            && ($command['status'] ?? null) === CommandStatus::Accepted->value
+            && ($command['outcome'] ?? null) === CommandOutcome::ReturnedExistingActive->value
+            && ($command['rejection_reason'] ?? null) === null;
+    }
+
     /**
      * Who may have recorded a reused-step payload in this run's history: the
      * run itself, or, for a step it reused, its immediate redrive source while