Clarify durable message stream consumption

Clarify durable message stream consumption

Author: durable-workflow-ops

docs/api-stability.md

@@ -102,6 +102,25 @@ preference; both produce identical `Support\*` Call value objects.
 Adding new static methods to the facade is an additive (non-breaking)
 change. Removing or renaming a documented method is a major change.
 
+## Durable Message Stream Contract
+
+The v2 durable message service is the stable lower-level contract backing
+signals, updates, workflow-to-workflow messages, and repeated human-input
+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:
+  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.
+  Batch consumption is same-stream only; mixed-stream batches are rejected so
+  each `MessageCursorAdvanced` event names exactly one `stream_key`.
+- `MessageService::transferMessagesToContinuedRun()` moves pending inbound
+  messages and the cursor position from the closing run to the continued run.
+  Consumed messages stay attached to the original run as historical record.
+
 ## Continue-As-New Interleaving Contract
 
 Continue-as-new keeps one logical workflow instance while closing one run

docs/workflow-messages-architecture.md

@@ -182,11 +182,17 @@ Stream B: msg_seq_1, msg_seq_2, msg_seq_3
 
 ### Cursor Position
 
-Each workflow run tracks a cursor position per stream:
+Each workflow run tracks one monotonic cursor position across the instance
+message sequence:
 - `workflow_runs.message_cursor_position` - last consumed sequence
 - Default: 0 (no messages consumed yet)
 - Cursor only moves forward (monotonic)
 
+Stream keys scope receive and consume queries, but cursor advancement is still
+against the shared instance sequence. A single batch consume must therefore
+contain messages from one stream only; mixed-stream batches are rejected so the
+`MessageCursorAdvanced` event has one unambiguous `stream_key`.
+
 ### Cursor Advancement
 
 When a message is consumed:
@@ -198,9 +204,10 @@ When a message is consumed:
 
 **Idempotency:** `advanceCursor()` returns null if cursor already at or past requested position.
 
-### Receiving Messages
+### Peeking And Receiving Messages
 
 ```php
+$messages = $messageService->peekMessages($run, $streamKey);
 $messages = $messageService->receiveMessages($run, $streamKey);
 ```
 
@@ -211,7 +218,11 @@ Returns messages where:
 - `consume_state = 'pending'`
 - `sequence > $run->message_cursor_position`
 
-**Only unconsumed messages after cursor are returned.**
+`peekMessages()` and `receiveMessages()` are read-only. They return only
+unconsumed messages after the cursor, 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.
 
 ## Continue-As-New Handoff
 
@@ -341,7 +352,14 @@ class MessageService
         ?\DateTimeInterface $expiresAt = null,
     ): WorkflowMessage;
 
-    // Receive unconsumed inbound messages
+    // Peek unconsumed inbound messages without consuming or moving the cursor.
+    public function peekMessages(
+        WorkflowRun $run,
+        ?string $streamKey = null,
+        int $limit = 100,
+    ): Collection<WorkflowMessage>;
+
+    // Compatibility alias for peekMessages().
     public function receiveMessages(
         WorkflowRun $run,
         ?string $streamKey = null,
@@ -355,7 +373,14 @@ class MessageService
         int $consumedBySequence,
     ): void;
 
-    // Consume multiple messages (cursor advances to highest sequence)
+    // Consume multiple same-stream messages (cursor advances to highest sequence)
+    public function consumeMessages(
+        WorkflowRun $run,
+        array $messages,
+        int $consumedBySequence,
+    ): void;
+
+    // Compatibility alias for consumeMessages().
     public function consumeMessageBatch(
         WorkflowRun $run,
         array $messages,

src/V2/Support/MessageService.php

@@ -136,15 +136,19 @@ public function sendMessage(
     }
 
     /**
-     * Receive unconsumed inbound messages for a workflow run.
+     * Peek at unconsumed inbound messages for a workflow run.
+     *
+     * Peeking is read-only: it never marks messages consumed and never advances
+     * the run cursor. Call {@see consumeMessage()} or {@see consumeMessages()}
+     * after workflow code has durably recorded how the messages were handled.
      *
      * @param WorkflowRun $run The receiving workflow run
      * @param string|null $streamKey Stream key (defaults to instance stream)
      * @param int $limit Maximum number of messages to fetch
      *
      * @return \Illuminate\Database\Eloquent\Collection<WorkflowMessage>
      */
-    public function receiveMessages(
+    public function peekMessages(
         WorkflowRun $run,
         ?string $streamKey = null,
         int $limit = 100,
@@ -155,6 +159,26 @@ public function receiveMessages(
         return WorkflowMessage::getUnconsumedForStream($run, $streamKey, $afterSequence, $limit);
     }
 
+    /**
+     * Compatibility alias for {@see peekMessages()}.
+     *
+     * "Receive" in the v2 message stream contract means "read pending
+     * messages without consuming them". Cursor movement remains explicit.
+     *
+     * @param WorkflowRun $run The receiving workflow run
+     * @param string|null $streamKey Stream key (defaults to instance stream)
+     * @param int $limit Maximum number of messages to fetch
+     *
+     * @return \Illuminate\Database\Eloquent\Collection<WorkflowMessage>
+     */
+    public function receiveMessages(
+        WorkflowRun $run,
+        ?string $streamKey = null,
+        int $limit = 100,
+    ): \Illuminate\Database\Eloquent\Collection {
+        return $this->peekMessages($run, $streamKey, $limit);
+    }
+
     /**
      * Consume a message and advance the cursor.
      *
@@ -196,7 +220,7 @@ public function consumeMessage(WorkflowRun $run, WorkflowMessage $message, int $
      * @param array<WorkflowMessage> $messages Messages to consume
      * @param int $consumedBySequence The history sequence that consumed these messages
      */
-    public function consumeMessageBatch(WorkflowRun $run, array $messages, int $consumedBySequence): void
+    public function consumeMessages(WorkflowRun $run, array $messages, int $consumedBySequence): void
     {
         if (empty($messages)) {
             return;
@@ -233,11 +257,19 @@ public function consumeMessageBatch(WorkflowRun $run, array $messages, int $cons
                     ));
                 }
 
+                if ($streamKey !== null && $message->stream_key !== $streamKey) {
+                    throw new InvalidArgumentException(sprintf(
+                        'Cannot consume messages from multiple streams in one batch (%s, %s).',
+                        $streamKey,
+                        $message->stream_key,
+                    ));
+                }
+
+                $streamKey ??= $message->stream_key;
                 $message->markConsumed($consumedBySequence);
 
                 if ($message->sequence > $maxSequence) {
                     $maxSequence = $message->sequence;
-                    $streamKey = $message->stream_key;
                 }
             }
 
@@ -248,6 +280,18 @@ public function consumeMessageBatch(WorkflowRun $run, array $messages, int $cons
         });
     }
 
+    /**
+     * Compatibility alias for {@see consumeMessages()}.
+     *
+     * @param WorkflowRun $run The consuming workflow run
+     * @param array<WorkflowMessage> $messages Messages to consume
+     * @param int $consumedBySequence The history sequence that consumed these messages
+     */
+    public function consumeMessageBatch(WorkflowRun $run, array $messages, int $consumedBySequence): void
+    {
+        $this->consumeMessages($run, $messages, $consumedBySequence);
+    }
+
     /**
      * Get count of unconsumed messages for a stream.
      *

tests/Unit/V2/MessageServiceTest.php

@@ -148,6 +148,26 @@ public function testItReceivesUnconsumedInboundMessages(): void
         $this->assertEquals([1, 2, 3], $sequences);
     }
 
+    public function testPeekMessagesIsReadOnlyAndReceiveIsCompatibilityAlias(): void
+    {
+        $sourceRun = $this->createRun();
+        $targetRun = $this->createRun();
+
+        $this->service->sendMessage($sourceRun, MessageChannel::Signal, $targetRun->workflow_instance_id);
+        $this->service->sendMessage($sourceRun, MessageChannel::Signal, $targetRun->workflow_instance_id);
+
+        $peeked = $this->service->peekMessages($targetRun);
+        $received = $this->service->receiveMessages($targetRun);
+
+        $this->assertEquals($peeked->pluck('id')->all(), $received->pluck('id')->all());
+
+        $targetRun->refresh();
+        $this->assertSame(0, (int) $targetRun->message_cursor_position);
+        $this->assertTrue($peeked->every(
+            static fn (WorkflowMessage $message): bool => $message->consume_state === MessageConsumeState::Pending
+        ));
+    }
+
     public function testItConsumesMessageAndAdvancesCursor(): void
     {
         $sourceRun = $this->createRun();
@@ -202,6 +222,41 @@ public function testItConsumesMessageBatchAndAdvancesToHighestSequence(): void
         $this->assertEquals(3, $targetRun->message_cursor_position);
     }
 
+    public function testItRejectsMixedStreamBatchConsumption(): void
+    {
+        $sourceRun = $this->createRun();
+        $targetRun = $this->createRun();
+
+        $this->service->sendMessage(
+            $sourceRun,
+            MessageChannel::Signal,
+            $targetRun->workflow_instance_id,
+            null,
+            'stream_a'
+        );
+        $this->service->sendMessage(
+            $sourceRun,
+            MessageChannel::Signal,
+            $targetRun->workflow_instance_id,
+            null,
+            'stream_b'
+        );
+
+        $messages = [
+            $this->service->receiveMessages($targetRun, 'stream_a')
+                ->first(),
+            $this->service->receiveMessages($targetRun, 'stream_b')
+                ->first(),
+        ];
+
+        $this->expectException(InvalidArgumentException::class);
+        $this->expectExceptionMessage(
+            'Cannot consume messages from multiple streams in one batch (stream_a, stream_b).'
+        );
+
+        $this->service->consumeMessages($targetRun, $messages, 10);
+    }
+
     public function testItOnlyReceivesMessagesAfterCursorPosition(): void
     {
         $sourceRun = $this->createRun();