feat(producer): version distributed plan protocol (#2777)

Author: jrusso1020

packages/aws-lambda/src/cdk/HyperframesRenderStack.snapshot.test.ts

@@ -64,6 +64,8 @@ const EXPECTED_NON_RETRYABLE_ERRORS = new Set([
   "BROWSER_GPU_NOT_SOFTWARE",
   "FONT_FETCH_FAILED",
   "PLAN_TOO_LARGE",
+  "PLAN_PROTOCOL_UNSUPPORTED",
+  "PlanProtocolUnsupportedError",
   "FORMAT_NOT_SUPPORTED_IN_DISTRIBUTED",
   "ChromeBinaryUnavailableError",
 ]);

packages/aws-lambda/src/cdk/HyperframesRenderStack.ts

@@ -200,6 +200,8 @@ export class HyperframesRenderStack extends Construct {
       "BROWSER_GPU_NOT_SOFTWARE",
       "FONT_FETCH_FAILED",
       "PLAN_TOO_LARGE",
+      "PLAN_PROTOCOL_UNSUPPORTED",
+      "PlanProtocolUnsupportedError",
       "FORMAT_NOT_SUPPORTED_IN_DISTRIBUTED",
       "ChromeBinaryUnavailableError",
     ];
@@ -208,12 +210,16 @@ export class HyperframesRenderStack extends Construct {
       "PLAN_HASH_MISMATCH",
       "S3_URI_NOT_ALLOWED",
       "BROWSER_GPU_NOT_SOFTWARE",
+      "PLAN_PROTOCOL_UNSUPPORTED",
+      "PlanProtocolUnsupportedError",
       "ChromeBinaryUnavailableError",
     ];
     const NON_RETRYABLE_ASSEMBLE = [
       "FFMPEG_VERSION_MISMATCH",
       "PLAN_HASH_MISMATCH",
       "S3_URI_NOT_ALLOWED",
+      "PLAN_PROTOCOL_UNSUPPORTED",
+      "PlanProtocolUnsupportedError",
       "FORMAT_NOT_SUPPORTED_IN_DISTRIBUTED",
       "ChromeBinaryUnavailableError",
     ];

packages/aws-lambda/src/handler.test.ts

@@ -18,7 +18,12 @@ import { afterEach, beforeEach, describe, expect, it, mock } from "bun:test";
 import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
-import type { AssembleResult, ChunkResult, PlanResult } from "@hyperframes/producer/distributed";
+import {
+  CURRENT_PLAN_PROTOCOL,
+  type AssembleResult,
+  type ChunkResult,
+  type PlanResult,
+} from "@hyperframes/producer/distributed";
 import type { AssembleEvent, LambdaEvent, PlanEvent, RenderChunkEvent } from "./events.js";
 import { handler, unwrapEvent } from "./handler.js";
 
@@ -157,6 +162,7 @@ describe("handler dispatch", () => {
         writeFileSync(join(planDir, "meta", "chunks.json"), "[]");
         return {
           planDir,
+          planProtocol: CURRENT_PLAN_PROTOCOL,
           planHash: "fakehash",
           chunkCount: 4,
           totalFrames: 720,
@@ -224,6 +230,7 @@ describe("handler dispatch", () => {
         writeFileSync(join(planDir, "meta", "chunks.json"), "[]");
         return {
           planDir,
+          planProtocol: CURRENT_PLAN_PROTOCOL,
           planHash: "fakehash",
           chunkCount: 1,
           totalFrames: 30,

packages/gcp-cloud-run/src/server.test.ts

@@ -18,7 +18,13 @@ import { afterEach, describe, expect, it } from "bun:test";
 import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
-import type { AssembleResult, ChunkResult, PlanResult } from "@hyperframes/producer/distributed";
+import {
+  CURRENT_PLAN_PROTOCOL,
+  PlanProtocolUnsupportedError,
+  type AssembleResult,
+  type ChunkResult,
+  type PlanResult,
+} from "@hyperframes/producer/distributed";
 import { asStorage, FakeGcs } from "./__fixtures__/fakeGcs.js";
 import type { AssembleEvent, CloudRunEvent, PlanEvent, RenderChunkEvent } from "./events.js";
 import { createApp, dispatch, type HandlerDeps, unwrapEvent } from "./server.js";
@@ -56,6 +62,7 @@ async function seedPlanTar(gcs: FakeGcs, uri: string, planHash: string): Promise
 
 const planResult: PlanResult = {
   planDir: "(set at call time)",
+  planProtocol: CURRENT_PLAN_PROTOCOL,
   planHash: PLAN_HASH,
   chunkCount: 3,
   totalFrames: 90,
@@ -295,6 +302,34 @@ describe("createApp HTTP mapping", () => {
     expect(body.error).toBe("PLAN_HASH_MISMATCH");
   });
 
+  it("returns 400 for an unsupported plan protocol", async () => {
+    const gcs = new FakeGcs();
+    await seedPlanTar(gcs, "gs://b/renders/r1/plan.tar.gz", PLAN_HASH);
+    const app = createApp(
+      depsWith(gcs, {
+        renderChunk: async () => {
+          throw new PlanProtocolUnsupportedError("unsupported test protocol");
+        },
+      }),
+    );
+    const res = await app.request("/", {
+      method: "POST",
+      headers: { "content-type": "application/json" },
+      body: JSON.stringify({
+        Action: "renderChunk",
+        PlanGcsUri: "gs://b/renders/r1/plan.tar.gz",
+        PlanHash: PLAN_HASH,
+        ChunkIndex: 0,
+        ChunkOutputGcsPrefix: "gs://b/renders/r1/",
+        Format: "mp4",
+      }),
+    });
+
+    expect(res.status).toBe(400);
+    const body = (await res.json()) as { error: string };
+    expect(body.error).toBe("PlanProtocolUnsupportedError");
+  });
+
   it("returns 500 for a retryable/unknown error", async () => {
     const gcs = new FakeGcs(); // plan tar NOT seeded → download fails (retryable)
     const app = createApp(depsWith(gcs));

packages/gcp-cloud-run/src/server.ts

@@ -584,10 +584,12 @@ const NON_RETRYABLE_ERROR_NAMES = new Set([
   // non-retryable list.
   "FormatNotSupportedInDistributedError",
   "PlanTooLargeError",
+  "PlanProtocolUnsupportedError",
   "RenderChunkValidationError",
   "FFMPEG_VERSION_MISMATCH",
   "FORMAT_NOT_SUPPORTED_IN_DISTRIBUTED",
   "PLAN_TOO_LARGE",
+  "PLAN_PROTOCOL_UNSUPPORTED",
   "BROWSER_GPU_NOT_SOFTWARE",
   "FONT_FETCH_FAILED",
   "ChromeBinaryUnavailableError",

packages/producer/src/distributed.ts

@@ -82,6 +82,25 @@ export {
 } from "./services/distributed/renderConfigValidation.js";
 export { hashProjectDir } from "./services/distributed/projectHash.js";
 
+// ── Plan protocol compatibility ────────────────────────────────────────────
+// Workers validate this descriptor before consuming layout-specific
+// artifacts. Missing descriptors remain compatible with legacy v1 plans.
+export {
+  CURRENT_PLAN_PROTOCOL,
+  DISTRIBUTED_RENDER_CAPABILITIES,
+  getDistributedRenderCapabilities,
+  PLAN_ARTIFACT_LAYOUT,
+  PLAN_HASH_SCHEMA,
+  PLAN_PROTOCOL_UNSUPPORTED,
+  PLAN_SCHEMA_VERSION,
+  PlanProtocolUnsupportedError,
+  readPlanProtocol,
+  type DistributedRenderCapabilities,
+  type PlanProtocolConsumerCapabilities,
+  type PlanProtocolDescriptor,
+  type PlanProtocolV1Descriptor,
+} from "./services/distributed/planProtocol.js";
+
 // ── Format union ────────────────────────────────────────────────────────────
 // Canonical output-format type. The aws-lambda package re-exports it so
 // CLI / adopter SDKs can derive runtime allowlists from one source.

packages/producer/src/index.ts

@@ -133,10 +133,23 @@ export {
 // separate subpath import.
 export {
   assemble,
+  CURRENT_PLAN_PROTOCOL,
+  DISTRIBUTED_RENDER_CAPABILITIES,
+  getDistributedRenderCapabilities,
+  PLAN_ARTIFACT_LAYOUT,
+  PLAN_HASH_SCHEMA,
+  PLAN_PROTOCOL_UNSUPPORTED,
+  PLAN_SCHEMA_VERSION,
   plan,
+  PlanProtocolUnsupportedError,
+  readPlanProtocol,
   renderChunk,
   type AssembleResult,
   type ChunkResult,
+  type DistributedRenderCapabilities,
   type DistributedRenderConfig,
+  type PlanProtocolConsumerCapabilities,
+  type PlanProtocolDescriptor,
+  type PlanProtocolV1Descriptor,
   type PlanResult,
 } from "./distributed.js";

packages/producer/src/services/distributed/assemble.ts

@@ -40,6 +40,7 @@ import { defaultLogger, type ProducerLogger } from "../../logger.js";
 import { formatExportFrameName } from "../../utils/paths.js";
 import { padOrTrimAudioToVideoFrameCount } from "../render/audioPadTrim.js";
 import type { ChunkSliceJson } from "../render/stages/freezePlan.js";
+import { DISTRIBUTED_RENDER_CAPABILITIES, readPlanProtocol } from "./planProtocol.js";
 import type { DistributedFormat } from "./shared.js";
 
 /**
@@ -56,6 +57,7 @@ export interface AssembleResult {
 
 /** Shape of the planDir's top-level `plan.json` — only the fields `assemble` needs. */
 interface PlanJsonForAssemble {
+  protocol?: unknown;
   planHash: string;
   totalFrames: number;
   hasAudio: boolean;
@@ -118,10 +120,11 @@ export async function assemble(
   if (!existsSync(planJsonPath)) {
     throw new Error(`[assemble] planDir missing plan.json: ${planJsonPath}`);
   }
+  const plan = JSON.parse(readFileSync(planJsonPath, "utf-8")) as PlanJsonForAssemble;
+  readPlanProtocol(plan, DISTRIBUTED_RENDER_CAPABILITIES.roles.assembler);
   if (!existsSync(chunksJsonPath)) {
     throw new Error(`[assemble] planDir missing meta/chunks.json: ${chunksJsonPath}`);
   }
-  const plan = JSON.parse(readFileSync(planJsonPath, "utf-8")) as PlanJsonForAssemble;
   const chunks = JSON.parse(readFileSync(chunksJsonPath, "utf-8")) as ChunkSliceJson[];
   if (chunkPaths.length !== chunks.length) {
     throw new Error(

packages/producer/src/services/distributed/plan.test.ts

@@ -21,6 +21,7 @@ import { tmpdir } from "node:os";
 import { join } from "node:path";
 import { recomputePlanHashFromPlanDir } from "../render/stages/freezePlan.js";
 import { RenderQualityError } from "../renderOrchestrator.js";
+import { CURRENT_PLAN_PROTOCOL } from "./planProtocol.js";
 import {
   applyDistributedAudioWarningPolicy,
   buildChunkSlices,
@@ -401,6 +402,7 @@ describe("plan() — golden planDir + planHash determinism", () => {
 
       // ── PlanResult contract ─────────────────────────────────────────────
       expect(result.planDir).toBe(planDir);
+      expect(result.planProtocol).toEqual(CURRENT_PLAN_PROTOCOL);
       expect(result.planHash).toMatch(/^[0-9a-f]{64}$/);
       expect(result.chunkCount).toBe(1);
       expect(result.totalFrames).toBe(30); // 1s @ 30fps
@@ -429,6 +431,7 @@ describe("plan() — golden planDir + planHash determinism", () => {
         unknown
       >;
       expect(planJson.planHash).toBe(result.planHash);
+      expect(planJson.protocol).toEqual(CURRENT_PLAN_PROTOCOL);
       expect(planJson.hasAudio).toBe(false);
       expect(planJson.totalFrames).toBe(result.totalFrames);
     },
@@ -516,8 +519,18 @@ describe("plan() — golden planDir + planHash determinism", () => {
       expect(recomputed).toBe(result.planHash);
       const planJson = JSON.parse(readFileSync(join(planDir, "plan.json"), "utf-8")) as {
         planHash: string;
+        protocol?: unknown;
       };
       expect(planJson.planHash).toBe(result.planHash);
+      expect(planJson.protocol).toEqual(CURRENT_PLAN_PROTOCOL);
+
+      delete planJson.protocol;
+      writeFileSync(join(planDir, "plan.json"), `${JSON.stringify(planJson, null, 2)}\n`, "utf-8");
+      expect(recomputePlanHashFromPlanDir(planDir)).toBe(result.planHash);
+
+      planJson.protocol = CURRENT_PLAN_PROTOCOL;
+      writeFileSync(join(planDir, "plan.json"), `${JSON.stringify(planJson, null, 2)}\n`, "utf-8");
+      expect(recomputePlanHashFromPlanDir(planDir)).toBe(result.planHash);
     },
     TIMEOUT_MS,
   );

packages/producer/src/services/distributed/plan.ts

@@ -80,6 +80,7 @@ import {
   readFfmpegVersion,
   readProducerVersion,
 } from "./shared.js";
+import { CURRENT_PLAN_PROTOCOL, type PlanProtocolV1Descriptor } from "./planProtocol.js";
 
 /**
  * Caller-supplied configuration for a distributed render. `fps`, `width`,
@@ -255,6 +256,7 @@ export interface DistributedRenderConfig {
  */
 export interface PlanResult {
   planDir: string;
+  planProtocol: Readonly<PlanProtocolV1Descriptor>;
   planHash: string;
   chunkCount: number;
   totalFrames: number;
@@ -1100,6 +1102,7 @@ export async function plan(
 
   return {
     planDir,
+    planProtocol: CURRENT_PLAN_PROTOCOL,
     planHash,
     chunkCount,
     totalFrames,