Report a redrive link as typed lineage

CHANGELOG.md

@@ -31,6 +31,18 @@
   null one. Before,
   redriving such a successor was refused with `mismatched_signal_receipt`
   (#864).
+- Run detail and describe report a redrive link as typed lineage. The
+  successor's `parents` and the failed run's `continuedWorkflows` show
+  `link_type: redrive` with `history_authority: typed_history` and
+  `diagnostic_only: false`, because the successor's `WorkflowStarted` records
+  `recovery_kind: redrive`, `continued_from_run_id` and `workflow_link_id`.
+  Both sides take the step (`resume_step_sequence`) and time from that event,
+  not from the link row. The start must name the successor's own run and
+  instance, an earlier run of the same instance as the source, a nonblank link
+  id and a positive resume step, and exactly one redrive link row must join
+  that source to the successor under that link id. Before, both sides showed
+  a diagnostic-only fallback. A link the successor's history does not confirm stays a single
+  diagnostic fallback (#864).
 
 ## 2.19.0 - 2026-11-03
 

src/V2/Support/RunLineageView.php

@@ -12,7 +12,7 @@
 use Workflow\V2\Models\WorkflowLink;
 use Workflow\V2\Models\WorkflowRun;
 
 final class RunLineageView
 {
     /**
      * @return list<array<string, mixed>>
@@ -24,7 +24,11 @@
         $entries = [];
         $seen = [];
 
-        foreach ([self::continuedParentEntry($run), self::childParentEntry($run)] as $entry) {
+        foreach ([
+            self::continuedParentEntry($run),
+            self::redriveParentEntry($run),
+            self::childParentEntry($run),
+        ] as $entry) {
             if ($entry !== null) {
                 $entries[] = $entry;
                 $seen[self::key((string) $entry['link_type'], (string) $entry['workflow_run_id'])] = true;
@@ -76,7 +80,7 @@
     /**
      * @return list<array<string, mixed>>
      */
     public static function continuedWorkflowsForRun(WorkflowRun $run): array
     {
         $run->loadMissing(['historyEvents', 'childLinks', 'instance.runs.summary']);
 
@@ -115,6 +119,11 @@
             }
         }
 
+        foreach (self::redriveSuccessorEntries($run) as $entry) {
+            $entries[] = $entry;
+            $seen[self::key('redrive', (string) $entry['workflow_run_id'])] = true;
+        }
+
         foreach (self::childSequences($run) as $sequence) {
             $childRun = ChildRunHistory::childRunForSequence($run, $sequence);
             if ($childRun === null && ImportedLineageReferences::imported($run)) {
@@ -277,6 +286,160 @@
         return null;
     }
 
+    /**
+     * A redrive successor names the failed run it continues in its own
+     * WorkflowStarted, so the link is typed history, not a fallback.
+     *
+     * @return array<string, mixed>|null
+     */
+    private static function redriveParentEntry(WorkflowRun $run): ?array
+    {
+        $start = self::redriveStart($run);
+
+        if ($start === null) {
+            return null;
+        }
+
+        // The source is an earlier run of this instance, and exactly one
+        // redrive link joins it to this run under the recorded link id.
+        $source = self::resolveRun($run, $start['source_run_id']);
+
+        if (
+            $source === null
+            || $source->workflow_instance_id !== $run->workflow_instance_id
+            || (int) $source->run_number >= (int) $run->run_number
+        ) {
+            return null;
+        }
+
+        $links = $run->parentLinks->filter(
+            static fn (mixed $link): bool => $link instanceof WorkflowLink
+                && $link->link_type === 'redrive'
+                && $link->id === $start['link_id']
+                && $link->parent_workflow_run_id === $source->id
+                && $link->parent_workflow_instance_id === $run->workflow_instance_id
+                && $link->child_workflow_run_id === $run->id
+                && $link->child_workflow_instance_id === $run->workflow_instance_id
+        );
+
+        if ($links->count() !== 1) {
+            return null;
+        }
+
+        return self::entry(
+            id: $start['link_id'],
+            linkType: 'redrive',
+            sequence: $start['resume_step_sequence'],
+            isPrimaryParent: true,
+            relatedRun: $source,
+            instanceId: $run->workflow_instance_id,
+            runId: $start['source_run_id'],
+            createdAt: $start['created_at'],
+            historyAuthority: ChildRunHistory::HISTORY_AUTHORITY_TYPED,
+        );
+    }
+
+    /**
+     * The failed run's history is closed, so its redrive successor is typed
+     * only when the successor's WorkflowStarted names this run and this link.
+     * The entry takes its step and time from that event, not the link row.
+     *
+     * @return list<array<string, mixed>>
+     */
+    private static function redriveSuccessorEntries(WorkflowRun $run): array
+    {
+        $entries = [];
+
+        foreach ($run->childLinks as $link) {
+            if (
+                ! $link instanceof WorkflowLink
+                || $link->link_type !== 'redrive'
+                || $link->parent_workflow_run_id !== $run->id
+                || $link->parent_workflow_instance_id !== $run->workflow_instance_id
+                || $link->child_workflow_instance_id !== $run->workflow_instance_id
+            ) {
+                continue;
+            }
+
+            $successor = self::resolveRun($run, $link->child_workflow_run_id);
+            $start = $successor === null ? null : self::redriveStart($successor);
+
+            if (
+                $successor === null
+                || $start === null
+                || $successor->id !== $link->child_workflow_run_id
+                || $successor->workflow_instance_id !== $run->workflow_instance_id
+                || (int) $successor->run_number <= (int) $run->run_number
+                || $start['source_run_id'] !== $run->id
+                || $start['link_id'] !== $link->id
+            ) {
+                continue;
+            }
+
+            $entries[] = self::entry(
+                id: $link->id,
+                linkType: 'redrive',
+                sequence: $start['resume_step_sequence'],
+                isPrimaryParent: true,
+                relatedRun: $successor,
+                instanceId: $run->workflow_instance_id,
+                runId: $successor->id,
+                relationPrefix: 'child',
+                createdAt: $start['created_at'],
+                historyAuthority: ChildRunHistory::HISTORY_AUTHORITY_TYPED,
+            );
+        }
+
+        return $entries;
+    }
+
+    /**
+     * The typed redrive identity a run's WorkflowStarted records: it names
+     * this exact run and instance, the failed source, the redrive link and the
+     * step the run resumes at. Anything less is not typed lineage.
+     *
+     * @return array{source_run_id: string, link_id: string, resume_step_sequence: int, created_at: CarbonInterface|null}|null
+     */
+    private static function redriveStart(WorkflowRun $run): ?array
+    {
+        $started = self::workflowStartedEvent($run);
+        $payload = $started?->payload;
+
+        if (! is_array($payload) || ($payload['recovery_kind'] ?? null) !== 'redrive') {
+            return null;
+        }
+
+        $sourceRunId = self::nonBlank($payload['continued_from_run_id'] ?? null);
+        $linkId = self::nonBlank($payload['workflow_link_id'] ?? null);
+        $resume = $payload['resume_step_sequence'] ?? null;
+
+        if (
+            ($payload['workflow_run_id'] ?? null) !== $run->id
+            || ($payload['workflow_instance_id'] ?? null) !== $run->workflow_instance_id
+            || $sourceRunId === null
+            || $sourceRunId === $run->id
+            || $linkId === null
+            || ! is_int($resume)
+            || $resume < 1
+        ) {
+            return null;
+        }
+
+        return [
+            'source_run_id' => $sourceRunId,
+            'link_id' => $linkId,
+            'resume_step_sequence' => $resume,
+            'created_at' => $started->recorded_at ?? $started->created_at,
+        ];
+    }
+
+    private static function nonBlank(mixed $value): ?string
+    {
+        return is_string($value) && trim($value) !== ''
+            ? $value
+            : null;
+    }
+
     /**
      * @return array<string, mixed>|null
      */

tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php

@@ -94,6 +94,7 @@
 use Workflow\V2\Support\ReplayDiff;
 use Workflow\V2\Support\RunActivityView;
 use Workflow\V2\Support\RunDetailView;
+use Workflow\V2\Support\RunLineageView;
 use Workflow\V2\Support\RunTimelineProjector;
 use Workflow\V2\Support\ScheduleManager;
 use Workflow\V2\Support\ServiceCallPrincipal;
@@ -638,6 +639,52 @@ public function testRedriveReusedUpdateCorpusReappliesItsStateChange(): void
         }
     }
 
+    public function testRedriveTypedLineageCorpusLinksBothRunsFromHistory(): void
+    {
+        $path = self::FIXTURE_DIR . '/redrive-typed-lineage.json';
+        if (! is_file($path)) {
+            $this->markTestSkipped('The redrive typed-lineage evidence is absent.');
+        }
+        $fixture = json_decode(file_get_contents($path), true, flags: JSON_THROW_ON_ERROR);
+        $scenario = $fixture['redrive_typed_lineage'];
+        self::stopWorkers();
+        config([
+            'queue.default' => 'sync',
+            'queue.connections.sync.driver' => 'sync',
+        ]);
+        Cache::forget('test:redrive:first-calls');
+        Cache::forget('test:redrive:second-calls');
+        $this->clearWorkflowState();
+        WorkflowStub::fake();
+
+        $stub = WorkflowStub::make($fixture['workflow']['type'], 'regression-corpus-redrive-lineage');
+        $stub->start(...$fixture['workflow']['arguments']);
+        $this->assertTrue($stub->refresh()->failed());
+        $source = WorkflowRun::query()->findOrFail($stub->runId());
+        $result = app(WorkflowControlPlane::class)->redrive('regression-corpus-redrive-lineage', $source->id);
+        $this->assertTrue($result['accepted'], (string) $result['reason']);
+        $successor = WorkflowRun::query()->findOrFail($result['workflow_run_id']);
+
+        foreach ([
+            [RunLineageView::parentsForRun($successor->fresh()), $source->id, $scenario['source_run_number']],
+            [
+                RunLineageView::continuedWorkflowsForRun($source->fresh()),
+                $successor->id,
+                $scenario['successor_run_number'],
+            ],
+        ] as [$entries, $runId, $runNumber]) {
+            $redrive = array_values(array_filter(
+                $entries,
+                static fn (array $entry): bool => $entry['link_type'] === $scenario['link_type'],
+            ));
+            $this->assertCount(1, $redrive);
+            $this->assertSame($runId, $redrive[0]['workflow_run_id']);
+            $this->assertSame($runNumber, $redrive[0]['run_number']);
+            $this->assertSame($scenario['history_authority'], $redrive[0]['history_authority']);
+            $this->assertSame($scenario['diagnostic_only'], $redrive[0]['diagnostic_only']);
+        }
+    }
+
     public function testRedriveReusedTimerCorpusDoesNotWaitAgain(): void
     {
         $path = self::FIXTURE_DIR . '/redrive-reused-timer-clock.json';