feat(producer): add content-addressed plan v2

packages/producer/src/distributed.ts

@@ -54,6 +54,26 @@ export {
   PlanTooLargeError,
 } from "./services/distributed/plan.js";
 
+// ── Plan v2 content-addressed transport ────────────────────────────────────
+export {
+  createPlanV2FromV1,
+  listPlanV2ArtifactsForTarget,
+  materializePlanV2Target,
+  planV2,
+  readPlanV2Manifest,
+  validatePlanV2MaterializedTarget,
+  PLAN_V2_INTEGRITY_UNRECOVERABLE,
+  PLAN_V2_MATERIALIZATION_MARKER,
+  PlanV2IntegrityError,
+  type PlanV2Artifact,
+  type PlanV2Limitations,
+  type PlanV2Manifest,
+  type PlanV2MaterializationResult,
+  type PlanV2MaterializationTarget,
+  type PlanV2Result,
+} from "./services/distributed/planV2.js";
+export { assembleV2, renderChunkV2 } from "./services/distributed/planV2Execution.js";
+
 // ── RenderChunk (Activity B) ────────────────────────────────────────────────
 export {
   applyRuntimeEnvSnapshot,
@@ -91,14 +111,21 @@ export {
   getDistributedRenderCapabilities,
   PLAN_ARTIFACT_LAYOUT,
   PLAN_HASH_SCHEMA,
+  PLAN_PROTOCOL_V2,
   PLAN_PROTOCOL_UNSUPPORTED,
   PLAN_SCHEMA_VERSION,
+  PLAN_V2_ARTIFACT_LAYOUT,
+  PLAN_V2_HASH_SCHEMA,
+  PLAN_V2_SCHEMA_VERSION,
   PlanProtocolUnsupportedError,
   readPlanProtocol,
+  readPlanProtocolV1,
   type DistributedRenderCapabilities,
   type PlanProtocolConsumerCapabilities,
   type PlanProtocolDescriptor,
   type PlanProtocolV1Descriptor,
+  type PlanProtocolV2Descriptor,
+  type SupportedPlanProtocolDescriptor,
 } from "./services/distributed/planProtocol.js";
 
 // ── Format union ────────────────────────────────────────────────────────────

packages/producer/src/index.ts

@@ -133,23 +133,47 @@ export {
 // separate subpath import.
 export {
   assemble,
+  assembleV2,
   CURRENT_PLAN_PROTOCOL,
   DISTRIBUTED_RENDER_CAPABILITIES,
   getDistributedRenderCapabilities,
   PLAN_ARTIFACT_LAYOUT,
   PLAN_HASH_SCHEMA,
+  PLAN_PROTOCOL_V2,
   PLAN_PROTOCOL_UNSUPPORTED,
   PLAN_SCHEMA_VERSION,
+  PLAN_V2_ARTIFACT_LAYOUT,
+  PLAN_V2_HASH_SCHEMA,
+  PLAN_V2_INTEGRITY_UNRECOVERABLE,
+  PLAN_V2_MATERIALIZATION_MARKER,
+  PLAN_V2_SCHEMA_VERSION,
+  createPlanV2FromV1,
+  listPlanV2ArtifactsForTarget,
+  materializePlanV2Target,
   plan,
+  planV2,
+  PlanV2IntegrityError,
   PlanProtocolUnsupportedError,
   readPlanProtocol,
+  readPlanProtocolV1,
+  readPlanV2Manifest,
   renderChunk,
+  renderChunkV2,
+  validatePlanV2MaterializedTarget,
   type AssembleResult,
   type ChunkResult,
   type DistributedRenderCapabilities,
   type DistributedRenderConfig,
   type PlanProtocolConsumerCapabilities,
   type PlanProtocolDescriptor,
   type PlanProtocolV1Descriptor,
+  type PlanProtocolV2Descriptor,
   type PlanResult,
+  type PlanV2Artifact,
+  type PlanV2Limitations,
+  type PlanV2Manifest,
+  type PlanV2MaterializationResult,
+  type PlanV2MaterializationTarget,
+  type PlanV2Result,
+  type SupportedPlanProtocolDescriptor,
 } from "./distributed.js";

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

@@ -40,7 +40,8 @@ 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 { DISTRIBUTED_RENDER_CAPABILITIES, readPlanProtocolV1 } from "./planProtocol.js";
+import { validatePlanV2MaterializedTarget } from "./planV2.js";
 import type { DistributedFormat } from "./shared.js";
 
 /**
@@ -121,7 +122,8 @@ export async function assemble(
     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);
+  readPlanProtocolV1(plan, DISTRIBUTED_RENDER_CAPABILITIES.roles.assembler);
+  validatePlanV2MaterializedTarget(planDir, { role: "assembler" });
   if (!existsSync(chunksJsonPath)) {
     throw new Error(`[assemble] planDir missing meta/chunks.json: ${chunksJsonPath}`);
   }

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

@@ -75,6 +75,7 @@ import { snapshotRuntimeEnv } from "../render/runtimeEnvSnapshot.js";
 import {
   buildSyntheticRenderJob,
   type DistributedFormat,
+  PLAN_AUDIO_RELATIVE_PATH,
   PLAN_VIDEOS_META_RELATIVE_PATH,
   type PlanVideosJson,
   readFfmpegVersion,
@@ -225,7 +226,8 @@ export interface DistributedRenderConfig {
    * 10 GB `/tmp` budget alongside the chunk worker's frame buffer +
    * ffmpeg working set). Adapters that deploy onto storage with
    * tighter ceilings can pass a smaller cap; tests pass a tiny cap to
-   * exercise the throw path.
+   * exercise the throw path. This applies to the monolithic v1 transport;
+   * `planV2()` emits content-addressed role dependencies and bypasses it.
    */
   planDirSizeLimitBytes?: number;
 
@@ -340,9 +342,8 @@ export const MIN_CHUNK_SIZE = 10;
 /**
  * Default hard ceiling on `<planDir>/` size in bytes. 2 GB fits inside
  * AWS Lambda's 10 GB `/tmp` alongside the chunk worker's captured frames
- * and ffmpeg's temporary files. Compositions that exceed this have to
- * fall back to the in-process renderer until per-chunk video-frame
- * slicing lands.
+ * and ffmpeg's temporary files. Compositions that exceed this can opt into
+ * `planV2()` or fall back to the in-process renderer.
  */
 export const PLAN_DIR_SIZE_LIMIT_BYTES = 2 * 1024 * 1024 * 1024;
 
@@ -364,8 +365,9 @@ export class PlanTooLargeError extends Error {
       `[plan] planDir size ${formatBytes(sizeBytes)} exceeds the configured ceiling ` +
         `${formatBytes(limitBytes)} (PLAN_TOO_LARGE). The default 2 GB cap fits inside AWS ` +
         `Lambda's 10 GB /tmp budget alongside the chunk worker's frame buffer and ffmpeg's ` +
-        `working set. To unblock: shorten the composition, lower the framerate, or use the ` +
-        `in-process renderer (\`executeRenderJob\`) — it has no planDir size cap.`,
+        `working set. To unblock: use the content-addressed \`planV2()\` transport, shorten ` +
+        `the composition, lower the framerate, or use the in-process renderer ` +
+        `(\`executeRenderJob\`) — it has no planDir size cap.`,
     );
     this.name = "PlanTooLargeError";
     this.sizeBytes = sizeBytes;
@@ -1006,7 +1008,7 @@ export async function plan(
     "utf-8",
   );
 
-  const planAudioPath = join(planDir, "audio.aac");
+  const planAudioPath = join(planDir, PLAN_AUDIO_RELATIVE_PATH);
   if (audioResult.hasAudio && existsSync(audioResult.audioOutputPath)) {
     renameSync(audioResult.audioOutputPath, planAudioPath);
   }

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

@@ -1,3 +1,7 @@
+// These protocol rejection cases intentionally repeat the arrange/assert shape
+// so each malformed wire descriptor remains independently readable.
+// fallow-ignore-file code-duplication
+
 import { afterEach, describe, expect, it } from "bun:test";
 import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs";
 import { tmpdir } from "node:os";
@@ -9,6 +13,7 @@ import {
   getDistributedRenderCapabilities,
   PLAN_ARTIFACT_LAYOUT,
   PLAN_HASH_SCHEMA,
+  PLAN_PROTOCOL_V2,
   PLAN_PROTOCOL_UNSUPPORTED,
   PLAN_SCHEMA_VERSION,
   PlanProtocolUnsupportedError,
@@ -126,6 +131,10 @@ describe("readPlanProtocol()", () => {
     ).toBe(CURRENT_PLAN_PROTOCOL);
   });
 
+  it("accepts the explicit v2 descriptor", () => {
+    expect(readPlanProtocol({ protocol: PLAN_PROTOCOL_V2 })).toBe(PLAN_PROTOCOL_V2);
+  });
+
   it("rejects malformed and partial descriptors", () => {
     for (const protocol of [
       null,
@@ -166,19 +175,19 @@ describe("readPlanProtocol()", () => {
 });
 
 describe("getDistributedRenderCapabilities()", () => {
-  it("reports explicit v1 support for every distributed role", () => {
+  it("reports explicit v1 and v2 support for every distributed role", () => {
     expect(getDistributedRenderCapabilities()).toBe(DISTRIBUTED_RENDER_CAPABILITIES);
     expect(DISTRIBUTED_RENDER_CAPABILITIES).toEqual({
       roles: {
         planner: {
-          produces: [CURRENT_PLAN_PROTOCOL],
+          produces: [CURRENT_PLAN_PROTOCOL, PLAN_PROTOCOL_V2],
         },
         chunk: {
-          accepts: [CURRENT_PLAN_PROTOCOL],
+          accepts: [CURRENT_PLAN_PROTOCOL, PLAN_PROTOCOL_V2],
           acceptsLegacyV1WithoutDescriptor: true,
         },
         assembler: {
-          accepts: [CURRENT_PLAN_PROTOCOL],
+          accepts: [CURRENT_PLAN_PROTOCOL, PLAN_PROTOCOL_V2],
           acceptsLegacyV1WithoutDescriptor: true,
         },
       },
@@ -273,6 +282,21 @@ describe("distributed plan protocol readers", () => {
     expect((caught as PlanProtocolUnsupportedError).code).toBe(PLAN_PROTOCOL_UNSUPPORTED);
   });
 
+  it("legacy activities reject a recognized v2 root before v1 layout access", async () => {
+    const planDir = createReaderPlan({
+      includeProtocol: true,
+      protocol: PLAN_PROTOCOL_V2,
+      omitDownstreamArtifacts: true,
+    });
+
+    await expect(renderChunk(planDir, 0, join(planDir, "unused-output"))).rejects.toThrow(
+      "must be materialized before v1 layout access",
+    );
+    await expect(assemble(planDir, [], null, join(planDir, "unused-output"))).rejects.toThrow(
+      "must be materialized before v1 layout access",
+    );
+  });
+
   it("assemble rejects a partial protocol before parsing chunks", async () => {
     const planDir = createReaderPlan({
       includeProtocol: true,

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

@@ -11,6 +11,9 @@
 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_V2_SCHEMA_VERSION = 2 as const;
+export const PLAN_V2_ARTIFACT_LAYOUT = "content-addressed-plan-v2" as const;
+export const PLAN_V2_HASH_SCHEMA = "hyperframes-plan-manifest-hash-v2" as const;
 export const PLAN_PROTOCOL_UNSUPPORTED = "PLAN_PROTOCOL_UNSUPPORTED" as const;
 
 export interface PlanProtocolDescriptor {
@@ -25,13 +28,28 @@ export interface PlanProtocolV1Descriptor extends PlanProtocolDescriptor {
   readonly hashSchema: typeof PLAN_HASH_SCHEMA;
 }
 
+export interface PlanProtocolV2Descriptor extends PlanProtocolDescriptor {
+  readonly schemaVersion: typeof PLAN_V2_SCHEMA_VERSION;
+  readonly artifactLayout: typeof PLAN_V2_ARTIFACT_LAYOUT;
+  readonly hashSchema: typeof PLAN_V2_HASH_SCHEMA;
+}
+
+export type SupportedPlanProtocolDescriptor = PlanProtocolV1Descriptor | PlanProtocolV2Descriptor;
+
 /** 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,
 });
 
+/** Explicit opt-in descriptor for the content-addressed v2 transport layout. */
+export const PLAN_PROTOCOL_V2: Readonly<PlanProtocolV2Descriptor> = Object.freeze({
+  schemaVersion: PLAN_V2_SCHEMA_VERSION,
+  artifactLayout: PLAN_V2_ARTIFACT_LAYOUT,
+  hashSchema: PLAN_V2_HASH_SCHEMA,
+});
+
 export interface PlanProtocolConsumerCapabilities {
   readonly accepts: readonly Readonly<PlanProtocolDescriptor>[];
   readonly acceptsLegacyV1WithoutDescriptor: boolean;
@@ -52,14 +70,14 @@ export const DISTRIBUTED_RENDER_CAPABILITIES: Readonly<DistributedRenderCapabili
   Object.freeze({
     roles: Object.freeze({
       planner: Object.freeze({
-        produces: Object.freeze([CURRENT_PLAN_PROTOCOL]),
+        produces: Object.freeze([CURRENT_PLAN_PROTOCOL, PLAN_PROTOCOL_V2]),
       }),
       chunk: Object.freeze({
-        accepts: Object.freeze([CURRENT_PLAN_PROTOCOL]),
+        accepts: Object.freeze([CURRENT_PLAN_PROTOCOL, PLAN_PROTOCOL_V2]),
         acceptsLegacyV1WithoutDescriptor: true,
       }),
       assembler: Object.freeze({
-        accepts: Object.freeze([CURRENT_PLAN_PROTOCOL]),
+        accepts: Object.freeze([CURRENT_PLAN_PROTOCOL, PLAN_PROTOCOL_V2]),
         acceptsLegacyV1WithoutDescriptor: true,
       }),
     }),
@@ -87,7 +105,7 @@ function isRecord(value: unknown): value is Record<string, unknown> {
 
 function protocolMatches(
   descriptor: Record<string, unknown>,
-  expected: PlanProtocolDescriptor,
+  expected: SupportedPlanProtocolDescriptor,
 ): boolean {
   return (
     descriptor.schemaVersion === expected.schemaVersion &&
@@ -117,7 +135,7 @@ export function readPlanProtocol(
   planJson: unknown,
   capabilities: Readonly<PlanProtocolConsumerCapabilities> = DISTRIBUTED_RENDER_CAPABILITIES.roles
     .chunk,
-): Readonly<PlanProtocolV1Descriptor> {
+): Readonly<SupportedPlanProtocolDescriptor> {
   if (!isRecord(planJson)) {
     throw new PlanProtocolUnsupportedError("plan.json must contain a JSON object");
   }
@@ -145,11 +163,32 @@ export function readPlanProtocol(
     }
   }
 
-  if (
-    !protocolMatches(descriptor, CURRENT_PLAN_PROTOCOL) ||
-    !capabilitiesAccept(capabilities, CURRENT_PLAN_PROTOCOL)
-  ) {
+  const protocol = protocolMatches(descriptor, CURRENT_PLAN_PROTOCOL)
+    ? CURRENT_PLAN_PROTOCOL
+    : protocolMatches(descriptor, PLAN_PROTOCOL_V2)
+      ? PLAN_PROTOCOL_V2
+      : null;
+  if (protocol === null || !capabilitiesAccept(capabilities, protocol)) {
     throw new PlanProtocolUnsupportedError("unsupported plan.json protocol descriptor");
   }
+  return protocol;
+}
+
+/**
+ * Validate that a directory is directly consumable by the legacy execution
+ * functions. A v2 transport must be materialized first; rejecting it here
+ * prevents readers from probing paths that have different meanings in v2.
+ */
+export function readPlanProtocolV1(
+  planJson: unknown,
+  capabilities: Readonly<PlanProtocolConsumerCapabilities> = DISTRIBUTED_RENDER_CAPABILITIES.roles
+    .chunk,
+): Readonly<PlanProtocolV1Descriptor> {
+  const protocol = readPlanProtocol(planJson, capabilities);
+  if (protocol !== CURRENT_PLAN_PROTOCOL) {
+    throw new PlanProtocolUnsupportedError(
+      "content-addressed v2 plan must be materialized before v1 layout access",
+    );
+  }
   return CURRENT_PLAN_PROTOCOL;
 }