From 462dd92d0e9f392ccbbade5cf0b0e746832cc732 Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Tue, 3 Nov 2026 21:00:22 -0600
Subject: [PATCH 1/6] Keep a stream's pending messages receivable whatever
 moves the run cursor

A run's message cursor is shared by signals, updates and every stream.
Stream reads filtered on sequence > cursor, so a signal, an update or a
receive on another stream hid older pending messages for good (#1011).
peek, receive, pendingCount and hasPending now read every pending
message of the stream in sequence order; consume_state already records
what each stream consumed. The cursor and its history events are
unchanged.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 CHANGELOG.md                                  | 11 ++++
 src/V2/Support/MessageService.php             | 15 ++++--
 .../Feature/V2/V2MessageStreamReplayTest.php  | 53 +++++++++++++++++++
 3 files changed, 76 insertions(+), 3 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 756fd8108..640cc992c 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,17 @@
 
 ## Unreleased
 
+- A message on one stream stays receivable until it is consumed or expires
+  (#1011). The run's message cursor is shared by signals, updates and every
+  stream, so a signal, an update or a receive on another stream used to move
+  it past older pending messages, and `inbox()->receive()`, `peek()`,
+  `pendingCount()` and `hasPending()` no longer returned them. They now read
+  every pending message of the stream, in sequence order.
+  - After upgrading, workflow code can receive messages that earlier
+    releases hid. They are the stream's oldest pending messages, so a
+    workflow that receives in a loop gets them first.
+  - A receive replayed from history still returns what that step consumed.
+
 ## 2.20.1 - 2026-11-04
 
 - Rollback, embedded Workflow package: 2.20.1 changes no schema, history
diff --git a/src/V2/Support/MessageService.php b/src/V2/Support/MessageService.php
index f0619ac77..1eb9bafc4 100644
--- a/src/V2/Support/MessageService.php
+++ b/src/V2/Support/MessageService.php
@@ -153,7 +153,10 @@ public function peekMessages(
         int $limit = 100,
     ): \Illuminate\Database\Eloquent\Collection {
         $streamKey = $streamKey ?? "instance:{$run->workflow_instance_id}";
-        $afterSequence = MessageStreamCursor::positionForRun($run);
+        // A stream message stays receivable until it is consumed or expires.
+        // The run's message cursor is shared with signals, updates and other
+        // streams, so it doesn't decide what this stream has left (#1011).
+        $afterSequence = 0;
 
         return WorkflowMessage::getUnconsumedForStream($run, $streamKey, $afterSequence, $limit);
     }
@@ -306,7 +309,10 @@ public function consumeMessageBatch(WorkflowRun $run, array $messages, int $cons
     public function getUnconsumedCount(WorkflowRun $run, ?string $streamKey = null): int
     {
         $streamKey = $streamKey ?? "instance:{$run->workflow_instance_id}";
-        $afterSequence = MessageStreamCursor::positionForRun($run);
+        // A stream message stays receivable until it is consumed or expires.
+        // The run's message cursor is shared with signals, updates and other
+        // streams, so it doesn't decide what this stream has left (#1011).
+        $afterSequence = 0;
 
         return WorkflowMessage::getUnconsumedCountForStream($run, $streamKey, $afterSequence);
     }
@@ -322,7 +328,10 @@ public function getUnconsumedCount(WorkflowRun $run, ?string $streamKey = null):
     public function hasUnconsumedMessages(WorkflowRun $run, ?string $streamKey = null): bool
     {
         $streamKey = $streamKey ?? "instance:{$run->workflow_instance_id}";
-        $afterSequence = MessageStreamCursor::positionForRun($run);
+        // A stream message stays receivable until it is consumed or expires.
+        // The run's message cursor is shared with signals, updates and other
+        // streams, so it doesn't decide what this stream has left (#1011).
+        $afterSequence = 0;
 
         return WorkflowMessage::hasUnconsumedMessages($run, $streamKey, $afterSequence);
     }
diff --git a/tests/Feature/V2/V2MessageStreamReplayTest.php b/tests/Feature/V2/V2MessageStreamReplayTest.php
index b523c854a..0d8630a36 100644
--- a/tests/Feature/V2/V2MessageStreamReplayTest.php
+++ b/tests/Feature/V2/V2MessageStreamReplayTest.php
@@ -277,6 +277,59 @@ public function testSendsThatDifferOnlyByCaseAccentOrTrailingSpaceStaySeparate()
         );
     }
 
+    /**
+     * A signal advances the run's message cursor past an older message that is
+     * still pending on a stream. The message stays receivable (#1011).
+     */
+    public function testASignalDoesNotHideAnOlderPendingStreamMessage(): void
+    {
+        $workflow = WorkflowStub::make(TestInboxTurnsWorkflow::class, 'inbox-before-signal');
+        $workflow->start(1);
+        $this->drainReadyTasks();
+        $run = WorkflowRun::query()->where('workflow_instance_id', 'inbox-before-signal')->sole();
+
+        // The message arrives before the ping that wakes the receive.
+        $this->sendToInbox($run, 'm1');
+        $workflow->signal('ping');
+        $this->drainReadyTasks();
+
+        $this->assertTrue($workflow->refresh()->completed());
+        $this->assertSame(['m1'], $workflow->output());
+    }
+
+    /**
+     * Consuming a newer message on one stream doesn't hide an older pending
+     * message on another stream of the same run.
+     */
+    public function testAReceiveOnOneStreamDoesNotHideAnotherStreamsOlderMessage(): void
+    {
+        $workflow = WorkflowStub::make(TestInboxTurnsWorkflow::class, 'inbox-two-streams');
+        $workflow->start(1);
+        $this->drainReadyTasks();
+        $run = WorkflowRun::query()->where('workflow_instance_id', 'inbox-two-streams')->sole();
+        $service = new MessageService();
+        $service->sendMessage($run, MessageChannel::WorkflowMessage, $run->workflow_instance_id, 'b1', 'b');
+        $service->sendMessage($run, MessageChannel::WorkflowMessage, $run->workflow_instance_id, 'a1', 'a');
+
+        $a1 = $service->peekMessages($run->fresh(), 'a')
+            ->sole();
+        $service->consumeMessage($run->fresh(), $a1, 1);
+        $this->assertGreaterThan(
+            (int) WorkflowMessage::query()->where('payload_reference', 'b1')->where(
+                'direction',
+                MessageDirection::Inbound->value
+            )->sole()
+->sequence,
+            (int) $run->fresh()
+->message_cursor_position,
+        );
+
+        $this->assertSame(['b1'], $service->peekMessages($run->fresh(), 'b')->pluck('payload_reference')->all());
+        $this->assertSame(1, $service->getUnconsumedCount($run->fresh(), 'b'));
+        $this->assertTrue($service->hasUnconsumedMessages($run->fresh(), 'b'));
+        $this->assertSame([], $service->peekMessages($run->fresh(), 'a')->pluck('payload_reference')->all());
+    }
+
     public function testAWorkflowCodeReceiveReplaysTheMessageItConsumedAndQueriesConsumeNothing(): void
     {
         $workflow = WorkflowStub::make(TestInboxTurnsWorkflow::class, 'inbox-turns');

From be836c266fda3532f37e07ef7bbd463111779f26 Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Tue, 3 Nov 2026 21:37:46 -0600
Subject: [PATCH 2/6] Restore the test boundary lost in the main merge

The conflict resolution in 999865a2 dropped the closing brace of the
#1011 cross-stream test and the opening of the next docblock, so the
file didn't parse. Both sides' tests are kept; the file now runs 10/10.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 tests/Feature/V2/V2MessageStreamReplayTest.php | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/tests/Feature/V2/V2MessageStreamReplayTest.php b/tests/Feature/V2/V2MessageStreamReplayTest.php
index 49224302f..6a3a3e78e 100644
--- a/tests/Feature/V2/V2MessageStreamReplayTest.php
+++ b/tests/Feature/V2/V2MessageStreamReplayTest.php
@@ -334,6 +334,9 @@ public function testAReceiveOnOneStreamDoesNotHideAnotherStreamsOlderMessage():
         $this->assertSame(1, $service->getUnconsumedCount($run->fresh(), 'b'));
         $this->assertTrue($service->hasUnconsumedMessages($run->fresh(), 'b'));
         $this->assertSame([], $service->peekMessages($run->fresh(), 'a')->pluck('payload_reference')->all());
+    }
+
+    /**
      * A message that arrives while the run is failed belongs to that run. The
      * redrive moves it to the successor, which receives it at a later step.
      */

From 050095ff0315208972412266ad6b5977b6b6deaf Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Tue, 3 Nov 2026 22:13:50 -0600
Subject: [PATCH 3/6] Never consume at a step history has already passed

General's hold on be836c26 (H1): with the cursor filter gone, a run
whose earlier receive found nothing because the cursor hid a message
would, on replay after upgrading, peek and consume that message at the
old step. A branching workflow then failed with a history mismatch, and
a straight-line one silently changed its result.

A workflow-code receive with nothing recorded at its step now consumes
only when history hasn't passed the step the code is at (read live, so
a stream kept in a property still consumes at later steps). Otherwise
it returns empty. This also closes the 2.20.1 note that an empty
replayed receive reads the stream again.

M2: api-stability.md and workflow-messages-architecture.md no longer
describe stream reads as filtered by the run cursor.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 CHANGELOG.md                                  | 10 ++-
 docs/api-stability.md                         |  6 +-
 docs/architecture/execution-guarantees.md     |  3 +-
 docs/workflow-messages-architecture.md        |  6 +-
 src/V2/MessageStream.php                      | 24 +++++-
 src/V2/Support/MessageReplayLedger.php        | 32 ++++++++
 src/V2/Workflow.php                           |  1 +
 .../Feature/V2/V2MessageStreamReplayTest.php  | 82 +++++++++++++++++++
 tests/Fixtures/V2/TestInboxBranchWorkflow.php | 42 ++++++++++
 .../V2/TestInboxKeptStreamWorkflow.php        | 41 ++++++++++
 10 files changed, 237 insertions(+), 10 deletions(-)
 create mode 100644 tests/Fixtures/V2/TestInboxBranchWorkflow.php
 create mode 100644 tests/Fixtures/V2/TestInboxKeptStreamWorkflow.php

diff --git a/CHANGELOG.md b/CHANGELOG.md
index d1e106718..f54a8f77b 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -11,12 +11,16 @@
   - After upgrading, workflow code can receive messages that earlier
     releases hid. They are the stream's oldest pending messages, so a
     workflow that receives in a loop gets them first.
-  - A receive replayed from history still returns what that step consumed.
+  - A receive replayed from history still returns what that step consumed,
+    and a replayed receive whose original call found nothing returns nothing
+    again. It no longer reads the stream at a step history has passed. So a
+    run where an earlier release's cursor hid a message keeps the result it
+    had, and receives the message at a later step instead of consuming it at
+    the old one.
 - A redrive moves the failed run's pending inbound messages to the successor
   (#1013). Before, only the message cursor carried over, so a message that
   arrived while the run was failed stayed on the failed run and the successor
-  never received it. A replayed receive whose original call found nothing can
-  now return one of the moved messages.
+  never received it.
 
 ## 2.20.1 - 2026-11-04
 
diff --git a/docs/api-stability.md b/docs/api-stability.md
index 3c820f7c5..c05365cbb 100644
--- a/docs/api-stability.md
+++ b/docs/api-stability.md
@@ -617,8 +617,10 @@ flows:
 
 - `MessageService::sendMessage()` creates paired outbound and inbound
   `workflow_messages` rows with one reserved instance sequence.
-- `MessageService::peekMessages()` and `receiveMessages()` read pending
-  inbound messages after the run cursor. They are intentionally read-only:
+- `MessageService::peekMessages()` and `receiveMessages()` read every pending
+  inbound message of the stream, in sequence order, whatever the run cursor
+  says. The cursor is shared with signals, updates and other streams, so it
+  doesn't decide what a stream has left. They are intentionally read-only:
   they do not mark messages consumed and do not advance the cursor.
 - `MessageService::consumeMessage()` and `consumeMessages()` are the only
   message-service APIs that mark messages consumed and advance the cursor.
diff --git a/docs/architecture/execution-guarantees.md b/docs/architecture/execution-guarantees.md
index e6b5228ad..a034d560e 100644
--- a/docs/architecture/execution-guarantees.md
+++ b/docs/architecture/execution-guarantees.md
@@ -259,7 +259,8 @@ own identifier and its own durable row.
   already wrote for the same content, and a receive returns the
   messages already consumed at that step. A query or other read-only
   replay never writes or consumes. A replayed receive whose original
-  call found nothing reads the stream again.
+  call found nothing returns nothing again; a receive never consumes at a
+  step that history has already passed.
 
 ## Side effects and version markers
 
diff --git a/docs/workflow-messages-architecture.md b/docs/workflow-messages-architecture.md
index 32b109a3c..2430004ea 100644
--- a/docs/workflow-messages-architecture.md
+++ b/docs/workflow-messages-architecture.md
@@ -216,10 +216,12 @@ Returns messages where:
 - `workflow_run_id = $run->id`
 - `stream_key = $streamKey`
 - `consume_state = 'pending'`
-- `sequence > $run->message_cursor_position`
+
+ordered by `sequence`. The run cursor is shared with signals, updates and other
+streams, so it doesn't filter stream reads.
 
 `peekMessages()` and `receiveMessages()` are read-only. They return only
-unconsumed messages after the cursor, but they do not mark messages consumed and
+unconsumed messages, but they do not mark messages consumed and
 they do not advance the cursor. Cursor movement is explicit and happens only
 through `consumeMessage()` or `consumeMessages()` after workflow code has
 durably recorded how it handled the messages.
diff --git a/src/V2/MessageStream.php b/src/V2/MessageStream.php
index 25167bca6..35a2df758 100644
--- a/src/V2/MessageStream.php
+++ b/src/V2/MessageStream.php
@@ -31,6 +31,7 @@ public function __construct(
         private readonly ?int $defaultConsumedBySequence = null,
         private readonly ?MessageReplayLedger $replayLedger = null,
         private readonly bool|Closure $readOnly = false,
+        private readonly ?Closure $currentStep = null,
     ) {
         if ($streamKey === '') {
             throw new InvalidArgumentException('Message stream key must not be empty.');
@@ -66,6 +67,7 @@ public static function forWorkflowCode(
         int $defaultConsumedBySequence,
         MessageReplayLedger $replayLedger,
         Closure $readOnly,
+        ?Closure $currentStep = null,
     ): self {
         return new self(
             $run,
@@ -74,6 +76,7 @@ public static function forWorkflowCode(
             $defaultConsumedBySequence,
             $replayLedger,
             $readOnly,
+            $currentStep,
         );
     }
 
@@ -120,8 +123,11 @@ public function receive(int $limit = 1, ?int $consumedBySequence = null): Collec
             $sequence = $this->consumedBySequence($consumedBySequence);
             $recorded = $this->recordedReceive($sequence, $limit);
 
-            // A receive this step already made returns what it consumed.
-            if ($recorded->isNotEmpty() || $this->readOnly()) {
+            // A receive this step already made returns what it consumed. A
+            // receive at a step history has already passed, with nothing
+            // recorded, found nothing the first time, so it consumes nothing
+            // now either.
+            if ($recorded->isNotEmpty() || $this->readOnly() || $this->historyPassedCurrentStep()) {
                 return $recorded;
             }
 
@@ -341,6 +347,20 @@ private function recordedReceive(int $consumedBySequence, int $limit): Collectio
         return $messages;
     }
 
+    /**
+     * Whether history already records a step at or after the step workflow
+     * code is at now, so this code ran in an earlier task. The step is read on
+     * every call, so a stream kept in a property is not mistaken for replay.
+     */
+    private function historyPassedCurrentStep(): bool
+    {
+        if (! $this->currentStep instanceof Closure || ! $this->replayLedger instanceof MessageReplayLedger) {
+            return false;
+        }
+
+        return (int) ($this->currentStep)() <= $this->replayLedger->highestRecordedStep($this->run);
+    }
+
     /**
      * Checked on every call: a stream kept across a replayed update handler,
      * where dispatch is off, must still write once dispatch is back on.
diff --git a/src/V2/Support/MessageReplayLedger.php b/src/V2/Support/MessageReplayLedger.php
index e16292cf1..6cd28636d 100644
--- a/src/V2/Support/MessageReplayLedger.php
+++ b/src/V2/Support/MessageReplayLedger.php
@@ -42,6 +42,8 @@ final class MessageReplayLedger
      */
     private ?array $runIds = null;
 
+    private ?int $highestRecordedStep = null;
+
     /**
      * This run and the runs it was redriven from, newest first. A send or
      * receive in a reused prefix was made by one of them.
@@ -86,6 +88,36 @@ public function runIds(WorkflowRun $run): array
         return $this->runIds = $runIds;
     }
 
+    /**
+     * The highest step sequence this run's history recorded when this
+     * execution first asked. Steps this execution records later are at or
+     * past the step it runs, so a stale answer only ever errs toward "not
+     * passed yet" for them.
+     */
+    public function highestRecordedStep(WorkflowRun $run): int
+    {
+        if ($this->highestRecordedStep !== null) {
+            return $this->highestRecordedStep;
+        }
+
+        $highest = 0;
+
+        foreach (WorkflowHistoryEvent::query()
+            ->where('workflow_run_id', $run->id)
+            ->pluck('payload') as $payload) {
+            if (is_string($payload)) {
+                $payload = json_decode($payload, true);
+            }
+            $sequence = is_array($payload) ? ($payload['sequence'] ?? null) : null;
+
+            if (is_int($sequence) && $sequence > $highest) {
+                $highest = $sequence;
+            }
+        }
+
+        return $this->highestRecordedStep = $highest;
+    }
+
     public function sendOffset(string $identity): int
     {
         return $this->sendOffsets[$identity] ?? 0;
diff --git a/src/V2/Workflow.php b/src/V2/Workflow.php
index 14d1572bf..10b1891df 100644
--- a/src/V2/Workflow.php
+++ b/src/V2/Workflow.php
@@ -193,6 +193,7 @@ public function messages(?string $streamKey = null, ?MessageService $messages =
             $this->visibleSequence,
             $this->messageReplayLedger,
             readOnly: fn (): bool => ! $this->commandDispatchEnabled,
+            currentStep: fn (): int => $this->visibleSequence,
         );
     }
 
diff --git a/tests/Feature/V2/V2MessageStreamReplayTest.php b/tests/Feature/V2/V2MessageStreamReplayTest.php
index 6a3a3e78e..8431354f0 100644
--- a/tests/Feature/V2/V2MessageStreamReplayTest.php
+++ b/tests/Feature/V2/V2MessageStreamReplayTest.php
@@ -6,6 +6,8 @@
 
 use Illuminate\Support\Facades\Cache;
 use Illuminate\Support\Facades\Queue;
+use Tests\Fixtures\V2\TestInboxBranchWorkflow;
+use Tests\Fixtures\V2\TestInboxKeptStreamWorkflow;
 use Tests\Fixtures\V2\TestInboxTurnsWorkflow;
 use Tests\Fixtures\V2\TestMessageStreamExactContentWorkflow;
 use Tests\Fixtures\V2\TestMessageStreamKeptStreamWorkflow;
@@ -384,6 +386,86 @@ public function testARedriveSuccessorReceivesAMessageThatArrivedWhileTheSourceWa
         $this->assertSame(MessageConsumeState::Consumed, $m2->consume_state);
     }
 
+    /**
+     * A receive that found nothing replays as nothing, even when a message
+     * has arrived since. The message waits for a receive the run hasn't made
+     * yet (#1011 review): a history where an earlier release's cursor hid a
+     * message must not consume it at the old step after upgrading.
+     */
+    public function testAReceiveThatFoundNothingReplaysAsNothingAndTheMessageWaitsForTheNextTurn(): void
+    {
+        $workflow = WorkflowStub::make(TestInboxTurnsWorkflow::class, 'inbox-empty-first');
+        $workflow->start(2);
+        $this->drainReadyTasks();
+        $run = WorkflowRun::query()->where('workflow_instance_id', 'inbox-empty-first')->sole();
+
+        // Turn 1 finds the inbox empty.
+        $workflow->signal('ping');
+        $this->drainReadyTasks();
+        $this->sendToInbox($run, 'm1');
+        $before = $this->messageRows();
+        $this->assertSame([null], $workflow->refresh()->query('received'));
+        $this->assertSame($before, $this->messageRows());
+
+        $workflow->signal('ping');
+        $this->drainReadyTasks();
+
+        $this->assertTrue($workflow->refresh()->completed());
+        $this->assertSame([null, 'm1'], $workflow->output());
+    }
+
+    /**
+     * A stream kept from an earlier step still consumes at the step the code
+     * is at now: the replay check reads the current step on every call.
+     */
+    public function testAKeptInboxStillReceivesAtLaterTurns(): void
+    {
+        $workflow = WorkflowStub::make(TestInboxKeptStreamWorkflow::class, 'inbox-kept');
+        $workflow->start(2);
+        $this->drainReadyTasks();
+        $run = WorkflowRun::query()->where('workflow_instance_id', 'inbox-kept')->sole();
+
+        $workflow->signal('ping');
+        $this->sendToInbox($run, 'm1');
+        $this->drainReadyTasks();
+        $workflow->signal('ping');
+        $this->sendToInbox($run, 'm2');
+        $this->drainReadyTasks();
+
+        $this->assertTrue($workflow->refresh()->completed());
+        $this->assertSame(['m1', 'm2'], $workflow->output());
+    }
+
+    /**
+     * The same history when workflow code branches on the receive: replay
+     * takes the branch the original run took, so the run doesn't fail with a
+     * history mismatch.
+     */
+    public function testABranchOnAnEmptyReceiveReplaysTheBranchTheRunTook(): void
+    {
+        $workflow = WorkflowStub::make(TestInboxBranchWorkflow::class, 'inbox-branch');
+        $workflow->start(2);
+        $this->drainReadyTasks();
+        $run = WorkflowRun::query()->where('workflow_instance_id', 'inbox-branch')->sole();
+
+        $workflow->signal('ping');
+        $this->drainReadyTasks();
+        $this->sendToInbox($run, 'm1');
+        $workflow->signal('ping');
+        $this->drainReadyTasks();
+
+        $this->assertTrue($workflow->refresh()->completed(), (string) $workflow->refresh()->status());
+        $this->assertSame(['none:Hello, none!', 'got:m1'], $workflow->output());
+        $this->assertSame(
+            MessageConsumeState::Consumed,
+            WorkflowMessage::query()->where('payload_reference', 'm1')->where(
+                'direction',
+                MessageDirection::Inbound->value
+            )->sole()
+->consume_state,
+        );
+    }
+
     public function testAWorkflowCodeReceiveReplaysTheMessageItConsumedAndQueriesConsumeNothing(): void
     {
         $workflow = WorkflowStub::make(TestInboxTurnsWorkflow::class, 'inbox-turns');
diff --git a/tests/Fixtures/V2/TestInboxBranchWorkflow.php b/tests/Fixtures/V2/TestInboxBranchWorkflow.php
new file mode 100644
index 000000000..794acc896
--- /dev/null
+++ b/tests/Fixtures/V2/TestInboxBranchWorkflow.php
@@ -0,0 +1,42 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Tests\Fixtures\V2;
+
+use function Workflow\V2\activity;
+use Workflow\V2\Attributes\Signal;
+use Workflow\V2\Attributes\Type;
+use function Workflow\V2\await;
+use function Workflow\V2\sideEffect;
+use Workflow\V2\Workflow;
+
+/**
+ * Each turn waits for a ping and takes a different step depending on whether
+ * the inbox had a message, so a replay that receives differently from the
+ * original run takes a different step.
+ */
+#[Type('test-inbox-branch-workflow')]
+#[Signal('ping')]
+final class TestInboxBranchWorkflow extends Workflow
+{
+    /**
+     * @return list<string>
+     */
+    public function handle(int $maxTurns = 2): array
+    {
+        $outcomes = [];
+
+        while (count($outcomes) < $maxTurns) {
+            await('ping');
+            $message = $this->inbox('chat')
+                ->receiveOne();
+
+            $outcomes[] = $message === null
+                ? 'none:' . activity(TestRedriveFirstActivity::class, 'none')
+                : 'got:' . sideEffect(static fn (): string => (string) $message->payload_reference);
+        }
+
+        return $outcomes;
+    }
+}
diff --git a/tests/Fixtures/V2/TestInboxKeptStreamWorkflow.php b/tests/Fixtures/V2/TestInboxKeptStreamWorkflow.php
new file mode 100644
index 000000000..2d938a591
--- /dev/null
+++ b/tests/Fixtures/V2/TestInboxKeptStreamWorkflow.php
@@ -0,0 +1,41 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Tests\Fixtures\V2;
+
+use function Workflow\V2\activity;
+use Workflow\V2\Attributes\Signal;
+use Workflow\V2\Attributes\Type;
+use function Workflow\V2\await;
+use Workflow\V2\MessageStream;
+use Workflow\V2\Workflow;
+
+/**
+ * Like TestInboxTurnsWorkflow, but opens its inbox once and keeps it, so
+ * later turns receive through a stream created at an earlier step.
+ */
+#[Type('test-inbox-kept-stream-workflow')]
+#[Signal('ping')]
+final class TestInboxKeptStreamWorkflow extends Workflow
+{
+    private ?MessageStream $chat = null;
+
+    /**
+     * @return list<string|null>
+     */
+    public function handle(int $maxTurns = 2): array
+    {
+        $received = [];
+
+        while (count($received) < $maxTurns) {
+            await('ping');
+            $this->chat ??= $this->inbox('chat');
+            $message = $this->chat->receiveOne();
+            $received[] = $message?->payload_reference;
+            activity(TestRedriveFirstActivity::class, (string) $message?->payload_reference);
+        }
+
+        return $received;
+    }
+}

From a27bcc3f8b6a97c09fbdf44fb072d273547db92b Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Tue, 3 Nov 2026 22:14:30 -0600
Subject: [PATCH 4/6] Add replay corpus evidence for the empty-receive replay

The branching workflow from the #1011 review as a corpus fixture: turn
one finds the inbox empty, a message arrives, and turn two must replay
turn one's branch. On main the replay consumes the message at turn one's
step and the run stays waiting with a history mismatch.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../V2EmbeddedReplayRegressionCorpusTest.php  | 54 +++++++++++++++++++
 ...ge-stream-empty-receive-replays-empty.json | 34 ++++++++++++
 2 files changed, 88 insertions(+)
 create mode 100644 tests/Fixtures/V2/ReplayRegression/message-stream-empty-receive-replays-empty.json

diff --git a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php
index d5806e038..680163cfc 100644
--- a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php
+++ b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php
@@ -744,6 +744,60 @@ public function testRedrivePendingMessageCorpusReachesTheSuccessor(): void
         $this->assertSame($scenario['output'], $successor->workflowOutput());
     }
 
+    public function testMessageStreamEmptyReceiveCorpusReplaysTheBranchTheRunTook(): void
+    {
+        $path = self::FIXTURE_DIR . '/message-stream-empty-receive-replays-empty.json';
+        if (! is_file($path)) {
+            $this->markTestSkipped('The empty-receive replay evidence is absent.');
+        }
+        $fixture = json_decode(file_get_contents($path), true, flags: JSON_THROW_ON_ERROR);
+        $scenario = $fixture['message_stream_empty_receive_replays_empty'];
+        self::stopWorkers();
+        $this->clearWorkflowState();
+        Queue::fake();
+        config([
+            'workflows.v2.task_dispatch_mode' => 'poll',
+        ]);
+        Cache::forget('test:redrive:first-calls');
+        $drain = function (): void {
+            for ($i = 0; $i < 50; ++$i) {
+                $task = \Workflow\V2\Models\WorkflowTask::query()
+                    ->where('status', \Workflow\V2\Enums\TaskStatus::Ready->value)
+                    ->orderBy('created_at')
+                    ->first();
+                if ($task === null) {
+                    return;
+                }
+                $job = match ($task->task_type) {
+                    \Workflow\V2\Enums\TaskType::Workflow => new RunWorkflowTask($task->id),
+                    \Workflow\V2\Enums\TaskType::Activity => new \Workflow\V2\Jobs\RunActivityTask($task->id),
+                    \Workflow\V2\Enums\TaskType::Timer => new \Workflow\V2\Jobs\RunTimerTask($task->id),
+                };
+                $this->app->call([$job, 'handle']);
+            }
+        };
+
+        $instanceId = 'regression-corpus-empty-receive';
+        $stub = WorkflowStub::make($fixture['workflow']['type'], $instanceId);
+        $stub->start(...$fixture['workflow']['arguments']);
+        $drain();
+        $run = WorkflowRun::query()->where('workflow_instance_id', $instanceId)->sole();
+        $stub->signal($scenario['signal']);
+        $drain();
+        (new \Workflow\V2\Support\MessageService())->sendMessage(
+            $run,
+            \Workflow\V2\Enums\MessageChannel::WorkflowMessage,
+            $instanceId,
+            $scenario['arrives_after_first_turn'],
+            $scenario['stream'],
+        );
+        $stub->signal($scenario['signal']);
+        $drain();
+
+        $this->assertTrue($stub->refresh()->completed(), (string) $stub->refresh()->status());
+        $this->assertSame($scenario['output'], $stub->output());
+    }
+
     public function testRedriveTypedLineageCorpusLinksBothRunsFromHistory(): void
     {
         $path = self::FIXTURE_DIR . '/redrive-typed-lineage.json';
diff --git a/tests/Fixtures/V2/ReplayRegression/message-stream-empty-receive-replays-empty.json b/tests/Fixtures/V2/ReplayRegression/message-stream-empty-receive-replays-empty.json
new file mode 100644
index 000000000..1c4469292
--- /dev/null
+++ b/tests/Fixtures/V2/ReplayRegression/message-stream-empty-receive-replays-empty.json
@@ -0,0 +1,34 @@
+{
+  "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json",
+  "fixture_schema": "durable-workflow.replay-regression/v1",
+  "id": "message-stream-empty-receive-replays-empty",
+  "protocol_version": "1.0",
+  "bindings": ["php"],
+  "workflow": {
+    "type": "Tests\\Fixtures\\V2\\TestInboxBranchWorkflow",
+    "arguments": [2],
+    "payload_codec": "avro"
+  },
+  "message_stream_empty_receive_replays_empty": {
+    "issue": "durable-workflow/workflow#1011",
+    "stream": "chat",
+    "signal": "ping",
+    "arrives_after_first_turn": "m1",
+    "output": ["none:Hello, none!", "got:m1"]
+  },
+  "history": [
+    {
+      "sequence": 1,
+      "event_type": "WorkflowStarted",
+      "recorded_at": "2026-11-04T00:00:00+00:00",
+      "payload": {}
+    }
+  ],
+  "expected": {
+    "completed": false,
+    "result": null,
+    "commands": [
+      {"type": "open_signal_wait", "signal_name": "ping"}
+    ]
+  }
+}

From 9e3bf3697611c19454628474c6d31f5418764cfe Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Tue, 3 Nov 2026 23:06:22 -0600
Subject: [PATCH 5/6] Record a kept stream's receive at the step the code is at

General's re-review of a27bcc3f (residual H1): a stream kept in a
property recorded its consumption at the step it was opened at, while
the replay guard checked the live step. After an empty first turn, a
later turn's consumption was found again at the first turn's step on
replay, so straight-line runs returned it early and branching runs got
stuck.

Without an explicit step, a workflow-code receive now records and looks
up its consumption at the live step, falling back to the opened step so
rows earlier releases wrote still replay. Also: highestRecordedStep
reuses loaded history and accepts numeric-string steps, and two stale
'after cursor' lines in workflow-messages-architecture.md are fixed.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 CHANGELOG.md                                  |  4 +
 docs/workflow-messages-architecture.md        |  4 +-
 src/V2/MessageStream.php                      | 18 +++-
 src/V2/Support/MessageReplayLedger.php        | 17 +++-
 .../Feature/V2/V2MessageStreamReplayTest.php  | 82 +++++++++++++++++++
 .../V2/TestInboxKeptBranchWorkflow.php        | 44 ++++++++++
 .../V2/TestInboxKeptStreamWorkflow.php        | 23 ++++--
 7 files changed, 180 insertions(+), 12 deletions(-)
 create mode 100644 tests/Fixtures/V2/TestInboxKeptBranchWorkflow.php

diff --git a/CHANGELOG.md b/CHANGELOG.md
index f54a8f77b..cc9265a84 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -17,6 +17,10 @@
     run where an earlier release's cursor hid a message keeps the result it
     had, and receives the message at a later step instead of consuming it at
     the old one.
+  - A receive through a stream kept in a property records its consumption
+    at the step the code is at, not the step the stream was opened at, so
+    its replay matches too. Consumptions that earlier releases recorded at
+    the opened step still replay.
 - A redrive moves the failed run's pending inbound messages to the successor
   (#1013). Before, only the message cursor carried over, so a message that
   arrived while the run was failed stayed on the failed run and the successor
diff --git a/docs/workflow-messages-architecture.md b/docs/workflow-messages-architecture.md
index 2430004ea..ef52c36f5 100644
--- a/docs/workflow-messages-architecture.md
+++ b/docs/workflow-messages-architecture.md
@@ -544,7 +544,7 @@ foreach ($webhooks as $webhook) {
 
 [Receive Phase]  
 8. Receiver calls receiveMessages()
-9. Fetches pending inbound messages after cursor
+9. Fetches the stream's pending inbound messages
 10. Returns messages in sequence order
 
 [Consume Phase]
@@ -635,7 +635,7 @@ Uses index: `correlation_id`.
    - Batch consumption
 
 2. **Cursor Semantics**
-   - Only receive messages after cursor
+   - Receive every pending message of the stream, whatever the cursor says
    - Unconsumed count tracking
    - hasUnconsumedMessages check
 
diff --git a/src/V2/MessageStream.php b/src/V2/MessageStream.php
index 35a2df758..05272e343 100644
--- a/src/V2/MessageStream.php
+++ b/src/V2/MessageStream.php
@@ -120,9 +120,25 @@ public function receive(int $limit = 1, ?int $consumedBySequence = null): Collec
         $limit = $this->positiveLimit($limit);
 
         if ($this->replayLedger instanceof MessageReplayLedger) {
-            $sequence = $this->consumedBySequence($consumedBySequence);
+            // Without an explicit step, a receive records and finds its
+            // consumption at the step the code is at now, not the step a kept
+            // stream was opened at. Rows written by earlier releases sit at the
+            // opened step, so that step is the fallback.
+            $currentStep = $consumedBySequence === null && $this->currentStep instanceof Closure
+                ? (int) ($this->currentStep)()
+                : null;
+            $sequence = $currentStep ?? $this->consumedBySequence($consumedBySequence);
             $recorded = $this->recordedReceive($sequence, $limit);
 
+            if (
+                $recorded->isEmpty()
+                && $currentStep !== null
+                && $this->defaultConsumedBySequence !== null
+                && $this->defaultConsumedBySequence !== $currentStep
+            ) {
+                $recorded = $this->recordedReceive($this->defaultConsumedBySequence, $limit);
+            }
+
             // A receive this step already made returns what it consumed. A
             // receive at a step history has already passed, with nothing
             // recorded, found nothing the first time, so it consumes nothing
diff --git a/src/V2/Support/MessageReplayLedger.php b/src/V2/Support/MessageReplayLedger.php
index 6cd28636d..14cf5ecbd 100644
--- a/src/V2/Support/MessageReplayLedger.php
+++ b/src/V2/Support/MessageReplayLedger.php
@@ -101,15 +101,24 @@ public function highestRecordedStep(WorkflowRun $run): int
         }
 
         $highest = 0;
-
-        foreach (WorkflowHistoryEvent::query()
-            ->where('workflow_run_id', $run->id)
-            ->pluck('payload') as $payload) {
+        // The executor usually has the run's history loaded already.
+        $payloads = $run->relationLoaded('historyEvents')
+            ? $run->historyEvents->pluck('payload')
+            : WorkflowHistoryEvent::query()
+                ->where('workflow_run_id', $run->id)
+                ->pluck('payload');
+
+        foreach ($payloads as $payload) {
             if (is_string($payload)) {
                 $payload = json_decode($payload, true);
             }
             $sequence = is_array($payload) ? ($payload['sequence'] ?? null) : null;
 
+            // Imported histories can carry the step as a numeric string.
+            if (is_string($sequence) && ctype_digit($sequence)) {
+                $sequence = (int) $sequence;
+            }
+
             if (is_int($sequence) && $sequence > $highest) {
                 $highest = $sequence;
             }
diff --git a/tests/Feature/V2/V2MessageStreamReplayTest.php b/tests/Feature/V2/V2MessageStreamReplayTest.php
index 8431354f0..c446d2fff 100644
--- a/tests/Feature/V2/V2MessageStreamReplayTest.php
+++ b/tests/Feature/V2/V2MessageStreamReplayTest.php
@@ -7,6 +7,7 @@
 use Illuminate\Support\Facades\Cache;
 use Illuminate\Support\Facades\Queue;
 use Tests\Fixtures\V2\TestInboxBranchWorkflow;
+use Tests\Fixtures\V2\TestInboxKeptBranchWorkflow;
 use Tests\Fixtures\V2\TestInboxKeptStreamWorkflow;
 use Tests\Fixtures\V2\TestInboxTurnsWorkflow;
 use Tests\Fixtures\V2\TestMessageStreamExactContentWorkflow;
@@ -436,6 +437,87 @@ public function testAKeptInboxStillReceivesAtLaterTurns(): void
         $this->assertSame(['m1', 'm2'], $workflow->output());
     }
 
+    /**
+     * A kept inbox whose first turn found nothing records a later turn's
+     * consumption at that later step, so the first turn still replays as
+     * nothing.
+     */
+    public function testAKeptInboxWhoseFirstTurnWasEmptyReplaysThatTurnAsNothing(): void
+    {
+        $workflow = WorkflowStub::make(TestInboxKeptStreamWorkflow::class, 'inbox-kept-empty-first');
+        $workflow->start(2);
+        $this->drainReadyTasks();
+        $run = WorkflowRun::query()->where('workflow_instance_id', 'inbox-kept-empty-first')->sole();
+
+        $workflow->signal('ping');
+        $this->drainReadyTasks();
+        $this->sendToInbox($run, 'm1');
+        $workflow->signal('ping');
+        $this->drainReadyTasks();
+
+        $this->assertTrue($workflow->refresh()->completed());
+        $this->assertSame([null, 'm1'], $workflow->output());
+    }
+
+    /**
+     * Rows an earlier release wrote for a kept stream sit at the step the
+     * stream was opened at. A later turn finds its consumption there when
+     * nothing is recorded at its own step.
+     */
+    public function testAKeptInboxStillReplaysConsumptionsRecordedAtItsOpenedStep(): void
+    {
+        $workflow = WorkflowStub::make(TestInboxKeptStreamWorkflow::class, 'inbox-kept-legacy');
+        $workflow->start(3);
+        $this->drainReadyTasks();
+        $run = WorkflowRun::query()->where('workflow_instance_id', 'inbox-kept-legacy')->sole();
+
+        $workflow->signal('ping');
+        $this->sendToInbox($run, 'm1');
+        $this->drainReadyTasks();
+        $workflow->signal('ping');
+        $this->sendToInbox($run, 'm2');
+        $this->drainReadyTasks();
+        $m1 = WorkflowMessage::query()->where('payload_reference', 'm1')->where(
+            'direction',
+            MessageDirection::Inbound->value
+        )->sole();
+        $m2 = WorkflowMessage::query()->where('payload_reference', 'm2')->where(
+            'direction',
+            MessageDirection::Inbound->value
+        )->sole();
+        $this->assertNotSame($m1->consumed_by_sequence, $m2->consumed_by_sequence);
+
+        // As an earlier release recorded it: both at the step the stream opened.
+        $m2->forceFill([
+            'consumed_by_sequence' => $m1->consumed_by_sequence,
+        ])->save();
+
+        $this->assertSame(['m1', 'm2'], $workflow->refresh()->query('received'));
+    }
+
+    /**
+     * The same with a branch and a third turn, whose task replays the first
+     * two turns.
+     */
+    public function testAKeptInboxBranchReplaysAnEmptyFirstTurn(): void
+    {
+        $workflow = WorkflowStub::make(TestInboxKeptBranchWorkflow::class, 'inbox-kept-branch');
+        $workflow->start(3);
+        $this->drainReadyTasks();
+        $run = WorkflowRun::query()->where('workflow_instance_id', 'inbox-kept-branch')->sole();
+
+        $workflow->signal('ping');
+        $this->drainReadyTasks();
+        $this->sendToInbox($run, 'm1');
+        $workflow->signal('ping');
+        $this->drainReadyTasks();
+        $workflow->signal('ping');
+        $this->drainReadyTasks();
+
+        $this->assertTrue($workflow->refresh()->completed(), (string) $workflow->refresh()->status());
+        $this->assertSame(['none:Hello, none!', 'got:m1', 'none:Hello, none!'], $workflow->output());
+    }
+
     /**
      * The same history when workflow code branches on the receive: replay
      * takes the branch the original run took, so the run doesn't fail with a
diff --git a/tests/Fixtures/V2/TestInboxKeptBranchWorkflow.php b/tests/Fixtures/V2/TestInboxKeptBranchWorkflow.php
new file mode 100644
index 000000000..865513bb3
--- /dev/null
+++ b/tests/Fixtures/V2/TestInboxKeptBranchWorkflow.php
@@ -0,0 +1,44 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Tests\Fixtures\V2;
+
+use function Workflow\V2\activity;
+use Workflow\V2\Attributes\Signal;
+use Workflow\V2\Attributes\Type;
+use function Workflow\V2\await;
+use Workflow\V2\MessageStream;
+use function Workflow\V2\sideEffect;
+use Workflow\V2\Workflow;
+
+/**
+ * TestInboxBranchWorkflow with an inbox opened once and kept, so later turns
+ * receive through a stream opened at an earlier step.
+ */
+#[Type('test-inbox-kept-branch-workflow')]
+#[Signal('ping')]
+final class TestInboxKeptBranchWorkflow extends Workflow
+{
+    private ?MessageStream $chat = null;
+
+    /**
+     * @return list<string>
+     */
+    public function handle(int $maxTurns = 3): array
+    {
+        $outcomes = [];
+
+        while (count($outcomes) < $maxTurns) {
+            await('ping');
+            $this->chat ??= $this->inbox('chat');
+            $message = $this->chat->receiveOne();
+
+            $outcomes[] = $message === null
+                ? 'none:' . activity(TestRedriveFirstActivity::class, 'none')
+                : 'got:' . sideEffect(static fn (): string => (string) $message->payload_reference);
+        }
+
+        return $outcomes;
+    }
+}
diff --git a/tests/Fixtures/V2/TestInboxKeptStreamWorkflow.php b/tests/Fixtures/V2/TestInboxKeptStreamWorkflow.php
index 2d938a591..6071e2ae8 100644
--- a/tests/Fixtures/V2/TestInboxKeptStreamWorkflow.php
+++ b/tests/Fixtures/V2/TestInboxKeptStreamWorkflow.php
@@ -4,6 +4,7 @@
 
 namespace Tests\Fixtures\V2;
 
+use Workflow\QueryMethod;
 use function Workflow\V2\activity;
 use Workflow\V2\Attributes\Signal;
 use Workflow\V2\Attributes\Type;
@@ -21,21 +22,33 @@ final class TestInboxKeptStreamWorkflow extends Workflow
 {
     private ?MessageStream $chat = null;
 
+    /**
+     * @var list<string|null>
+     */
+    private array $received = [];
+
     /**
      * @return list<string|null>
      */
     public function handle(int $maxTurns = 2): array
     {
-        $received = [];
-
-        while (count($received) < $maxTurns) {
+        while (count($this->received) < $maxTurns) {
             await('ping');
             $this->chat ??= $this->inbox('chat');
             $message = $this->chat->receiveOne();
-            $received[] = $message?->payload_reference;
+            $this->received[] = $message?->payload_reference;
             activity(TestRedriveFirstActivity::class, (string) $message?->payload_reference);
         }
 
-        return $received;
+        return $this->received;
+    }
+
+    /**
+     * @return list<string|null>
+     */
+    #[QueryMethod('received')]
+    public function received(): array
+    {
+        return $this->received;
     }
 }

From 2c46077b787d8113a800e6be464f9257dfa8cf19 Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Wed, 4 Nov 2026 00:03:46 -0600
Subject: [PATCH 6/6] fix: let a newly applied update handler receive past
 later steps (#1011)

An update applied while the run waits at a later step runs its handler for
the first time at the update's recorded step. The past-step guard treated
that as a replay and returned nothing, so the message stayed pending.
Exempt live update handlers from the guard. Query replay of update handler
receives is tracked in #1024.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 CHANGELOG.md                                  |  4 +-
 docs/architecture/execution-guarantees.md     |  4 +-
 src/V2/MessageStream.php                      |  9 +++
 src/V2/Support/WorkflowExecutor.php           |  4 +-
 src/V2/Workflow.php                           | 22 +++++++
 .../Feature/V2/V2MessageStreamReplayTest.php  | 39 ++++++++++++
 .../V2/TestInboxUpdateReceiveWorkflow.php     | 60 +++++++++++++++++++
 7 files changed, 139 insertions(+), 3 deletions(-)
 create mode 100644 tests/Fixtures/V2/TestInboxUpdateReceiveWorkflow.php

diff --git a/CHANGELOG.md b/CHANGELOG.md
index cc9265a84..a925b80ac 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -16,7 +16,9 @@
     again. It no longer reads the stream at a step history has passed. So a
     run where an earlier release's cursor hid a message keeps the result it
     had, and receives the message at a later step instead of consuming it at
-    the old one.
+    the old one. An update handler that runs for the first time still
+    receives, even when its update is recorded at an earlier step than the
+    one the run waits at.
   - A receive through a stream kept in a property records its consumption
     at the step the code is at, not the step the stream was opened at, so
     its replay matches too. Consumptions that earlier releases recorded at
diff --git a/docs/architecture/execution-guarantees.md b/docs/architecture/execution-guarantees.md
index a034d560e..06681eada 100644
--- a/docs/architecture/execution-guarantees.md
+++ b/docs/architecture/execution-guarantees.md
@@ -260,7 +260,9 @@ own identifier and its own durable row.
   messages already consumed at that step. A query or other read-only
   replay never writes or consumes. A replayed receive whose original
   call found nothing returns nothing again; a receive never consumes at a
-  step that history has already passed.
+  step that history has already passed. An update handler that runs for
+  the first time is not a replay, so its receives consume even when its
+  update is recorded at an earlier step.
 
 ## Side effects and version markers
 
diff --git a/src/V2/MessageStream.php b/src/V2/MessageStream.php
index 05272e343..a5243fec4 100644
--- a/src/V2/MessageStream.php
+++ b/src/V2/MessageStream.php
@@ -32,6 +32,7 @@ public function __construct(
         private readonly ?MessageReplayLedger $replayLedger = null,
         private readonly bool|Closure $readOnly = false,
         private readonly ?Closure $currentStep = null,
+        private readonly ?Closure $runningLiveUpdate = null,
     ) {
         if ($streamKey === '') {
             throw new InvalidArgumentException('Message stream key must not be empty.');
@@ -68,6 +69,7 @@ public static function forWorkflowCode(
         MessageReplayLedger $replayLedger,
         Closure $readOnly,
         ?Closure $currentStep = null,
+        ?Closure $runningLiveUpdate = null,
     ): self {
         return new self(
             $run,
@@ -77,6 +79,7 @@ public static function forWorkflowCode(
             $replayLedger,
             $readOnly,
             $currentStep,
+            $runningLiveUpdate,
         );
     }
 
@@ -374,6 +377,12 @@ private function historyPassedCurrentStep(): bool
             return false;
         }
 
+        // A newly applied update handler runs at its recorded step for the
+        // first time, so it isn't replaying even when history has moved on.
+        if ($this->runningLiveUpdate instanceof Closure && ($this->runningLiveUpdate)() === true) {
+            return false;
+        }
+
         return (int) ($this->currentStep)() <= $this->replayLedger->highestRecordedStep($this->run);
     }
 
diff --git a/src/V2/Support/WorkflowExecutor.php b/src/V2/Support/WorkflowExecutor.php
index b5411ae76..3cfdbb203 100644
--- a/src/V2/Support/WorkflowExecutor.php
+++ b/src/V2/Support/WorkflowExecutor.php
@@ -6593,7 +6593,9 @@ private function applyPendingUpdates(
                 if ($command instanceof WorkflowCommand && ExecutionTracing::enabled()) {
                     ExecutionTracing::linkUpdate($run, $command);
                 }
-                $result = $workflow->{$resolvedTarget['method']}(...$parameters);
+                $result = $workflow->runLiveUpdateHandler(
+                    static fn (): mixed => $workflow->{$resolvedTarget['method']}(...$parameters),
+                );
 
                 $appliedEvent = WorkflowHistoryEvent::record($run, HistoryEventType::UpdateApplied, [
                     'workflow_command_id' => $command?->id,
diff --git a/src/V2/Workflow.php b/src/V2/Workflow.php
index 10b1891df..9293c8cac 100644
--- a/src/V2/Workflow.php
+++ b/src/V2/Workflow.php
@@ -63,6 +63,8 @@ abstract class Workflow
 
     private ?MessageReplayLedger $messageReplayLedger = null;
 
+    private bool $runningLiveUpdate = false;
+
     /**
      * @var list<callable>
      */
@@ -194,6 +196,7 @@ public function messages(?string $streamKey = null, ?MessageService $messages =
             $this->messageReplayLedger,
             readOnly: fn (): bool => ! $this->commandDispatchEnabled,
             currentStep: fn (): int => $this->visibleSequence,
+            runningLiveUpdate: fn (): bool => $this->runningLiveUpdate,
         );
     }
 
@@ -282,6 +285,25 @@ public function executionCursor(): int
         return $this->visibleSequence;
     }
 
+    /**
+     * Run a newly applied update handler. Its code runs for the first time,
+     * at the step the update is recorded at, which history may already have
+     * passed, so its receives may consume.
+     *
+     * @internal
+     */
+    public function runLiveUpdateHandler(callable $handler): mixed
+    {
+        $previous = $this->runningLiveUpdate;
+        $this->runningLiveUpdate = true;
+
+        try {
+            return $handler();
+        } finally {
+            $this->runningLiveUpdate = $previous;
+        }
+    }
+
     public function setCommandDispatchEnabled(bool $enabled): void
     {
         $this->commandDispatchEnabled = $enabled;
diff --git a/tests/Feature/V2/V2MessageStreamReplayTest.php b/tests/Feature/V2/V2MessageStreamReplayTest.php
index c446d2fff..09ee6c021 100644
--- a/tests/Feature/V2/V2MessageStreamReplayTest.php
+++ b/tests/Feature/V2/V2MessageStreamReplayTest.php
@@ -10,6 +10,7 @@
 use Tests\Fixtures\V2\TestInboxKeptBranchWorkflow;
 use Tests\Fixtures\V2\TestInboxKeptStreamWorkflow;
 use Tests\Fixtures\V2\TestInboxTurnsWorkflow;
+use Tests\Fixtures\V2\TestInboxUpdateReceiveWorkflow;
 use Tests\Fixtures\V2\TestMessageStreamExactContentWorkflow;
 use Tests\Fixtures\V2\TestMessageStreamKeptStreamWorkflow;
 use Tests\Fixtures\V2\TestMessageStreamRedriveLaterReceiveWorkflow;
@@ -548,6 +549,44 @@ public function testABranchOnAnEmptyReceiveReplaysTheBranchTheRunTook(): void
         );
     }
 
+    public function testAnUpdateAppliedAfterLaterStepsStillReceivesAndReplaysWhatItTook(): void
+    {
+        $workflow = WorkflowStub::make(TestInboxUpdateReceiveWorkflow::class, 'inbox-update');
+        $workflow->start();
+        $this->drainReadyTasks();
+        $run = WorkflowRun::query()->where('workflow_instance_id', 'inbox-update')->sole();
+
+        $workflow->signal('go');
+        $this->drainReadyTasks();
+        $this->assertSame('waiting', $workflow->refresh()->status());
+
+        // The run now waits at a step after the one the update is recorded
+        // at; the handler runs for the first time, so it consumes.
+        $this->sendToInbox($run, 'j1');
+        $update = $workflow->attemptUpdate('take');
+
+        $this->assertTrue($update->completed(), (string) $update->outcome());
+        $this->assertSame('j1', $update->result());
+
+        $this->sendToInbox($run, 'j2');
+        $this->assertSame('j2', $workflow->attemptUpdate('take')->result());
+
+        // A query consumes nothing. What it returns for update handler
+        // receives is tracked separately in #1024.
+        $before = $this->messageRows();
+        $workflow->refresh()
+            ->query('taken');
+        $this->assertSame($before, $this->messageRows());
+
+        $workflow->signal('done');
+        $this->drainReadyTasks();
+
+        $this->assertTrue($workflow->refresh()->completed(), (string) $workflow->refresh()->status());
+        // Replays of both handlers return what each one took.
+        $this->assertSame(['j1', 'j2'], $workflow->output());
+        $this->assertSame($before, $this->messageRows());
+    }
+
     public function testAWorkflowCodeReceiveReplaysTheMessageItConsumedAndQueriesConsumeNothing(): void
     {
         $workflow = WorkflowStub::make(TestInboxTurnsWorkflow::class, 'inbox-turns');
diff --git a/tests/Fixtures/V2/TestInboxUpdateReceiveWorkflow.php b/tests/Fixtures/V2/TestInboxUpdateReceiveWorkflow.php
new file mode 100644
index 000000000..8fb53ed1d
--- /dev/null
+++ b/tests/Fixtures/V2/TestInboxUpdateReceiveWorkflow.php
@@ -0,0 +1,60 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Tests\Fixtures\V2;
+
+use Workflow\QueryMethod;
+use Workflow\UpdateMethod;
+use function Workflow\V2\activity;
+use Workflow\V2\Attributes\Signal;
+use Workflow\V2\Attributes\Type;
+use function Workflow\V2\await;
+use Workflow\V2\Workflow;
+
+/**
+ * An update handler receives from a stream while the run waits at a step
+ * after the one the update is recorded at.
+ */
+#[Type('test-inbox-update-receive-workflow')]
+#[Signal('go')]
+#[Signal('done')]
+final class TestInboxUpdateReceiveWorkflow extends Workflow
+{
+    /**
+     * @var list<string|null>
+     */
+    private array $taken = [];
+
+    /**
+     * @return list<string|null>
+     */
+    public function handle(): array
+    {
+        await('go');
+        activity(TestRedriveFirstActivity::class, 'step');
+        await('done');
+
+        return $this->taken;
+    }
+
+    #[UpdateMethod]
+    public function take(): ?string
+    {
+        $message = $this->inbox('chat')
+            ->receiveOne();
+        $reference = $message === null ? null : (string) $message->payload_reference;
+        $this->taken[] = $reference;
+
+        return $reference;
+    }
+
+    /**
+     * @return list<string|null>
+     */
+    #[QueryMethod]
+    public function taken(): array
+    {
+        return $this->taken;
+    }
+}