v2 executor: pin activity result encode and child run codec to parent run

Two paired drifts surfaced once #330 made avro the default codec:

- ActivityOutcomeRecorder serialized activity results with the package
  default (Serializer::serialize) and stamped no codec on the resulting
  activity row, so when the workflow side decoded the result via the
  run's pinned codec, the bytes did not match the codec — for plain-array
  starts the run is json but the result was avro bytes labeled json.
  The recorder now uses the parent run's payload_codec to encode results
  (and exception payloads) when the worker has not specified one. A
  worker-supplied codec still wins (the worker is sending opaque bytes
  it serialized itself).

- WorkflowExecutor::startChildWorkflow encoded child arguments with the
  parent run codec but stamped the child run with
  CodecRegistry::defaultCodec(). When parent and default differ (e.g.
  parent is json from a plain-array start and default is avro), the
  child run row labels json bytes as avro and the next read fails with
  "Failed to base64-decode Avro payload." The child run now inherits
  the parent's codec so the bytes and the label agree.

- WorkflowExecutor::activityResult and ::sideEffectResult took an event
  with no run context, so they unpacked the result blob with the
  codec-blind Serializer::unserialize. Both now accept the active run
  and dispatch through the run's pinned codec when available.

Each path falls back to the legacy codec-blind sniffer when the row
has no codec stamped, so pre-pinned history continues to replay.

Part of #331 (thread Avro through every payload surface). Retires the
two server activity- and child-workflow tests that started failing
when avro became the default.

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

Author: durable-workflow-ops

src/V2/Support/ActivityOutcomeRecorder.php

@@ -106,13 +106,19 @@ public static function record(
                 return self::recorded(null);
             }
 
+            $runCodec = is_string($run->payload_codec) && $run->payload_codec !== ''
+                ? $run->payload_codec
+                : null;
+
             if (in_array($run->status, [RunStatus::Completed, RunStatus::Failed], true)) {
                 $lockedExecution->forceFill([
                     'status' => $throwable === null ? ActivityStatus::Completed : ActivityStatus::Failed,
-                    'result' => $throwable === null ? Serializer::serialize($result) : $lockedExecution->result,
+                    'result' => $throwable === null
+                        ? self::serializeWithCodec($result, $codec, $runCodec)
+                        : $lockedExecution->result,
                     'exception' => $throwable === null
                         ? $lockedExecution->exception
-                        : Serializer::serialize(FailureFactory::payload($throwable)),
+                        : self::serializeWithCodec(FailureFactory::payload($throwable), null, $runCodec),
                     'closed_at' => $lockedExecution->closed_at ?? now(),
                 ])->save();
 
@@ -138,9 +144,7 @@ public static function record(
             $parallelMetadata = ParallelChildGroup::payloadForPath($parallelMetadataPath);
 
             if ($throwable === null) {
-                $serializedResult = $codec !== null
-                    ? $result
-                    : Serializer::serialize($result);
+                $serializedResult = self::serializeWithCodec($result, $codec, $runCodec);
 
                 $lockedExecution->forceFill([
                     'status' => ActivityStatus::Completed,
@@ -435,4 +439,27 @@ private static function shouldRetry(Throwable $throwable, int $attemptCount, int
         return ! $throwable instanceof NonRetryableExceptionContract
             && $attemptCount < $maxAttempts;
     }
+
+    /**
+     * Serialize an activity payload, preferring the worker-supplied codec
+     * (treats $value as already-serialized bytes), then the parent run's
+     * codec, then falling back to the package default.
+     *
+     * The encoded blob is stamped on the activity row; the workflow side
+     * later decodes it with the run codec (see ActivityExecution::activityResult).
+     * Pinning to the run codec keeps the bytes and the decode codec aligned
+     * even when the package default differs from the run codec.
+     */
+    private static function serializeWithCodec(mixed $value, ?string $workerCodec, ?string $runCodec): string
+    {
+        if (is_string($workerCodec) && $workerCodec !== '' && is_string($value)) {
+            return $value;
+        }
+
+        if (is_string($runCodec) && $runCodec !== '') {
+            return Serializer::serializeWithCodec($runCodec, $value);
+        }
+
+        return Serializer::serialize($value);
+    }
 }

src/V2/Support/WorkflowExecutor.php

@@ -142,7 +142,7 @@ public function run(WorkflowRun $run, WorkflowTask $task): ?WorkflowTask
                     try {
                         $this->syncWorkflowCursor($workflow, $sequence + 1);
                         if ($activityCompletion->event_type === HistoryEventType::ActivityCompleted) {
-                            $current = $workflowExecution->send($this->activityResult($activityCompletion));
+                            $current = $workflowExecution->send($this->activityResult($activityCompletion, $run));
                         } else {
                             $failureId = $activityCompletion->payload['failure_id'] ?? null;
 
@@ -411,7 +411,7 @@ public function run(WorkflowRun $run, WorkflowTask $task): ?WorkflowTask
                     }
 
                     $this->syncWorkflowCursor($workflow, $sequence + 1);
-                    $current = $workflowExecution->send($this->sideEffectResult($sideEffectEvent));
+                    $current = $workflowExecution->send($this->sideEffectResult($sideEffectEvent, $run));
                 } catch (Throwable $throwable) {
                     $this->failRun($run, $task, $throwable, 'workflow_run', $run->id);
 
@@ -909,7 +909,7 @@ public function run(WorkflowRun $run, WorkflowTask $task): ?WorkflowTask
 
                         if ($activityCompletion !== null) {
                             if ($activityCompletion->event_type === HistoryEventType::ActivityCompleted) {
-                                $results[$offset] = $this->activityResult($activityCompletion);
+                                $results[$offset] = $this->activityResult($activityCompletion, $run);
 
                                 continue;
                             }
@@ -1426,7 +1426,14 @@ private function scheduleChildWorkflow(
         $commandContract = RunCommandContract::snapshot($childWorkflowCall->workflow);
         $now = now();
 
-        $serializedChildArguments = Serializer::serializeWithCodec($run->payload_codec, $metadata->arguments);
+        // Inherit the parent's run codec for the child so the encoded
+        // arguments (next line) match the codec stamped on the child run.
+        // Falling back to the package default when the parent has none
+        // keeps pre-pinned parents working.
+        $childCodec = is_string($run->payload_codec) && $run->payload_codec !== ''
+            ? $run->payload_codec
+            : CodecRegistry::defaultCodec();
+        $serializedChildArguments = Serializer::serializeWithCodec($childCodec, $metadata->arguments);
         StructuralLimits::guardPayloadSize($serializedChildArguments);
 
         /** @var WorkflowInstance $childInstance */
@@ -1448,7 +1455,7 @@ private function scheduleChildWorkflow(
             'visibility_labels' => null,
             'status' => RunStatus::Pending->value,
             'compatibility' => $run->compatibility ?? WorkerCompatibility::current(),
-            'payload_codec' => CodecRegistry::defaultCodec(),
+            'payload_codec' => $childCodec,
             'arguments' => $serializedChildArguments,
             'connection' => RoutingResolver::workflowConnection($childWorkflowCall->workflow, $metadata),
             'queue' => RoutingResolver::workflowQueue($childWorkflowCall->workflow, $metadata),
@@ -3120,25 +3127,39 @@ private function recordClosedParallelChildResolutions(
         return true;
     }
 
-    private function activityResult(WorkflowHistoryEvent $event): mixed
+    private function activityResult(WorkflowHistoryEvent $event, ?WorkflowRun $run = null): mixed
     {
         $serialized = $event->payload['result'] ?? null;
 
         if (! is_string($serialized)) {
             return null;
         }
 
-        return Serializer::unserialize($serialized);
+        return $this->unserializePayloadWithRun($serialized, $run);
     }
 
-    private function sideEffectResult(WorkflowHistoryEvent $event): mixed
+    private function sideEffectResult(WorkflowHistoryEvent $event, ?WorkflowRun $run = null): mixed
     {
         $serialized = $event->payload['result'] ?? null;
 
         if (! is_string($serialized)) {
             return null;
         }
 
+        return $this->unserializePayloadWithRun($serialized, $run);
+    }
+
+    /**
+     * Decode a payload bytes string using the run's pinned codec, falling
+     * back to the legacy codec-blind sniffer when no run codec is available
+     * (history events written before payload_codec was populated).
+     */
+    private function unserializePayloadWithRun(string $serialized, ?WorkflowRun $run): mixed
+    {
+        if ($run !== null && is_string($run->payload_codec) && $run->payload_codec !== '') {
+            return Serializer::unserializeWithCodec($run->payload_codec, $serialized);
+        }
+
         return Serializer::unserialize($serialized);
     }