#331: decode replayed signal values with the run's pinned payload codec

WorkflowExecutor::signalValue() read the serialized signal payload from the
history event and decoded it with the codec-blind Serializer::unserialize().
For runs pinned to Avro (or any codec other than the deployment default),
this either returned a base64 blob where the workflow expected the decoded
value, or raised a RuntimeException at resume time. Signal payloads are
already written with $run->payload_codec on apply (line 1862), so the
asymmetry was purely on read.

- signalValue() now accepts an optional $run and delegates to the existing
  unserializePayloadWithRun() helper, which reads $run->payload_codec and
  falls back to the codec-blind sniffer when the column is empty (legacy
  history rows written before payload_codec was populated).
- Both replay call sites (applied signal, pending signal command) thread
  the current $run through so the pinned codec is used.
- Regression test exercises the private method via reflection with Avro,
  JSON, null-codec fallback, and missing-payload paths.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: durable-workflow-ops

src/V2/Support/WorkflowExecutor.php

@@ -621,7 +621,7 @@ public function run(WorkflowRun $run, WorkflowTask $task): ?WorkflowTask
                 if ($signalEvent !== null) {
                     try {
                         $this->syncWorkflowCursor($workflow, $sequence + 1);
-                        $current = $workflowExecution->send($this->signalValue($signalEvent));
+                        $current = $workflowExecution->send($this->signalValue($signalEvent, $run));
                     } catch (Throwable $throwable) {
                         $this->failRun($run, $task, $throwable, 'workflow_run', $run->id);
 
@@ -688,7 +688,7 @@ public function run(WorkflowRun $run, WorkflowTask $task): ?WorkflowTask
 
                     try {
                         $this->syncWorkflowCursor($workflow, $sequence + 1);
-                        $current = $workflowExecution->send($this->signalValue($signalEvent));
+                        $current = $workflowExecution->send($this->signalValue($signalEvent, $run));
                     } catch (Throwable $throwable) {
                         $this->failRun($run, $task, $throwable, 'workflow_run', $run->id);
 
@@ -1943,15 +1943,15 @@ private static function isInternalTimeoutTimerKind(mixed $value): bool
         return in_array($value, ['condition_timeout', 'signal_timeout'], true);
     }
 
-    private function signalValue(WorkflowHistoryEvent $event): mixed
+    private function signalValue(WorkflowHistoryEvent $event, ?WorkflowRun $run = null): mixed
     {
         $serialized = $event->payload['value'] ?? null;
 
         if (! is_string($serialized)) {
             return null;
         }
 
-        return Serializer::unserialize($serialized);
+        return $this->unserializePayloadWithRun($serialized, $run);
     }
 
     private function signalPayloadValue(WorkflowCommand $command): mixed

tests/Unit/V2/SignalValueCodecTest.php

@@ -0,0 +1,92 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Tests\Unit\V2;
+
+use Orchestra\Testbench\TestCase;
+use ReflectionMethod;
+use Workflow\Serializers\Serializer;
+use Workflow\V2\Models\WorkflowHistoryEvent;
+use Workflow\V2\Models\WorkflowRun;
+use Workflow\V2\Support\WorkflowExecutor;
+
+/**
+ * #331 regression: signalValue() must decode the serialized signal payload
+ * using the run's pinned payload_codec. Previously it called the codec-blind
+ * Serializer::unserialize(), which would silently mis-decode an Avro-encoded
+ * signal as JSON (yielding a base64 blob or a RuntimeException) on Avro-pinned
+ * runs. The fix is to thread the run through signalValue() and delegate to
+ * unserializePayloadWithRun().
+ */
+final class SignalValueCodecTest extends TestCase
+{
+    public function testSignalValueDecodesAvroEncodedPayloadWithRunCodec(): void
+    {
+        $value = ['approved' => true, 'source' => 'waterline'];
+
+        $event = new WorkflowHistoryEvent();
+        $event->payload = [
+            'value' => Serializer::serializeWithCodec('avro', $value),
+        ];
+
+        $run = new WorkflowRun();
+        $run->payload_codec = 'avro';
+
+        $this->assertSame($value, $this->invokeSignalValue($event, $run));
+    }
+
+    public function testSignalValueDecodesJsonEncodedPayloadWithRunCodec(): void
+    {
+        $value = ['count' => 3];
+
+        $event = new WorkflowHistoryEvent();
+        $event->payload = [
+            'value' => Serializer::serializeWithCodec('json', $value),
+        ];
+
+        $run = new WorkflowRun();
+        $run->payload_codec = 'json';
+
+        $this->assertSame($value, $this->invokeSignalValue($event, $run));
+    }
+
+    public function testSignalValueFallsBackToCodecBlindWhenRunCodecUnavailable(): void
+    {
+        $value = ['legacy' => 'payload'];
+
+        $event = new WorkflowHistoryEvent();
+        $event->payload = [
+            'value' => Serializer::serializeWithCodec('json', $value),
+        ];
+
+        // Legacy rows written before payload_codec was populated must still
+        // round-trip via the codec-blind sniffer path.
+        $this->assertSame($value, $this->invokeSignalValue($event, null));
+    }
+
+    public function testSignalValueReturnsNullForMissingSerializedValue(): void
+    {
+        $event = new WorkflowHistoryEvent();
+        $event->payload = [];
+
+        $run = new WorkflowRun();
+        $run->payload_codec = 'avro';
+
+        $this->assertNull($this->invokeSignalValue($event, $run));
+    }
+
+    private function invokeSignalValue(WorkflowHistoryEvent $event, ?WorkflowRun $run): mixed
+    {
+        $executor = new WorkflowExecutor();
+        $method = new ReflectionMethod($executor, 'signalValue');
+        $method->setAccessible(true);
+
+        return $method->invoke($executor, $event, $run);
+    }
+
+    protected function getPackageProviders($app)
+    {
+        return [\Workflow\Providers\WorkflowServiceProvider::class];
+    }
+}