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

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,

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

@@ -0,0 +1,296 @@
+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 { assemble } from "./assemble.js";
+import {
+  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,
+} from "./planProtocol.js";
+import {
+  CHUNK_INDEX_OUT_OF_RANGE,
+  renderChunk,
+  RenderChunkValidationError,
+} from "./renderChunk.js";
+
+const tempDirs: string[] = [];
+
+afterEach(() => {
+  for (const dir of tempDirs.splice(0)) {
+    rmSync(dir, { recursive: true, force: true });
+  }
+});
+
+function expectUnsupported(run: () => unknown): PlanProtocolUnsupportedError {
+  let caught: unknown;
+  try {
+    run();
+  } catch (error) {
+    caught = error;
+  }
+  expect(caught).toBeInstanceOf(PlanProtocolUnsupportedError);
+  expect((caught as PlanProtocolUnsupportedError).code).toBe(PLAN_PROTOCOL_UNSUPPORTED);
+  return caught as PlanProtocolUnsupportedError;
+}
+
+function createReaderPlan(options: {
+  protocol?: unknown;
+  includeProtocol?: boolean;
+  malformedDownstreamArtifacts?: boolean;
+  omitDownstreamArtifacts?: boolean;
+}): string {
+  const planDir = mkdtempSync(join(tmpdir(), "hf-plan-protocol-"));
+  tempDirs.push(planDir);
+
+  const planJson: Record<string, unknown> = {
+    planHash: "fake",
+    totalFrames: 1,
+    hasAudio: false,
+    dimensions: {
+      fpsNum: 30,
+      fpsDen: 1,
+      width: 16,
+      height: 16,
+      format: "png-sequence",
+    },
+  };
+  if (options.includeProtocol === true) {
+    planJson.protocol = options.protocol;
+  }
+
+  writeFileSync(join(planDir, "plan.json"), JSON.stringify(planJson), "utf-8");
+  if (options.omitDownstreamArtifacts === true) {
+    return planDir;
+  }
+
+  mkdirSync(join(planDir, "meta"), { recursive: true });
+  writeFileSync(
+    join(planDir, "meta", "encoder.json"),
+    options.malformedDownstreamArtifacts ? "{not-json" : "{}",
+    "utf-8",
+  );
+  writeFileSync(
+    join(planDir, "meta", "chunks.json"),
+    options.malformedDownstreamArtifacts
+      ? "{not-json"
+      : JSON.stringify([{ index: 0, startFrame: 0, endFrame: 1 }]),
+    "utf-8",
+  );
+  return planDir;
+}
+
+describe("readPlanProtocol()", () => {
+  it("treats an absent descriptor as legacy v1", () => {
+    expect(readPlanProtocol({ planHash: "legacy" })).toBe(CURRENT_PLAN_PROTOCOL);
+  });
+
+  it("enforces whether a worker accepts descriptor-less legacy v1 plans", () => {
+    const capabilities: PlanProtocolConsumerCapabilities = {
+      accepts: [CURRENT_PLAN_PROTOCOL],
+      acceptsLegacyV1WithoutDescriptor: false,
+    };
+
+    expectUnsupported(() => readPlanProtocol({ planHash: "legacy" }, capabilities));
+  });
+
+  it("enforces the worker's accepted protocol set", () => {
+    const capabilities: PlanProtocolConsumerCapabilities = {
+      accepts: [],
+      acceptsLegacyV1WithoutDescriptor: true,
+    };
+
+    expectUnsupported(() => readPlanProtocol({ protocol: CURRENT_PLAN_PROTOCOL }, capabilities));
+    expectUnsupported(() => readPlanProtocol({ planHash: "legacy" }, capabilities));
+  });
+
+  it("accepts the known v1 descriptor and ignores unknown optional fields", () => {
+    expect(
+      readPlanProtocol({
+        protocol: {
+          schemaVersion: PLAN_SCHEMA_VERSION,
+          artifactLayout: PLAN_ARTIFACT_LAYOUT,
+          hashSchema: PLAN_HASH_SCHEMA,
+          producerBuildId: "optional-future-metadata",
+        },
+      }),
+    ).toBe(CURRENT_PLAN_PROTOCOL);
+  });
+
+  it("rejects malformed and partial descriptors", () => {
+    for (const protocol of [
+      null,
+      [],
+      "v1",
+      {},
+      { schemaVersion: PLAN_SCHEMA_VERSION },
+      {
+        schemaVersion: PLAN_SCHEMA_VERSION,
+        artifactLayout: PLAN_ARTIFACT_LAYOUT,
+      },
+    ]) {
+      expectUnsupported(() => readPlanProtocol({ protocol }));
+    }
+  });
+
+  it("rejects unknown schema, layout, and hash-schema values", () => {
+    for (const protocol of [
+      { ...CURRENT_PLAN_PROTOCOL, schemaVersion: 2 },
+      { ...CURRENT_PLAN_PROTOCOL, artifactLayout: "plan-dir-v2" },
+      { ...CURRENT_PLAN_PROTOCOL, hashSchema: "hyperframes-plan-hash-v2" },
+    ]) {
+      expectUnsupported(() => readPlanProtocol({ protocol }));
+    }
+  });
+
+  it("keeps unsupported-protocol messages bounded and non-reflective", () => {
+    const untrustedValue = "secret-".repeat(1_000);
+    const error = expectUnsupported(() =>
+      readPlanProtocol({
+        protocol: { ...CURRENT_PLAN_PROTOCOL, hashSchema: untrustedValue },
+      }),
+    );
+
+    expect(error.message).not.toContain("secret-");
+    expect(error.message.length).toBeLessThan(200);
+  });
+});
+
+describe("getDistributedRenderCapabilities()", () => {
+  it("reports explicit v1 support for every distributed role", () => {
+    expect(getDistributedRenderCapabilities()).toBe(DISTRIBUTED_RENDER_CAPABILITIES);
+    expect(DISTRIBUTED_RENDER_CAPABILITIES).toEqual({
+      roles: {
+        planner: {
+          produces: [CURRENT_PLAN_PROTOCOL],
+        },
+        chunk: {
+          accepts: [CURRENT_PLAN_PROTOCOL],
+          acceptsLegacyV1WithoutDescriptor: true,
+        },
+        assembler: {
+          accepts: [CURRENT_PLAN_PROTOCOL],
+          acceptsLegacyV1WithoutDescriptor: true,
+        },
+      },
+    });
+  });
+
+  it("can express a v2 planner with dual-version readers", () => {
+    const futureV2: PlanProtocolDescriptor = {
+      schemaVersion: 2,
+      artifactLayout: "plan-dir-v2",
+      hashSchema: "hyperframes-plan-hash-v2",
+    };
+    const rolloutCapabilities: DistributedRenderCapabilities = {
+      roles: {
+        planner: {
+          produces: [futureV2],
+        },
+        chunk: {
+          accepts: [CURRENT_PLAN_PROTOCOL, futureV2],
+          acceptsLegacyV1WithoutDescriptor: true,
+        },
+        assembler: {
+          accepts: [CURRENT_PLAN_PROTOCOL, futureV2],
+          acceptsLegacyV1WithoutDescriptor: true,
+        },
+      },
+    };
+
+    expect(rolloutCapabilities.roles.planner.produces).toEqual([futureV2]);
+    expect(rolloutCapabilities.roles.chunk.accepts).toEqual([CURRENT_PLAN_PROTOCOL, futureV2]);
+    expect(rolloutCapabilities.roles.assembler.accepts).toEqual([CURRENT_PLAN_PROTOCOL, futureV2]);
+  });
+});
+
+describe("distributed plan protocol readers", () => {
+  it("renderChunk accepts a legacy plan without a descriptor", async () => {
+    const planDir = createReaderPlan({});
+
+    let caught: unknown;
+    try {
+      await renderChunk(planDir, 999, join(planDir, "unused-output"));
+    } catch (error) {
+      caught = error;
+    }
+
+    expect(caught).toBeInstanceOf(RenderChunkValidationError);
+    expect((caught as RenderChunkValidationError).code).toBe(CHUNK_INDEX_OUT_OF_RANGE);
+  });
+
+  it("assemble accepts a legacy plan without a descriptor", async () => {
+    const planDir = createReaderPlan({});
+    const missingChunk = join(planDir, "missing-chunk");
+
+    await expect(
+      assemble(planDir, [missingChunk], null, join(planDir, "unused-output")),
+    ).rejects.toThrow("chunk path does not exist");
+  });
+
+  it("renderChunk rejects an unknown v2 protocol before requiring v1 artifacts", async () => {
+    const planDir = createReaderPlan({
+      includeProtocol: true,
+      protocol: { ...CURRENT_PLAN_PROTOCOL, schemaVersion: 2 },
+      omitDownstreamArtifacts: true,
+    });
+
+    let caught: unknown;
+    try {
+      await renderChunk(planDir, 0, join(planDir, "unused-output"));
+    } catch (error) {
+      caught = error;
+    }
+
+    expect(caught).toBeInstanceOf(PlanProtocolUnsupportedError);
+    expect((caught as PlanProtocolUnsupportedError).code).toBe(PLAN_PROTOCOL_UNSUPPORTED);
+  });
+
+  it("assemble rejects an unknown v2 protocol before requiring v1 artifacts", async () => {
+    const planDir = createReaderPlan({
+      includeProtocol: true,
+      protocol: { ...CURRENT_PLAN_PROTOCOL, schemaVersion: 2 },
+      omitDownstreamArtifacts: true,
+    });
+
+    let caught: unknown;
+    try {
+      await assemble(planDir, [], null, join(planDir, "unused-output"));
+    } catch (error) {
+      caught = error;
+    }
+
+    expect(caught).toBeInstanceOf(PlanProtocolUnsupportedError);
+    expect((caught as PlanProtocolUnsupportedError).code).toBe(PLAN_PROTOCOL_UNSUPPORTED);
+  });
+
+  it("assemble rejects a partial protocol before parsing chunks", async () => {
+    const planDir = createReaderPlan({
+      includeProtocol: true,
+      protocol: {
+        schemaVersion: PLAN_SCHEMA_VERSION,
+        artifactLayout: PLAN_ARTIFACT_LAYOUT,
+      },
+      malformedDownstreamArtifacts: true,
+    });
+
+    let caught: unknown;
+    try {
+      await assemble(planDir, [], null, join(planDir, "unused-output"));
+    } catch (error) {
+      caught = error;
+    }
+
+    expect(caught).toBeInstanceOf(PlanProtocolUnsupportedError);
+    expect((caught as PlanProtocolUnsupportedError).code).toBe(PLAN_PROTOCOL_UNSUPPORTED);
+  });
+});

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

@@ -0,0 +1,155 @@
+/**
+ * Compatibility contract for the on-disk distributed render plan.
+ *
+ * A missing descriptor means the original v1 plan layout. This preserves
+ * replay compatibility with plan directories produced before the descriptor
+ * existed. Once a descriptor is present it must be complete and recognized:
+ * silently guessing across a partially-written or newer layout could render
+ * incorrect pixels or make assemble consume the wrong artifacts.
+ */
+
+export const PLAN_SCHEMA_VERSION = 1 as const;
+export const PLAN_ARTIFACT_LAYOUT = "plan-dir-v1" as const;
+export const PLAN_HASH_SCHEMA = "hyperframes-plan-hash-v1" as const;
+export const PLAN_PROTOCOL_UNSUPPORTED = "PLAN_PROTOCOL_UNSUPPORTED" as const;
+
+export interface PlanProtocolDescriptor {
+  readonly schemaVersion: number;
+  readonly artifactLayout: string;
+  readonly hashSchema: string;
+}
+
+export interface PlanProtocolV1Descriptor extends PlanProtocolDescriptor {
+  readonly schemaVersion: typeof PLAN_SCHEMA_VERSION;
+  readonly artifactLayout: typeof PLAN_ARTIFACT_LAYOUT;
+  readonly hashSchema: typeof PLAN_HASH_SCHEMA;
+}
+
+/** Descriptor written by the current producer and accepted by v1 workers. */
+export const CURRENT_PLAN_PROTOCOL: Readonly<PlanProtocolV1Descriptor> = Object.freeze({
+  schemaVersion: PLAN_SCHEMA_VERSION,
+  artifactLayout: PLAN_ARTIFACT_LAYOUT,
+  hashSchema: PLAN_HASH_SCHEMA,
+});
+
+export interface PlanProtocolConsumerCapabilities {
+  readonly accepts: readonly Readonly<PlanProtocolDescriptor>[];
+  readonly acceptsLegacyV1WithoutDescriptor: boolean;
+}
+
+export interface DistributedRenderCapabilities {
+  readonly roles: Readonly<{
+    planner: Readonly<{
+      produces: readonly Readonly<PlanProtocolDescriptor>[];
+    }>;
+    chunk: Readonly<PlanProtocolConsumerCapabilities>;
+    assembler: Readonly<PlanProtocolConsumerCapabilities>;
+  }>;
+}
+
+/** Serializable capability payload for fleet rollout and worker handshakes. */
+export const DISTRIBUTED_RENDER_CAPABILITIES: Readonly<DistributedRenderCapabilities> =
+  Object.freeze({
+    roles: Object.freeze({
+      planner: Object.freeze({
+        produces: Object.freeze([CURRENT_PLAN_PROTOCOL]),
+      }),
+      chunk: Object.freeze({
+        accepts: Object.freeze([CURRENT_PLAN_PROTOCOL]),
+        acceptsLegacyV1WithoutDescriptor: true,
+      }),
+      assembler: Object.freeze({
+        accepts: Object.freeze([CURRENT_PLAN_PROTOCOL]),
+        acceptsLegacyV1WithoutDescriptor: true,
+      }),
+    }),
+  });
+
+export function getDistributedRenderCapabilities(): Readonly<DistributedRenderCapabilities> {
+  return DISTRIBUTED_RENDER_CAPABILITIES;
+}
+
+/** Typed, deterministic compatibility failure. Retrying on the same worker cannot heal it. */
+export class PlanProtocolUnsupportedError extends Error {
+  // Public adapters inspect this typed code even though OSS producer does not.
+  // fallow-ignore-next-line unused-class-member
+  readonly code: typeof PLAN_PROTOCOL_UNSUPPORTED = PLAN_PROTOCOL_UNSUPPORTED;
+
+  constructor(reason: string) {
+    super(`[planProtocol] ${reason} (${PLAN_PROTOCOL_UNSUPPORTED})`);
+    this.name = "PlanProtocolUnsupportedError";
+  }
+}
+
+function isRecord(value: unknown): value is Record<string, unknown> {
+  return value !== null && typeof value === "object" && !Array.isArray(value);
+}
+
+function protocolMatches(
+  descriptor: Record<string, unknown>,
+  expected: PlanProtocolDescriptor,
+): boolean {
+  return (
+    descriptor.schemaVersion === expected.schemaVersion &&
+    descriptor.artifactLayout === expected.artifactLayout &&
+    descriptor.hashSchema === expected.hashSchema
+  );
+}
+
+function capabilitiesAccept(
+  capabilities: Readonly<PlanProtocolConsumerCapabilities>,
+  protocol: Readonly<PlanProtocolDescriptor>,
+): boolean {
+  return capabilities.accepts.some(
+    (accepted) =>
+      accepted.schemaVersion === protocol.schemaVersion &&
+      accepted.artifactLayout === protocol.artifactLayout &&
+      accepted.hashSchema === protocol.hashSchema,
+  );
+}
+
+/**
+ * Read and validate the plan protocol before a worker consumes other plan
+ * artifacts. Unknown fields on a recognized v1 descriptor are intentionally
+ * ignored so optional metadata can be added without a lockstep fleet deploy.
+ */
+export function readPlanProtocol(
+  planJson: unknown,
+  capabilities: Readonly<PlanProtocolConsumerCapabilities> = DISTRIBUTED_RENDER_CAPABILITIES.roles
+    .chunk,
+): Readonly<PlanProtocolV1Descriptor> {
+  if (!isRecord(planJson)) {
+    throw new PlanProtocolUnsupportedError("plan.json must contain a JSON object");
+  }
+
+  if (!Object.prototype.hasOwnProperty.call(planJson, "protocol")) {
+    if (
+      !capabilities.acceptsLegacyV1WithoutDescriptor ||
+      !capabilitiesAccept(capabilities, CURRENT_PLAN_PROTOCOL)
+    ) {
+      throw new PlanProtocolUnsupportedError(
+        "legacy v1 plan without a protocol descriptor is not accepted by this worker",
+      );
+    }
+    return CURRENT_PLAN_PROTOCOL;
+  }
+
+  const descriptor = planJson.protocol;
+  if (!isRecord(descriptor)) {
+    throw new PlanProtocolUnsupportedError("plan.json protocol descriptor must be an object");
+  }
+
+  for (const field of ["schemaVersion", "artifactLayout", "hashSchema"] as const) {
+    if (!Object.prototype.hasOwnProperty.call(descriptor, field)) {
+      throw new PlanProtocolUnsupportedError(`plan.json protocol descriptor is missing ${field}`);
+    }
+  }
+
+  if (
+    !protocolMatches(descriptor, CURRENT_PLAN_PROTOCOL) ||
+    !capabilitiesAccept(capabilities, CURRENT_PLAN_PROTOCOL)
+  ) {
+    throw new PlanProtocolUnsupportedError("unsupported plan.json protocol descriptor");
+  }
+  return CURRENT_PLAN_PROTOCOL;
+}

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

@@ -64,6 +64,34 @@ describe("@hyperframes/producer/distributed (subpath)", () => {
     expect(typeof distributedSubpath.applyRuntimeEnvSnapshot).toBe("function");
     expect(typeof distributedSubpath.readWebGlVendorInfoFromCanvas).toBe("function");
   });
+
+  it("exports the plan protocol contract", () => {
+    expect(distributedSubpath.PLAN_SCHEMA_VERSION).toBe(1);
+    expect(distributedSubpath.PLAN_ARTIFACT_LAYOUT).toBe("plan-dir-v1");
+    expect(distributedSubpath.PLAN_HASH_SCHEMA).toBe("hyperframes-plan-hash-v1");
+    expect(distributedSubpath.PLAN_PROTOCOL_UNSUPPORTED).toBe("PLAN_PROTOCOL_UNSUPPORTED");
+    expect(distributedSubpath.CURRENT_PLAN_PROTOCOL).toEqual({
+      schemaVersion: 1,
+      artifactLayout: "plan-dir-v1",
+      hashSchema: "hyperframes-plan-hash-v1",
+    });
+    expect(distributedSubpath.DISTRIBUTED_RENDER_CAPABILITIES.roles).toEqual({
+      planner: {
+        produces: [distributedSubpath.CURRENT_PLAN_PROTOCOL],
+      },
+      chunk: {
+        accepts: [distributedSubpath.CURRENT_PLAN_PROTOCOL],
+        acceptsLegacyV1WithoutDescriptor: true,
+      },
+      assembler: {
+        accepts: [distributedSubpath.CURRENT_PLAN_PROTOCOL],
+        acceptsLegacyV1WithoutDescriptor: true,
+      },
+    });
+    expect(typeof distributedSubpath.getDistributedRenderCapabilities).toBe("function");
+    expect(typeof distributedSubpath.readPlanProtocol).toBe("function");
+    expect(typeof distributedSubpath.PlanProtocolUnsupportedError).toBe("function");
+  });
 });
 
 describe("@hyperframes/producer (main entry)", () => {
@@ -73,6 +101,17 @@ describe("@hyperframes/producer (main entry)", () => {
     expect(typeof producerIndex.assemble).toBe("function");
   });
 
+  it("re-exports the plan protocol contract", () => {
+    expect(producerIndex.CURRENT_PLAN_PROTOCOL).toBe(distributedSubpath.CURRENT_PLAN_PROTOCOL);
+    expect(producerIndex.DISTRIBUTED_RENDER_CAPABILITIES).toBe(
+      distributedSubpath.DISTRIBUTED_RENDER_CAPABILITIES,
+    );
+    expect(typeof producerIndex.getDistributedRenderCapabilities).toBe("function");
+    expect(producerIndex.PLAN_PROTOCOL_UNSUPPORTED).toBe("PLAN_PROTOCOL_UNSUPPORTED");
+    expect(typeof producerIndex.readPlanProtocol).toBe("function");
+    expect(typeof producerIndex.PlanProtocolUnsupportedError).toBe("function");
+  });
+
   it("preserves the existing in-process exports (executeRenderJob unchanged)", () => {
     // The distributed primitives must NOT break the in-process surface;
     // spot-check the load-bearing exports the in-process callers rely on.

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

@@ -80,6 +80,7 @@ import {
   type PlanVideosJson,
   readFfmpegVersion,
 } from "./shared.js";
+import { DISTRIBUTED_RENDER_CAPABILITIES, readPlanProtocol } from "./planProtocol.js";
 
 /**
  * Non-retryable error codes raised when the planDir is structurally
@@ -229,6 +230,7 @@ export function rebuildExtractedFramesFromPlanDir(
 
 /** Plan-time JSON manifest written by `freezePlan`. */
 interface PlanJson {
+  protocol?: unknown;
   planHash: string;
   producerVersion: string;
   ffmpegVersion: string;
@@ -333,7 +335,15 @@ export async function renderChunk(
   const planJsonPath = join(planDir, "plan.json");
   const encoderJsonPath = join(planDir, "meta", "encoder.json");
   const chunksJsonPath = join(planDir, "meta", "chunks.json");
-  for (const required of [planJsonPath, encoderJsonPath, chunksJsonPath]) {
+  if (!existsSync(planJsonPath)) {
+    throw new RenderChunkValidationError(
+      MISSING_PLAN_ARTIFACT,
+      `[renderChunk] planDir is missing required artifact: ${planJsonPath}`,
+    );
+  }
+  const plan = JSON.parse(readFileSync(planJsonPath, "utf-8")) as PlanJson;
+  readPlanProtocol(plan, DISTRIBUTED_RENDER_CAPABILITIES.roles.chunk);
+  for (const required of [encoderJsonPath, chunksJsonPath]) {
     if (!existsSync(required)) {
       throw new RenderChunkValidationError(
         MISSING_PLAN_ARTIFACT,
@@ -341,7 +351,6 @@ export async function renderChunk(
       );
     }
   }
-  const plan = JSON.parse(readFileSync(planJsonPath, "utf-8")) as PlanJson;
   const encoder = JSON.parse(readFileSync(encoderJsonPath, "utf-8")) as LockedRenderConfig;
   const chunks = JSON.parse(readFileSync(chunksJsonPath, "utf-8")) as ChunkSliceJson[];
 

packages/producer/src/services/render/stages/freezePlan.ts

@@ -14,6 +14,7 @@ import { existsSync, mkdirSync, readFileSync, readdirSync, writeFileSync } from
 import { join, relative, resolve } from "node:path";
 
 import type { Fps } from "@hyperframes/core";
+import { CURRENT_PLAN_PROTOCOL } from "../../distributed/planProtocol.js";
 import {
   canonicalJsonStringify,
   computePlanHash,
@@ -355,6 +356,7 @@ export async function freezePlan(input: FreezePlanInput): Promise<FreezePlanResu
   });
 
   const planJson = {
+    protocol: CURRENT_PLAN_PROTOCOL,
     planHash,
     producerVersion,
     ffmpegVersion: encoder.ffmpegVersion,