#362: loud typed codec-mismatch errors at ingress and self-describing Avro export

durable-workflow-ops · claude · durable-workflow-ops · commit 386676805c9f · 2026-04-17T13:19:38.000Z
Two pieces of the release-gating Avro parity surface:

1. Loud, typed codec-mismatch ingress errors

Add Workflow\Serializers\CodecDecodeException — a typed exception that
names the declared codec, what the decoder actually saw, and a
remediation hint. The Avro and Json codecs now throw it instead of a
generic RuntimeException when the bytes don't match the declared codec.

The ingress diagnosers cover the two most common cross-codec mistakes:
 - JSON bytes labeled `avro` (producer JSON-encoded but tagged it Avro)
 - Avro bytes labeled `json` (producer Avro-encoded but tagged it JSON)

In both cases the message names the codec, describes the symptom, and
tells the operator how to fix it ("change the codec tag to X" or
"re-encode the payload"). The wrapping ValidationException at the HTTP
ingress (PayloadEnvelopeResolver) preserves the typed message verbatim
so it shows up in API responses without losing context.

Covers acceptance criteria from the issue:
 - "Sending JSON bytes under an `avro` codec tag produces a loud, typed
   error — not silent garbage"
 - "Sending Avro bytes under a `json` codec tag produces a loud, typed
   error"
 - "Avro decode failure names the codec, expected schema, and a
   remediation — not a generic RuntimeException"

2. Self-describing Avro history-export bundles

HistoryExport bundles now include a top-level `codec_schemas` map. When
the bundle contains any payload encoded with the Avro codec (whether on
the run, commands, updates, signals, or tasks), the map embeds the
generic-wrapper schema JSON plus the documented prefix bytes (0x00
generic wrapper, 0x01 typed schema). An offline consumer reading the
export — without the workflow runtime in scope — can decode the
0x00-prefixed Avro payloads using only the bundle.

Bundles with no Avro payloads keep `codec_schemas: {}` so the field is
always present and shape-stable, simplifying consumer code.

Tests: 7 new ingress-mismatch unit tests; 2 new HistoryExport tests
covering the Avro-present and Avro-absent codec_schemas cases. All 130
tests across the changed surface pass.

Co-Authored-By: Claude Opus 4.7 (1M context) &lt;noreply@anthropic.com&gt;
diff --git a/src/Serializers/Avro.php b/src/Serializers/Avro.php
@@ -10,7 +10,7 @@
 use Apache\Avro\Datum\AvroIODatumWriter;
 use Apache\Avro\IO\AvroStringIO;
 use Apache\Avro\Schema\AvroSchema;
-use RuntimeException;
+use Throwable;
 
 /**
  * Avro binary codec with optional schema support.
@@ -37,6 +37,24 @@ final class Avro implements SerializerInterface
      */
     private const WRAPPER_SCHEMA = '{"type":"record","name":"Payload","namespace":"durable_workflow","fields":[{"name":"json","type":"string"},{"name":"version","type":"int","default":1}]}';
 
+    /**
+     * Stable wire-protocol prefix bytes documented for SDK / export consumers.
+     */
+    public const PREFIX_GENERIC_WRAPPER = "\x00";
+    public const PREFIX_TYPED_SCHEMA = "\x01";
+
+    /**
+     * The generic-wrapper schema as canonical JSON.
+     *
+     * Exposed so that history-export bundles and similar self-describing
+     * artifacts can embed the schema needed to decode `0x00`-prefixed
+     * Avro payloads offline, without coupling consumers to this class.
+     */
+    public static function wrapperSchemaJson(): string
+    {
+        return self::WRAPPER_SCHEMA;
+    }
+
     private static ?AvroSchema $wrapperSchema = null;
 
     /** @var AvroSchema|null Typed schema set by the caller for the current encode/decode. */
@@ -127,21 +145,46 @@ private static function decodeWithSchema(string $data, AvroSchema $schema): mixe
         return self::suppressDeprecations(function () use ($data, $schema): mixed {
             $bytes = base64_decode($data, true);
             if ($bytes === false) {
-                throw new RuntimeException('Failed to base64-decode Avro payload.');
+                self::failWithIngressDiagnosis($data);
             }
 
             $io = new AvroStringIO($bytes);
 
             // Read and verify prefix
             $prefix = $io->read(1);
             if ($prefix !== "\x01") {
-                throw new RuntimeException('Expected typed Avro payload (prefix 0x01), got: 0x' . bin2hex($prefix));
+                $schemaName = method_exists($schema, 'fullname') ? $schema->fullname() : null;
+                throw new CodecDecodeException(
+                    'avro',
+                    sprintf(
+                        'Expected typed Avro payload (prefix 0x01) for schema "%s", got prefix 0x%s.',
+                        $schemaName ?: 'inline',
+                        bin2hex($prefix),
+                    ),
+                    'Re-encode the payload with the typed Avro path against the matching writer schema, or change the codec tag to match the bytes you are sending.',
+                );
             }
 
-            $reader = new AvroIODatumReader($schema);
-            $decoder = new AvroIOBinaryDecoder($io);
-
-            return $reader->read($decoder);
+            try {
+                $reader = new AvroIODatumReader($schema);
+                $decoder = new AvroIOBinaryDecoder($io);
+
+                return $reader->read($decoder);
+            } catch (CodecDecodeException $e) {
+                throw $e;
+            } catch (Throwable $e) {
+                $schemaName = method_exists($schema, 'fullname') ? $schema->fullname() : null;
+                throw new CodecDecodeException(
+                    'avro',
+                    sprintf(
+                        'Avro datum reader failed against schema "%s": %s',
+                        $schemaName ?: 'inline',
+                        $e->getMessage(),
+                    ),
+                    'Verify the writer schema matches the bytes (resolution: writer→reader compatibility per Avro spec). If you intended a different schema, supply it via Avro::withSchema() before decoding.',
+                    $e,
+                );
+            }
         });
     }
 
@@ -175,29 +218,95 @@ private static function decodeWrapped(string $data): mixed
         return self::suppressDeprecations(function () use ($data): mixed {
             $bytes = base64_decode($data, true);
             if ($bytes === false) {
-                throw new RuntimeException('Failed to base64-decode Avro payload.');
+                self::failWithIngressDiagnosis($data);
             }
 
             $io = new AvroStringIO($bytes);
 
             // Read prefix
             $prefix = $io->read(1);
             if ($prefix === "\x01") {
-                throw new RuntimeException('Typed Avro payload requires a schema context. Call Avro::withSchema() before unserialize().');
+                throw new CodecDecodeException(
+                    'avro',
+                    'Typed Avro payload (prefix 0x01) decoded without a schema context.',
+                    'Call Avro::withSchema($writerSchema) before unserialize() so the typed payload can be read with its writer schema.',
+                );
             }
             if ($prefix !== "\x00") {
-                throw new RuntimeException('Unknown Avro payload prefix: 0x' . bin2hex($prefix));
+                throw new CodecDecodeException(
+                    'avro',
+                    sprintf(
+                        'Unknown Avro payload prefix: 0x%s (expected 0x00 generic wrapper or 0x01 typed schema).',
+                        bin2hex($prefix),
+                    ),
+                    'These bytes were not produced by Workflow\\Serializers\\Avro::serialize(). Re-encode the payload with the Avro codec, or change the codec tag if the producer used a different codec.',
+                );
             }
 
-            $schema = self::wrapperSchema();
-            $reader = new AvroIODatumReader($schema);
-            $decoder = new AvroIOBinaryDecoder($io);
-            $record = $reader->read($decoder);
-
-            return json_decode($record['json'], true, 512, JSON_THROW_ON_ERROR);
+            try {
+                $schema = self::wrapperSchema();
+                $reader = new AvroIODatumReader($schema);
+                $decoder = new AvroIOBinaryDecoder($io);
+                $record = $reader->read($decoder);
+
+                return json_decode($record['json'], true, 512, JSON_THROW_ON_ERROR);
+            } catch (CodecDecodeException $e) {
+                throw $e;
+            } catch (Throwable $e) {
+                throw new CodecDecodeException(
+                    'avro',
+                    'Generic Avro wrapper decode failed: ' . $e->getMessage(),
+                    'Re-encode the payload with the Avro codec (generic wrapper produces a JSON-string-inside-Avro envelope), or change the codec tag if the producer used a different codec.',
+                    $e,
+                );
+            }
         });
     }
 
+    /**
+     * Diagnose why the bytes labeled as Avro could not be base64-decoded
+     * and throw a {@see CodecDecodeException} with the most actionable hint.
+     *
+     * The most common ingress mistakes are:
+     *  - A producer JSON-encoded the payload but tagged it `avro`. The bytes
+     *    will start with a JSON character ({, [, ", -, digit, t, f, n) and
+     *    base64_decode() in strict mode rejects them outright.
+     *  - A producer sent raw binary Avro bytes without base64-encoding them.
+     */
+    private static function failWithIngressDiagnosis(string $data): never
+    {
+        if (self::looksLikeJson($data)) {
+            throw new CodecDecodeException(
+                'avro',
+                'Payload bytes look like JSON, not base64-encoded Avro.',
+                'The producer appears to have JSON-encoded the payload but tagged it with codec "avro". Either change the codec tag to "json", or re-encode the payload with Workflow\\Serializers\\Avro::serialize() before tagging it "avro".',
+            );
+        }
+
+        throw new CodecDecodeException(
+            'avro',
+            'Failed to base64-decode Avro payload bytes.',
+            'Avro payloads on the wire must be base64-encoded bytes whose first byte is 0x00 (generic wrapper) or 0x01 (typed schema). Re-encode the payload, or change the codec tag if the producer used a different codec.',
+        );
+    }
+
+    private static function looksLikeJson(string $data): bool
+    {
+        if ($data === '') {
+            return false;
+        }
+
+        $first = $data[0];
+        if ($first === '{' || $first === '[' || $first === '"') {
+            return true;
+        }
+        if ($first === '-' || ($first >= '0' && $first <= '9')) {
+            return true;
+        }
+
+        return in_array($data, ['true', 'false', 'null'], true);
+    }
+
     private static function wrapperSchema(): AvroSchema
     {
         if (self::$wrapperSchema === null) {
diff --git a/src/Serializers/CodecDecodeException.php b/src/Serializers/CodecDecodeException.php
@@ -0,0 +1,39 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Workflow\Serializers;
+
+use RuntimeException;
+use Throwable;
+
+/**
+ * Thrown when a payload labelled with a specific codec cannot be decoded.
+ *
+ * The exception names the declared codec, describes what the decoder
+ * actually saw, and includes a remediation hint so that operators looking
+ * at a wire-protocol or HTTP-API failure can reach the right answer
+ * without spelunking through the codec internals.
+ *
+ * Loud, typed ingress failures are required by the Avro release-gating
+ * acceptance criteria — a JSON blob arriving under an `avro` codec tag
+ * (or vice versa) must surface as a clearly attributable error instead
+ * of a generic RuntimeException with binary noise.
+ *
+ * @see https://github.com/zorporation/durable-workflow/issues/362
+ */
+final class CodecDecodeException extends RuntimeException
+{
+    public function __construct(
+        public readonly string $declaredCodec,
+        public readonly string $detail,
+        public readonly string $remediation,
+        ?Throwable $previous = null,
+    ) {
+        parent::__construct(
+            sprintf('%s Remediation: %s', $detail, $remediation),
+            0,
+            $previous,
+        );
+    }
+}
diff --git a/src/Serializers/Json.php b/src/Serializers/Json.php
@@ -59,7 +59,45 @@ public static function unserialize(string $data)
         try {
             return json_decode($data, true, 512, JSON_THROW_ON_ERROR);
         } catch (JsonException $e) {
-            throw new RuntimeException('Failed to JSON-decode payload: ' . $e->getMessage(), 0, $e);
+            if (self::looksLikeBase64Avro($data)) {
+                throw new CodecDecodeException(
+                    'json',
+                    'Payload bytes look like base64-encoded Avro, not JSON: ' . $e->getMessage(),
+                    'The blob is valid base64 starting with an Avro framing prefix (0x00 generic wrapper or 0x01 typed schema). Either change the codec tag to "avro", or re-encode the payload as JSON.',
+                    $e,
+                );
+            }
+
+            throw new CodecDecodeException(
+                'json',
+                'Failed to JSON-decode payload: ' . $e->getMessage(),
+                'Re-encode the payload as valid UTF-8 JSON (RFC 8259), or change the codec tag if a different codec produced these bytes.',
+                $e,
+            );
+        }
+    }
+
+    /**
+     * Heuristic: do these bytes look like base64-encoded Avro?
+     *
+     * The cheapest reliable check: pure base64 alphabet, base64_decode in
+     * strict mode succeeds, and the first decoded byte is 0x00 (generic
+     * Avro wrapper) or 0x01 (typed Avro schema). JSON ASCII text never
+     * decodes to bytes leading with 0x00/0x01 because base64 alphabet
+     * cannot represent control characters in source form, so this check
+     * has effectively no false positives on misformatted JSON.
+     */
+    private static function looksLikeBase64Avro(string $data): bool
+    {
+        if ($data === '' || preg_match('/^[A-Za-z0-9+\/]+={0,2}$/', $data) !== 1) {
+            return false;
+        }
+
+        $decoded = base64_decode($data, true);
+        if ($decoded === false || $decoded === '') {
+            return false;
         }
+
+        return $decoded[0] === "\x00" || $decoded[0] === "\x01";
     }
 }
diff --git a/src/V2/Support/HistoryExport.php b/src/V2/Support/HistoryExport.php
@@ -5,6 +5,7 @@
 namespace Workflow\V2\Support;
 
 use Carbon\CarbonInterface;
+use Workflow\Serializers\Avro;
 use Workflow\Serializers\CodecRegistry;
 use Closure;
 use InvalidArgumentException;
@@ -163,9 +164,65 @@ public static function forRun(
             ],
         ];
 
+        $bundle['codec_schemas'] = self::collectCodecSchemas($bundle);
+
         return self::withIntegrity(self::withRedaction($bundle, $run, $redactor));
     }
 
+    /**
+     * Collect the well-known wire schemas needed to decode the payloads in
+     * this bundle offline.
+     *
+     * For Avro: embeds the generic-wrapper schema (used by every codec=avro
+     * payload that does not carry a typed schema). Consumers reading the
+     * export without the workflow runtime can use this to decode the
+     * `0x00`-prefixed Avro payloads.
+     *
+     * For JSON and other self-describing codecs: the map is empty.
+     *
+     * @param  array<string, mixed>  $bundle
+     * @return array<string, array<string, string>>
+     */
+    private static function collectCodecSchemas(array $bundle): array
+    {
+        $schemas = [];
+
+        if (self::bundleUsesCodec($bundle, 'avro')) {
+            $schemas['avro'] = [
+                'wrapper_schema' => Avro::wrapperSchemaJson(),
+                'wrapper_prefix_hex' => '00',
+                'typed_prefix_hex' => '01',
+            ];
+        }
+
+        return $schemas;
+    }
+
+    /**
+     * @param  array<string, mixed>  $bundle
+     */
+    private static function bundleUsesCodec(array $bundle, string $codec): bool
+    {
+        $payloadsCodec = $bundle['payloads']['codec'] ?? null;
+        if ($payloadsCodec === $codec) {
+            return true;
+        }
+
+        foreach (['commands', 'updates', 'signals', 'tasks'] as $section) {
+            if (! isset($bundle[$section]) || ! is_array($bundle[$section])) {
+                continue;
+            }
+
+            foreach ($bundle[$section] as $row) {
+                if (is_array($row) && ($row['payload_codec'] ?? null) === $codec) {
+                    return true;
+                }
+            }
+        }
+
+        return false;
+    }
+
     /**
      * @param array<string, mixed> $bundle
      *
diff --git a/tests/Unit/Serializers/CodecMismatchIngressTest.php b/tests/Unit/Serializers/CodecMismatchIngressTest.php
diff --git a/tests/Unit/V2/HistoryExportTest.php b/tests/Unit/V2/HistoryExportTest.php
