rmcdaniel · Nov 3, 2026 · Nov 3, 2026 · Nov 3, 2026 · Nov 3, 2026 · Nov 3, 2026
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -2,6 +2,17 @@
 
 ## Unreleased
 
+- Redrive a failed run whose workflow code upserted memos or search
+  attributes. Before, any `MemoUpserted` or `SearchAttributesUpserted` event
+  refused redrive with `unsupported_history`. The successor reuses those
+  events in step order, carrying `reused_from_run_id` and
+  `reused_recorded_at`, and replay uses the original time. Memo and search
+  attribute rows now carry to the successor unchanged, keeping the step that
+  wrote them and whether they were inherited, so a successor that fails again
+  can be redriven again. A row written by workflow code with no matching
+  reused upsert step is still refused with `unsupported_visibility_metadata`
+  (#864).
+
 ## 2.18.0 - 2026-11-03
 
 - Redrive a failed run that has start-time memos or search attributes. Before,

diff --git a/docs/api-stability.md b/docs/api-stability.md
@@ -774,8 +774,8 @@ class, source file path, or stack trace to replay or handle a failure.
 | `ServiceCallCompleted` | `sequence`, `service_call_id`, `workflow_instance_id`, `workflow_run_id`, `caller_workflow_instance_id`, `caller_workflow_run_id`, `caller_sdk_language`, `endpoint_name`, `service_name`, `operation_name`, `service_sdk_language`, `request_payload`, `response_payload`, `payload_codec`, `operation_mode`, `wait_for`, `status`, `outcome`, `resolved_binding_kind`, `resolved_target_reference`, `linked_workflow_instance_id`, `linked_workflow_run_id`, `linked_workflow_update_id`, `service_call`, `response_or_failure_surface` | `WorkflowStepHistory`, `WorkflowExecutor`, `WorkflowFiberRunner`, `QueryStateReplayer`, Nexus observability |
 | `ServiceCallFailed` | `sequence`, `service_call_id`, `workflow_instance_id`, `workflow_run_id`, `caller_workflow_instance_id`, `caller_workflow_run_id`, `caller_sdk_language`, `endpoint_name`, `service_name`, `operation_name`, `service_sdk_language`, `request_payload`, `response_payload`, `payload_codec`, `operation_mode`, `wait_for`, `status`, `outcome`, `resolved_binding_kind`, `resolved_target_reference`, `linked_workflow_instance_id`, `linked_workflow_run_id`, `linked_workflow_update_id`, `service_call`, `response_or_failure_surface`, `exception_type`, `exception_class`, `message`, `code`, `exception` | `WorkflowStepHistory`, `WorkflowExecutor`, `WorkflowFiberRunner`, `QueryStateReplayer`, typed service failure propagation |
 | `ServiceCallCancelled` | `sequence`, `service_call_id`, `workflow_instance_id`, `workflow_run_id`, `caller_workflow_instance_id`, `caller_workflow_run_id`, `caller_sdk_language`, `endpoint_name`, `service_name`, `operation_name`, `service_sdk_language`, `request_payload`, `response_payload`, `payload_codec`, `operation_mode`, `wait_for`, `status`, `outcome`, `resolved_binding_kind`, `resolved_target_reference`, `linked_workflow_instance_id`, `linked_workflow_run_id`, `linked_workflow_update_id`, `service_call`, `response_or_failure_surface`, `exception_type`, `exception_class`, `message`, `code`, `exception` | `WorkflowStepHistory`, `WorkflowExecutor`, `WorkflowFiberRunner`, `QueryStateReplayer`, typed service failure propagation |
-| `SearchAttributesUpserted` | `sequence`, `attributes`, `merged` | `WorkflowStepHistory`, `HistoryTimeline`, visibility/search projections, history export |
-| `MemoUpserted` | `sequence`, `entries`, `merged` | `WorkflowStepHistory`, `HistoryTimeline`, run detail projections, history export |
+| `SearchAttributesUpserted` | `sequence`, `attributes`, `attribute_types`, `merged`, `reused_from_run_id`, `reused_recorded_at` | `WorkflowStepHistory`, `WorkflowExecutor`, `QueryStateReplayer`, `WorkflowFiberRunner`, `HistoryTimeline`, visibility/search projections, history export, `RedriveReplayClock` |
+| `MemoUpserted` | `sequence`, `entries`, `merged`, `reused_from_run_id`, `reused_recorded_at` | `WorkflowStepHistory`, `WorkflowExecutor`, `QueryStateReplayer`, `WorkflowFiberRunner`, `HistoryTimeline`, run detail projections, history export, `RedriveReplayClock` |
 | `RepairRequested` | `workflow_command_id`, `workflow_instance_id`, `workflow_run_id`, `command_type`, `outcome`, `liveness_state`, `wait_kind`, `task_id`, `task_type` | `HistoryTimeline`, `RunCommandContract`, repair diagnostics, operator detail projections |
 | `CancelRequested` | `workflow_command_id`, `workflow_instance_id`, `workflow_run_id`, `command_type`, `reason` | `HistoryTimeline`, `RunCommandContract`, cancellation projections |
 | `WorkflowCancelled` | `workflow_command_id`, `workflow_instance_id`, `workflow_run_id`, `failure_id`, `failure_category`, `closed_reason`, `exception_class`, `message`, `reason`, `cancellation_cleanup` | `HistoryTimeline`, `FailureSnapshots`, `ChildRunHistory`, cancellation projections |

diff --git a/src/V2/Support/DefaultWorkflowControlPlane.php b/src/V2/Support/DefaultWorkflowControlPlane.php
@@ -240,8 +240,8 @@ public function redrive(string $instanceId, string $failedRunId, array $options
                 ],
             ), null, $command);
 
-            app(MemoUpsertService::class)->inheritFromParent($source, $continuedRun, 1);
-            app(SearchAttributeUpsertService::class)->inheritFromParent($source, $continuedRun, 1);
+            app(MemoUpsertService::class)->copyForRedrive($source, $continuedRun);
+            app(SearchAttributeUpsertService::class)->copyForRedrive($source, $continuedRun);
 
             foreach ($plan['reused_steps'] as $step) {
                 $reused = $step['payload'];

diff --git a/src/V2/Support/FailedRunRedriveEligibility.php b/src/V2/Support/FailedRunRedriveEligibility.php
@@ -59,16 +59,22 @@ public static function evaluate(
             return self::refuse((string) $plan['reason']);
         }
 
-        // Start-time memos and search attributes carry to the successor.
-        // Values upserted by workflow code belong to the replayed prefix,
-        // which redrive does not copy yet.
+        // Memo and search attribute rows carry to the successor as they are.
+        // A value written by workflow code must come from an upsert step the
+        // successor reuses, or its replay would not explain the row.
         if (
-            $source->memos()
-                ->where('upserted_at_sequence', '>', 0)
-                ->exists()
-            || $source->searchAttributes()
-                ->where('upserted_at_sequence', '>', 0)
-                ->exists()
+            ! self::rowsMatchReusedUpserts(
+                $source->memos()
+                    ->get(),
+                $plan['reused_steps'],
+                HistoryEventType::MemoUpserted,
+            )
+            || ! self::rowsMatchReusedUpserts(
+                $source->searchAttributes()
+                    ->get(),
+                $plan['reused_steps'],
+                HistoryEventType::SearchAttributesUpserted,
+            )
         ) {
             return self::refuse('unsupported_visibility_metadata');
         }
@@ -134,6 +140,32 @@ public static function blockedReason(WorkflowRun $run, ?WorkflowRun $currentRun)
         return self::evaluate($run, $currentRun, fingerprintPresentedAtRedrive: true)['reason'];
     }
 
+    /**
+     * @param iterable<\Workflow\V2\Models\WorkflowMemo|\Workflow\V2\Models\WorkflowSearchAttribute> $rows
+     * @param list<array{event_type: HistoryEventType, payload: array<string, mixed>, recorded_at: string|null}> $reusedSteps
+     */
+    private static function rowsMatchReusedUpserts(iterable $rows, array $reusedSteps, HistoryEventType $type): bool
+    {
+        $upsertSequences = [];
+        foreach ($reusedSteps as $step) {
+            $sequence = $step['payload']['sequence'] ?? null;
+            if ($step['event_type'] === $type && is_int($sequence)) {
+                $upsertSequences[$sequence] = true;
+            }
+        }
+
+        foreach ($rows as $row) {
+            $sequence = (int) $row->upserted_at_sequence;
+            // Start-time values (step 0) and values inherited from an earlier
+            // run in the chain are not written by this run's workflow code.
+            if ($sequence > 0 && ! $row->inherited_from_parent && ! isset($upsertSequences[$sequence])) {
+                return false;
+            }
+        }
+
+        return true;
+    }
+
     private static function definitionMatches(
         WorkflowRun $source,
         mixed $sourceFingerprint,

diff --git a/src/V2/Support/FailedRunRedrivePlan.php b/src/V2/Support/FailedRunRedrivePlan.php
@@ -87,6 +87,8 @@ public static function forRun(WorkflowRun $run): array
             if (in_array($event->event_type, [
                 HistoryEventType::SideEffectRecorded,
                 HistoryEventType::VersionMarkerRecorded,
+                HistoryEventType::MemoUpserted,
+                HistoryEventType::SearchAttributesUpserted,
             ], true)) {
                 $payload = $event->payload;
                 $sequence = is_array($payload) ? ($payload['sequence'] ?? null) : null;

diff --git a/src/V2/Support/HistoryEventPayloadContract.php b/src/V2/Support/HistoryEventPayloadContract.php
@@ -908,8 +908,15 @@ final class HistoryEventPayloadContract
             'code',
             'exception',
         ],
-        'SearchAttributesUpserted' => ['sequence', 'attributes', 'attribute_types', 'merged'],
-        'MemoUpserted' => ['sequence', 'entries', 'merged'],
+        'SearchAttributesUpserted' => [
+            'sequence',
+            'attributes',
+            'attribute_types',
+            'merged',
+            'reused_from_run_id',
+            'reused_recorded_at',
+        ],
+        'MemoUpserted' => ['sequence', 'entries', 'merged', 'reused_from_run_id', 'reused_recorded_at'],
         'RepairRequested' => [
             'workflow_command_id',
             'workflow_instance_id',

diff --git a/src/V2/Support/MemoUpsertService.php b/src/V2/Support/MemoUpsertService.php
@@ -102,6 +102,37 @@ public function inheritFromParent(
         });
     }
 
+    /**
+     * Copy memos to a redrive successor as they are. The successor reuses the
+     * source's history, so each row keeps the step that wrote it and whether
+     * it was inherited.
+     */
+    public function copyForRedrive(WorkflowRun $source, WorkflowRun $successor): void
+    {
+        $sourceMemos = WorkflowMemo::where('workflow_run_id', $source->id)->get();
+
+        ConfiguredV2Models::connection()->transaction(static function () use ($sourceMemos, $successor): void {
+            foreach ($sourceMemos as $sourceMemo) {
+                $memo = new WorkflowMemo([
+                    'workflow_run_id' => $successor->id,
+                    'workflow_instance_id' => $successor->workflow_instance_id,
+                    'key' => $sourceMemo->key,
+                    'upserted_at_sequence' => $sourceMemo->upserted_at_sequence,
+                    'inherited_from_parent' => $sourceMemo->inherited_from_parent,
+                ]);
+
+                $memo->setValue($sourceMemo->getValue());
+
+                $memo->save();
+            }
+
+            if ($sourceMemos->isNotEmpty()) {
+                WorkflowMemo::validateTotalSize($successor->id);
+                WorkflowMemo::validateCount($successor->id);
+            }
+        });
+    }
+
     /**
      * Get memos as a key-value array.
      *

diff --git a/src/V2/Support/QueryStateReplayer.php b/src/V2/Support/QueryStateReplayer.php
@@ -401,7 +401,7 @@ public function replayState(WorkflowRun $run): ReplayState
                 );
 
                 $this->syncWorkflowCursor($workflow, $sequence + 1);
-                $current = $workflowExecution->send(null, $upsertEvent->recorded_at);
+                $current = $workflowExecution->send(null, RedriveReplayClock::recordedAt($upsertEvent));
 
                 ++$sequence;
 
@@ -428,7 +428,7 @@ public function replayState(WorkflowRun $run): ReplayState
                 );
 
                 $this->syncWorkflowCursor($workflow, $sequence + 1);
-                $current = $workflowExecution->send(null, $upsertEvent->recorded_at);
+                $current = $workflowExecution->send(null, RedriveReplayClock::recordedAt($upsertEvent));
 
                 ++$sequence;
 

diff --git a/src/V2/Support/SearchAttributeUpsertService.php b/src/V2/Support/SearchAttributeUpsertService.php
@@ -236,6 +236,40 @@ public function inheritFromParent(
         });
     }
 
+    /**
+     * Copy search attributes to a redrive successor as they are. The successor
+     * reuses the source's history, so each row keeps the step that wrote it
+     * and whether it was inherited.
+     */
+    public function copyForRedrive(WorkflowRun $source, WorkflowRun $successor): void
+    {
+        ConfiguredV2Models::connection()->transaction(static function () use ($source, $successor): void {
+            $sourceAttributes = WorkflowSearchAttribute::where('workflow_run_id', $source->id)
+                ->get();
+
+            foreach ($sourceAttributes as $sourceAttribute) {
+                (new WorkflowSearchAttribute([
+                    'workflow_run_id' => $successor->id,
+                    'workflow_instance_id' => $successor->workflow_instance_id,
+                    'key' => $sourceAttribute->key,
+                    'type' => $sourceAttribute->type,
+                    'value_string' => $sourceAttribute->value_string,
+                    'value_keyword' => $sourceAttribute->value_keyword,
+                    'value_keyword_list' => $sourceAttribute->value_keyword_list,
+                    'value_int' => $sourceAttribute->value_int,
+                    'value_float' => $sourceAttribute->value_float,
+                    'value_bool' => $sourceAttribute->value_bool,
+                    'value_datetime' => $sourceAttribute->value_datetime,
+                    'upserted_at_sequence' => $sourceAttribute->upserted_at_sequence,
+                    'inherited_from_parent' => $sourceAttribute->inherited_from_parent,
+                ]))->save();
+            }
+
+            WorkflowSearchAttribute::validateCount($successor->id);
+            WorkflowSearchAttribute::validateTotalSize($successor->id);
+        });
+    }
+
     /**
      * Get all search attributes for a run as key-value array.
      *

diff --git a/src/V2/Support/WorkflowExecutor.php b/src/V2/Support/WorkflowExecutor.php
@@ -1162,7 +1162,11 @@ private function executeSearchAttributes(
                 SearchAttributeReplayIdentity::assertCompatible($sequence, $upsertEvent->payload, $current);
             }
             $this->syncWorkflowCursor($workflow, $sequence + 1);
-            $current = $workflowExecution->send(null, $upsertEvent?->recorded_at, advanceCancellationTime: false);
+            $current = $workflowExecution->send(
+                null,
+                RedriveReplayClock::recordedAt($upsertEvent),
+                advanceCancellationTime: false,
+            );
         } catch (Throwable $throwable) {
             $this->failRun($run, $task, $throwable, 'workflow_run', $run->id);
             return null;
@@ -1201,7 +1205,11 @@ private function executeMemo(
                 );
             }
             $this->syncWorkflowCursor($workflow, $sequence + 1);
-            $current = $workflowExecution->send(null, $memoEvent?->recorded_at, advanceCancellationTime: false);
+            $current = $workflowExecution->send(
+                null,
+                RedriveReplayClock::recordedAt($memoEvent),
+                advanceCancellationTime: false,
+            );
         } catch (Throwable $throwable) {
             $this->failRun($run, $task, $throwable, 'workflow_run', $run->id);
             return null;

diff --git a/src/V2/Support/WorkflowFiberRunner.php b/src/V2/Support/WorkflowFiberRunner.php
@@ -2428,7 +2428,8 @@ private static function indexRecordedSearchAttributeUpserts(array $historyEvents
 
             $upserts[$sequence] = [
                 'payload' => $payload,
-                'recorded_at' => self::eventRecordedAt($event, $payload),
+                'recorded_at' => self::eventTime($payload['reused_recorded_at'] ?? null)
+                    ?? self::eventRecordedAt($event, $payload),
             ];
         }
 
@@ -2457,7 +2458,8 @@ private static function indexRecordedMemoUpserts(array $historyEvents): array
 
             $upserts[$sequence] = [
                 'entries' => $payload['entries'] ?? null,
-                'recorded_at' => self::eventRecordedAt($event, $payload),
+                'recorded_at' => self::eventTime($payload['reused_recorded_at'] ?? null)
+                    ?? self::eventRecordedAt($event, $payload),
             ];
         }
 

diff --git a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php
@@ -316,6 +316,68 @@ public function testRedriveReusedMarkersCorpusKeepsTheirStepsAndTime(): void
         ]);
     }
 
+    public function testRedriveReusedUpsertsCorpusKeepsTheirStepsRowsAndTime(): void
+    {
+        $path = self::FIXTURE_DIR . '/redrive-reused-upserts-clock.json';
+        if (! is_file($path)) {
+            $this->markTestSkipped('The redrive reused-upsert evidence is absent.');
+        }
+        $fixture = json_decode(file_get_contents($path), true, flags: JSON_THROW_ON_ERROR);
+        $scenario = $fixture['redrive_reused_upserts'];
+        self::stopWorkers();
+        config([
+            'queue.default' => 'sync',
+            'queue.connections.sync.driver' => 'sync',
+        ]);
+        foreach (['first-calls', 'second-calls'] as $counter) {
+            Cache::forget("test:redrive:{$counter}");
+        }
+        $this->clearWorkflowState();
+        WorkflowStub::fake();
+
+        $stub = WorkflowStub::make($fixture['workflow']['type'], 'regression-corpus-redrive-upserts');
+        $stub->start(...$fixture['workflow']['arguments']);
+        $this->assertTrue($stub->refresh()->failed());
+        $sourceRunId = $stub->runId();
+        $sourceTimes = [];
+        foreach ($scenario['reused_steps'] as $step) {
+            $sourceTimes[$step['sequence']] = WorkflowHistoryEvent::query()
+                ->where('workflow_run_id', $sourceRunId)
+                ->where('event_type', $step['event_type'])
+                ->firstOrFail()
+                ->recorded_at?->toIso8601String();
+        }
+
+        $this->travel($scenario['redrive_after_minutes'])
+            ->minutes();
+        $result = app(WorkflowControlPlane::class)->redrive('regression-corpus-redrive-upserts', $sourceRunId);
+
+        $this->assertSame($scenario['redrive'] === 'accepted', $result['accepted'], (string) $result['reason']);
+        $this->assertSame($scenario['resume_step_sequence'], $result['resume_step_sequence']);
+        $successor = WorkflowRun::query()->findOrFail($result['workflow_run_id']);
+        $this->assertSame($scenario['successor_status'], $successor->status->value);
+        $this->assertSame(1, (int) Cache::get('test:redrive:first-calls'));
+        foreach ($scenario['reused_steps'] as $step) {
+            $reused = WorkflowHistoryEvent::query()
+                ->where('workflow_run_id', $successor->id)
+                ->where('event_type', $step['event_type'])
+                ->sole();
+            $this->assertSame($step['sequence'], $reused->payload['sequence']);
+            $this->assertSame($sourceRunId, $reused->payload['reused_from_run_id'] ?? null);
+            $this->assertSame($sourceTimes[$step['sequence']], $reused->payload['reused_recorded_at'] ?? null);
+        }
+        $this->assertEquals($scenario['memo'], $successor->typedMemos());
+        $this->assertEquals($scenario['search_attributes'], $successor->typedSearchAttributes());
+        $output = $successor->workflowOutput();
+        $this->assertSame([
+            'time_after_memo' => $sourceTimes[1],
+            'time_after_search_attributes' => $sourceTimes[2],
+        ], [
+            'time_after_memo' => $output['time_after_memo'],
+            'time_after_search_attributes' => $output['time_after_search_attributes'],
+        ]);
+    }
+
     public function testRedriveReusedTimerCorpusDoesNotWaitAgain(): void
     {
         $path = self::FIXTURE_DIR . '/redrive-reused-timer-clock.json';