Keep a stream's pending messages receivable whatever moves the run cursor

CHANGELOG.md

@@ -2,6 +2,28 @@
 
 ## 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,
+    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. 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
+    the opened step still replay.
+
 ## 2.20.2 - 2026-11-04
 
 - Rollback, embedded Workflow package: 2.20.2 changes no schema, history

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.

docs/architecture/execution-guarantees.md

@@ -259,7 +259,10 @@ 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. 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
 

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.
@@ -542,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]
@@ -633,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
 

src/V2/MessageStream.php

@@ -31,6 +31,8 @@
         private readonly ?int $defaultConsumedBySequence = null,
         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.');
@@ -59,14 +61,16 @@
      *
      * @internal
      */
     public static function forWorkflowCode(
         WorkflowRun $run,
         ?string $streamKey,
         ?MessageService $messages,
         int $defaultConsumedBySequence,
         MessageReplayLedger $replayLedger,
         Closure $readOnly,
+        ?Closure $currentStep = null,
+        ?Closure $runningLiveUpdate = null,
     ): self {
         return new self(
             $run,
             $streamKey ?? MessageStreamCursor::defaultStreamKey($run),
@@ -74,6 +78,8 @@
             $defaultConsumedBySequence,
             $replayLedger,
             $readOnly,
+            $currentStep,
+            $runningLiveUpdate,
         );
     }
 
@@ -117,11 +123,30 @@
         $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);
 
-            // A receive this step already made returns what it consumed.
-            if ($recorded->isNotEmpty() || $this->readOnly()) {
+            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
+            // now either.
+            if ($recorded->isNotEmpty() || $this->readOnly() || $this->historyPassedCurrentStep()) {
                 return $recorded;
             }
 
@@ -341,6 +366,26 @@
         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;
+        }
+
+        // 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);
+    }
+
     /**
      * Checked on every call: a stream kept across a replayed update handler,
      * where dispatch is off, must still write once dispatch is back on.

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,45 @@ 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;
+        // 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;
+            }
+        }
+
+        return $this->highestRecordedStep = $highest;
+    }
+
     public function sendOffset(string $identity): int
     {
         return $this->sendOffsets[$identity] ?? 0;

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);
     }

src/V2/Support/WorkflowExecutor.php

@@ -6595,7 +6595,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,

src/V2/Workflow.php

@@ -63,6 +63,8 @@ abstract class Workflow
 
     private ?MessageReplayLedger $messageReplayLedger = null;
 
+    private bool $runningLiveUpdate = false;
+
     /**
      * @var list<callable>
      */
@@ -193,6 +195,8 @@ public function messages(?string $streamKey = null, ?MessageService $messages =
             $this->visibleSequence,
             $this->messageReplayLedger,
             readOnly: fn (): bool => ! $this->commandDispatchEnabled,
+            currentStep: fn (): int => $this->visibleSequence,
+            runningLiveUpdate: fn (): bool => $this->runningLiveUpdate,
         );
     }
 
@@ -281,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;

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 testContinueAsNewPendingMessageCorpusReachesTheContinuedRun(): void
     {
         $path = self::FIXTURE_DIR . '/continue-as-new-pending-message-transfer.json';