From 9f3e9cee6980653fb34a850f077cb0e306725068 Mon Sep 17 00:00:00 2001 From: James Date: Thu, 30 Jul 2026 23:09:05 +0000 Subject: [PATCH] refactor(producer): share plan execution builder --- docs/packages/aws-lambda.mdx | 6 + docs/packages/gcp-cloud-run.mdx | 6 + examples/aws-lambda/README.md | 5 +- packages/aws-lambda/README.md | 5 +- packages/aws-lambda/src/events.ts | 24 ++- packages/aws-lambda/src/handler.test.ts | 4 +- .../aws-lambda/src/sdk/renderToLambda.test.ts | 19 +- packages/aws-lambda/src/sdk/renderToLambda.ts | 2 +- packages/gcp-cloud-run/README.md | 21 +- packages/gcp-cloud-run/src/events.ts | 24 ++- .../src/sdk/renderToCloudRun.test.ts | 35 +++- .../gcp-cloud-run/src/sdk/renderToCloudRun.ts | 2 +- packages/gcp-cloud-run/src/server.test.ts | 4 +- packages/producer/README.md | 31 +-- packages/producer/src/distributed.ts | 26 ++- packages/producer/src/index.ts | 4 + .../src/services/distributed/plan.test.ts | 21 +- .../producer/src/services/distributed/plan.ts | 94 +++++++-- .../services/distributed/planProtocol.test.ts | 8 + .../src/services/distributed/planProtocol.ts | 30 ++- .../services/distributed/planSizeCap.test.ts | 5 +- .../src/services/distributed/planV2.test.ts | 77 ++++++- .../src/services/distributed/planV2.ts | 196 +++++++++++------- .../distributed/publicExports.test.ts | 12 ++ .../src/services/render/stages/freezePlan.ts | 4 +- 25 files changed, 494 insertions(+), 171 deletions(-) diff --git a/docs/packages/aws-lambda.mdx b/docs/packages/aws-lambda.mdx index 6d44836d64..b2e4e18aaf 100644 --- a/docs/packages/aws-lambda.mdx +++ b/docs/packages/aws-lambda.mdx @@ -70,6 +70,7 @@ const site = await deploySite({ const handle = await renderToLambda({ siteHandle: site, + planProtocol: "v2", bucketName: site.bucketName, stateMachineArn: "arn:aws:states:us-east-1:123456789012:stateMachine:hyperframes-render", config: { @@ -87,6 +88,11 @@ const progress = await getRenderProgress({ executionArn: handle.executionArn }); console.log(progress.status, progress.overallProgress, progress.costs.displayCost); ``` +Plan v2 is recommended for new integrations because workers fetch +manifest-selected content-addressed artifacts. For backwards compatibility, +omitting `planProtocol` still selects v1; existing callers do not change +behavior until they opt in. + `renderToLambda()` validates the distributed render config before starting the Step Functions execution, so invalid dimensions, formats, chunk sizes, or payload sizes fail synchronously. ## Using the CDK Construct diff --git a/docs/packages/gcp-cloud-run.mdx b/docs/packages/gcp-cloud-run.mdx index bce0538480..4a71d688a3 100644 --- a/docs/packages/gcp-cloud-run.mdx +++ b/docs/packages/gcp-cloud-run.mdx @@ -79,6 +79,7 @@ import { getRenderProgress, renderToCloudRun } from "@hyperframes/gcp-cloud-run/ const handle = await renderToCloudRun({ projectDir: "./my-composition", + planProtocol: "v2", config: { fps: 30, width: 1920, height: 1080, format: "mp4" }, bucketName: "hyperframes-render-my-project", projectId: "my-project", @@ -96,6 +97,11 @@ while (progress.status === "running") { console.log(progress.status, progress.outputFile, progress.costs.displayCost); ``` +Plan v2 is recommended for new integrations because workers fetch +manifest-selected content-addressed artifacts. For backwards compatibility, +omitting `planProtocol` still selects v1; existing callers do not change +behavior until they opt in. + Pass `projectDir` for one-shot uploads, or call `deploySite()` separately and reuse the returned site handle across many renders. ## Related Guides diff --git a/examples/aws-lambda/README.md b/examples/aws-lambda/README.md index 7300ac141b..3a04edb25e 100644 --- a/examples/aws-lambda/README.md +++ b/examples/aws-lambda/README.md @@ -75,7 +75,7 @@ aws stepfunctions start-execution \ "ProjectS3Uri": "s3://${RENDER_BUCKET}/projects/my-project.tar.gz", "PlanOutputS3Prefix": "s3://${RENDER_BUCKET}/renders/$(date +%s)/", "OutputS3Uri": "s3://${RENDER_BUCKET}/output.mp4", - "PlanProtocol": "v1", + "PlanProtocol": "v2", "Config": { "fps": 30, "width": 1920, @@ -92,7 +92,8 @@ EOF The Step Functions execution kicks off Plan, fans out RenderChunk via the Map state, and finally Assemble. Final mp4 lands at `OutputS3Uri`. -`PlanProtocol` may be `"v1"` or `"v2"`; absent defaults to v1. V2 uses +Plan v2 is recommended for new integrations. `PlanProtocol` may be `"v1"` or +`"v2"`; absent still defaults to v1 for backwards compatibility. V2 uses separate manifest and content-addressed artifact locators throughout the workflow and never places a v2 object in `PlanS3Uri`. diff --git a/packages/aws-lambda/README.md b/packages/aws-lambda/README.md index 8f462ad190..14423f08e4 100644 --- a/packages/aws-lambda/README.md +++ b/packages/aws-lambda/README.md @@ -47,8 +47,9 @@ inside Step Functions' history budget (under 200 bytes per chunk). ### Plan transport selection -`renderToLambda` defaults to the existing monolithic v1 plan transport. -Plan v2 is an explicit whole-render opt-in: +Plan v2 is recommended for new integrations. `renderToLambda` still defaults +an omitted `planProtocol` to the existing monolithic v1 transport for +backwards compatibility, so select v2 explicitly: ```ts await renderToLambda({ diff --git a/packages/aws-lambda/src/events.ts b/packages/aws-lambda/src/events.ts index 0756575f9f..4e23b03dab 100644 --- a/packages/aws-lambda/src/events.ts +++ b/packages/aws-lambda/src/events.ts @@ -54,7 +54,11 @@ interface PlanEventBase { Config: SerializableDistributedRenderConfig; } -/** Legacy/default plan transport. Absence is deliberately interpreted as v1. */ +/** + * Legacy/default plan transport. Absence is deliberately interpreted as v1. + * + * @deprecated Use {@link PlanV2Event} for new integrations. + */ export interface PlanV1Event extends PlanEventBase { PlanProtocol?: "v1"; } @@ -85,7 +89,11 @@ interface RenderChunkEventBase { Format: DistributedFormat; } -/** Legacy/default chunk event. */ +/** + * Legacy/default chunk event. + * + * @deprecated Use {@link RenderChunkV2Event} for new integrations. + */ export interface RenderChunkV1Event extends RenderChunkEventBase { PlanProtocol?: "v1"; /** S3 URI of the v1 plan tar produced by a PlanEvent invocation. */ @@ -127,7 +135,11 @@ interface AssembleEventBase { Cfr?: boolean; } -/** Legacy/default assemble event. */ +/** + * Legacy/default assemble event. + * + * @deprecated Use {@link AssembleV2Event} for new integrations. + */ export interface AssembleV1Event extends AssembleEventBase { PlanProtocol?: "v1"; /** S3 URI of the v1 plan tar produced by a PlanEvent invocation. */ @@ -163,7 +175,11 @@ interface PlanLambdaResultBase { DurationMs: number; } -/** Existing v1 result. Kept unchanged for wire compatibility. */ +/** + * Existing v1 result. Kept unchanged for wire compatibility. + * + * @deprecated New integrations should consume {@link PlanV2LambdaResult}. + */ export interface PlanV1LambdaResult extends PlanLambdaResultBase { PlanS3Uri: string; } diff --git a/packages/aws-lambda/src/handler.test.ts b/packages/aws-lambda/src/handler.test.ts index 048c43c7c6..e7323412ef 100644 --- a/packages/aws-lambda/src/handler.test.ts +++ b/packages/aws-lambda/src/handler.test.ts @@ -28,7 +28,7 @@ import { type PlanResult, type PlanV2ArtifactPublisher, type PlanV2Manifest, - publishPlanV2FromV1, + publishPlanV2FromExecutionPlan, } from "@hyperframes/producer/distributed"; import { recomputePlanHashFromPlanDir } from "../../producer/src/services/render/stages/freezePlan.js"; import type { AssembleEvent, LambdaEvent, PlanEvent, RenderChunkEvent } from "./events.js"; @@ -555,7 +555,7 @@ describe("handler dispatch", () => { ): Promise => { const v1Dir = join(tmpRoot, `v1-${Date.now()}`); makeMinimalV1PlanDir(v1Dir, true); - const manifest = await publishPlanV2FromV1(v1Dir, publisher); + const manifest = await publishPlanV2FromExecutionPlan(v1Dir, publisher); expect(options.stagingParentDir).toBe(dirname(projectDir)); expect(existsSync(join(dirname(projectDir), "plan-v2"))).toBe(false); return manifest; diff --git a/packages/aws-lambda/src/sdk/renderToLambda.test.ts b/packages/aws-lambda/src/sdk/renderToLambda.test.ts index 9410936b96..61402c2add 100644 --- a/packages/aws-lambda/src/sdk/renderToLambda.test.ts +++ b/packages/aws-lambda/src/sdk/renderToLambda.test.ts @@ -79,6 +79,15 @@ describe("renderToLambda", () => { expect(handle.projectS3Uri).toMatch( /^s3:\/\/test-bucket\/sites\/[0-9a-f]{16}\/project\.tar\.gz$/, ); + expect(Object.keys(handle)).toEqual([ + "renderId", + "executionArn", + "bucketName", + "stateMachineArn", + "outputS3Uri", + "projectS3Uri", + "startedAt", + ]); expect(sfn.starts).toHaveLength(1); const start = sfn.starts[0]!; @@ -96,7 +105,7 @@ describe("renderToLambda", () => { it("opts the complete execution into plan protocol v2 explicitly", async () => { const sfn = new FakeSFN(); const s3 = new FakeS3(); - await renderToLambda({ + const handle = await renderToLambda({ projectDir, bucketName: "test-bucket", stateMachineArn: "arn:aws:states:us-east-1:1234:stateMachine:hf", @@ -107,7 +116,13 @@ describe("renderToLambda", () => { s3: asS3Client(s3), }); - expect(sfn.starts[0]?.input).toMatchObject({ PlanProtocol: "v2" }); + expect(sfn.starts[0]?.input).toEqual({ + ProjectS3Uri: handle.projectS3Uri, + PlanOutputS3Prefix: "s3://test-bucket/renders/smoke-v2/", + OutputS3Uri: "s3://test-bucket/renders/smoke-v2/output.mp4", + Config: baseConfig, + PlanProtocol: "v2", + }); }); it("derives the file extension from config.format", async () => { diff --git a/packages/aws-lambda/src/sdk/renderToLambda.ts b/packages/aws-lambda/src/sdk/renderToLambda.ts index 5e83482b0b..8df5e20c12 100644 --- a/packages/aws-lambda/src/sdk/renderToLambda.ts +++ b/packages/aws-lambda/src/sdk/renderToLambda.ts @@ -39,7 +39,7 @@ export interface RenderToLambdaOptions { config: SerializableDistributedRenderConfig; /** * Distributed plan transport. Defaults to `"v1"` for backwards - * compatibility; v2 is always an explicit whole-render opt-in. + * compatibility. New integrations should explicitly select `"v2"`. */ planProtocol?: LambdaPlanProtocol; /** S3 bucket from the SAM stack output (`RenderBucketName`). */ diff --git a/packages/gcp-cloud-run/README.md b/packages/gcp-cloud-run/README.md index c2aa7fb8c1..aa9c8c3ca4 100644 --- a/packages/gcp-cloud-run/README.md +++ b/packages/gcp-cloud-run/README.md @@ -30,8 +30,8 @@ GCS bucket ←→ Cloud Run service (plan / renderChunk / assemble) Cloud Workflows (Plan → parallel RenderChunk → Assemble) ``` -- **Plan** downloads the project tarball, runs `plan()`, uploads the planDir - tarball (+ audio) to GCS, and returns the chunk count. +- **Plan** downloads the project tarball and publishes either a legacy v1 + planDir tarball or a v2 manifest plus content-addressed artifacts. - **RenderChunk** runs in a parallel `for` loop in the workflow, fanned out up to the plan's chunk count. Each invocation renders one chunk and uploads it. @@ -43,6 +43,23 @@ The workflow accumulates each step's small result body and returns `{ Plan, Chunks, Assemble }` so `getRenderProgress` can read frame totals and per-step durations on success. +### Plan transport selection + +Plan v2 is recommended for new integrations. `renderToCloudRun` still +interprets an omitted `planProtocol` as `"v1"` for backwards compatibility, +so new callers should select v2 explicitly: + +```ts +await renderToCloudRun({ + // ...project, bucket, workflow, service, and config... + planProtocol: "v2", +}); +``` + +V2 uses separate manifest and content-addressed artifact locators throughout +the workflow. Unknown protocols and integrity failures fail closed; a render +never mixes v1 and v2 artifacts. + ## Chrome runtime Unlike the Lambda adapter — which fights a 250 MB ZIP ceiling and diff --git a/packages/gcp-cloud-run/src/events.ts b/packages/gcp-cloud-run/src/events.ts index a65e0df76c..8cc7ef4694 100644 --- a/packages/gcp-cloud-run/src/events.ts +++ b/packages/gcp-cloud-run/src/events.ts @@ -60,7 +60,11 @@ interface PlanEventBase { Config: SerializableDistributedRenderConfig; } -/** Legacy/default plan transport. Absence is deliberately interpreted as v1. */ +/** + * Legacy/default plan transport. Absence is deliberately interpreted as v1. + * + * @deprecated Use {@link PlanV2Event} for new integrations. + */ export interface PlanV1Event extends PlanEventBase { PlanProtocol?: "v1"; } @@ -91,7 +95,11 @@ interface RenderChunkEventBase { Format: DistributedFormat; } -/** Legacy/default chunk event. */ +/** + * Legacy/default chunk event. + * + * @deprecated Use {@link RenderChunkV2Event} for new integrations. + */ export interface RenderChunkV1Event extends RenderChunkEventBase { PlanProtocol?: "v1"; /** GCS URI of the v1 plan tar produced by a PlanEvent invocation. */ @@ -134,7 +142,11 @@ interface AssembleEventBase { Cfr?: boolean; } -/** Legacy/default assemble event. */ +/** + * Legacy/default assemble event. + * + * @deprecated Use {@link AssembleV2Event} for new integrations. + */ export interface AssembleV1Event extends AssembleEventBase { PlanProtocol?: "v1"; /** GCS URI of the v1 plan tar produced by a PlanEvent invocation. */ @@ -177,7 +189,11 @@ interface PlanResultBodyBase { DurationMs: number; } -/** Existing v1 result. Kept unchanged for wire compatibility. */ +/** + * Existing v1 result. Kept unchanged for wire compatibility. + * + * @deprecated New integrations should consume {@link PlanV2ResultBody}. + */ export interface PlanV1ResultBody extends PlanResultBodyBase { PlanGcsUri: string; PlanProtocol?: never; diff --git a/packages/gcp-cloud-run/src/sdk/renderToCloudRun.test.ts b/packages/gcp-cloud-run/src/sdk/renderToCloudRun.test.ts index e7d2e39818..6cf7168927 100644 --- a/packages/gcp-cloud-run/src/sdk/renderToCloudRun.test.ts +++ b/packages/gcp-cloud-run/src/sdk/renderToCloudRun.test.ts @@ -66,19 +66,30 @@ describe("renderToCloudRun", () => { ); expect(handle.outputGcsUri).toBe("gs://b/renders/hf-render-fixed/output.mp4"); expect(handle.projectGcsUri).toBe("gs://b/sites/abc/project.tar.gz"); + expect(Object.keys(handle)).toEqual([ + "renderId", + "executionName", + "bucketName", + "workflowId", + "outputGcsUri", + "projectGcsUri", + "startedAt", + ]); }); it("builds the workflow argument the YAML expects", async () => { const fake = new FakeExecutions(); await renderToCloudRun(opts(fake)); const arg = JSON.parse(fake.lastArgument ?? "{}"); - expect(arg.RenderId).toBe("hf-render-fixed"); - expect(arg.ProjectGcsUri).toBe("gs://b/sites/abc/project.tar.gz"); - expect(arg.PlanOutputGcsPrefix).toBe("gs://b/renders/hf-render-fixed/"); - expect(arg.OutputGcsUri).toBe("gs://b/renders/hf-render-fixed/output.mp4"); - expect(arg.ServiceUrl).toBe("https://render-abc.run.app"); - expect(arg.Config.format).toBe("mp4"); - expect(arg.PlanProtocol).toBe("v1"); + expect(arg).toEqual({ + RenderId: "hf-render-fixed", + ProjectGcsUri: "gs://b/sites/abc/project.tar.gz", + PlanOutputGcsPrefix: "gs://b/renders/hf-render-fixed/", + OutputGcsUri: "gs://b/renders/hf-render-fixed/output.mp4", + ServiceUrl: "https://render-abc.run.app", + Config: config, + PlanProtocol: "v1", + }); expect(fake.lastParent).toBe( "projects/proj/locations/us-central1/workflows/hyperframes-render", ); @@ -88,7 +99,15 @@ describe("renderToCloudRun", () => { const fake = new FakeExecutions(); await renderToCloudRun({ ...opts(fake), planProtocol: "v2" }); const arg = JSON.parse(fake.lastArgument ?? "{}"); - expect(arg.PlanProtocol).toBe("v2"); + expect(arg).toEqual({ + RenderId: "hf-render-fixed", + ProjectGcsUri: "gs://b/sites/abc/project.tar.gz", + PlanOutputGcsPrefix: "gs://b/renders/hf-render-fixed/", + OutputGcsUri: "gs://b/renders/hf-render-fixed/output.mp4", + ServiceUrl: "https://render-abc.run.app", + Config: config, + PlanProtocol: "v2", + }); }); it("derives the output extension from the format", async () => { diff --git a/packages/gcp-cloud-run/src/sdk/renderToCloudRun.ts b/packages/gcp-cloud-run/src/sdk/renderToCloudRun.ts index 1fc8937802..ae07d2e91e 100644 --- a/packages/gcp-cloud-run/src/sdk/renderToCloudRun.ts +++ b/packages/gcp-cloud-run/src/sdk/renderToCloudRun.ts @@ -54,7 +54,7 @@ export interface RenderToCloudRunOptions { config: SerializableDistributedRenderConfig; /** * Distributed plan transport. Defaults to `"v1"` for backwards - * compatibility; v2 is always an explicit whole-render opt-in. + * compatibility. New integrations should explicitly select `"v2"`. */ planProtocol?: CloudRunPlanProtocol; /** GCS bucket from the Terraform output (`render_bucket_name`). */ diff --git a/packages/gcp-cloud-run/src/server.test.ts b/packages/gcp-cloud-run/src/server.test.ts index e8e3e05cc3..eced0b048b 100644 --- a/packages/gcp-cloud-run/src/server.test.ts +++ b/packages/gcp-cloud-run/src/server.test.ts @@ -30,7 +30,7 @@ import { type PlanResult, type PlanV2ArtifactPublisher, type PlanV2Manifest, - publishPlanV2FromV1, + publishPlanV2FromExecutionPlan, } from "@hyperframes/producer/distributed"; import { recomputePlanHashFromPlanDir } from "../../producer/src/services/render/stages/freezePlan.js"; import { asStorage, FakeGcs } from "./__fixtures__/fakeGcs.js"; @@ -263,7 +263,7 @@ describe("dispatch", () => { ): Promise => { const v1Dir = join(root, "v1"); makeMinimalV1PlanDir(v1Dir, true); - const manifest = await publishPlanV2FromV1(v1Dir, publisher); + const manifest = await publishPlanV2FromExecutionPlan(v1Dir, publisher); expect(options.stagingParentDir).toBe(dirname(projectDir)); expect(existsSync(join(dirname(projectDir), "plan-v2"))).toBe(false); return manifest; diff --git a/packages/producer/README.md b/packages/producer/README.md index 9af9a80a43..7349071df1 100644 --- a/packages/producer/README.md +++ b/packages/producer/README.md @@ -111,31 +111,32 @@ Don't paint a fullscreen background in your HTML. The default body background is For renders too large for a single machine, the producer ships a public set of distributed-render primitives. They are pure functions over local file paths — networking and orchestration live in adapter packages (Temporal, AWS Lambda + Step Functions, Cloud Run Jobs, K8s Jobs). +Plan v2 is recommended for new integrations. It publishes an immutable +manifest plus content-addressed artifacts and materializes only each worker's +declared dependencies: + ```typescript -import { plan, renderChunk, assemble } from "@hyperframes/producer/distributed"; +import { planV2, renderChunkV2, assembleV2 } from "@hyperframes/producer/distributed"; -// Controller-side: produce a self-contained planDir + content-addressed planHash. -const planResult = await plan( +// Controller-side: produce a v2 manifest + local content-addressed store. +const planResult = await planV2( projectDir, { fps: 30, width: 1920, height: 1080, format: "mp4" }, - "/tmp/plan", + "/tmp/plan-v2", ); -// Worker-side: render one chunk. Byte-identical retries on the same -// `(planDir, chunkIndex)` — Temporal / Step Functions retry policies are safe -// to point at this. -const chunk = await renderChunk("/tmp/plan", 0, "/tmp/chunks/0.mp4"); +const chunk = await renderChunkV2("/tmp/plan-v2", 0, "/tmp/chunks/0.mp4"); // Controller-side: stitch chunks into the final deliverable. -await assemble( - "/tmp/plan", - ["/tmp/chunks/0.mp4", "/tmp/chunks/1.mp4"], - "/tmp/plan/audio.aac", - "/tmp/output.mp4", -); +await assembleV2("/tmp/plan-v2", ["/tmp/chunks/0.mp4", "/tmp/chunks/1.mp4"], "/tmp/output.mp4"); ``` -The three activity functions plus their result types are also re-exported from `@hyperframes/producer` so callers that pin the main package don't need a separate subpath import. Supported formats: `mp4` SDR, `mov` ProRes 4444, and `png-sequence`. webm and HDR mp4 trip a typed `FormatNotSupportedInDistributedError` — use the in-process renderer (`executeRenderJob`) for those. +Cloud adapters should use `planV2WithPublisher()` so artifacts publish +directly to object storage. The legacy `plan()` / `renderChunk()` / +`assemble()` v1 layout remains supported, and cloud SDKs still interpret an +omitted protocol as v1 for backwards compatibility. + +The activity functions plus their result types are also re-exported from `@hyperframes/producer` so callers that pin the main package don't need a separate subpath import. Supported formats: `mp4` SDR, `mov` ProRes 4444, and `png-sequence`. webm and HDR mp4 trip a typed `FormatNotSupportedInDistributedError` — use the in-process renderer (`executeRenderJob`) for those. ## How it works diff --git a/packages/producer/src/distributed.ts b/packages/producer/src/distributed.ts index 3000892fa2..14a15fa61d 100644 --- a/packages/producer/src/distributed.ts +++ b/packages/producer/src/distributed.ts @@ -1,29 +1,29 @@ /** * `@hyperframes/producer/distributed` — the distributed render primitives. * - * The three activities (`plan` → `renderChunk` × N → `assemble`) are pure - * functions over local file paths; networking + orchestration live in - * adapters. + * The distributed activities are pure functions over local file paths; + * networking + orchestration live in adapters. New integrations should use + * Plan v2; the v1 functions remain available for compatibility. * * Adopters (AWS Lambda, Cloud Run Jobs, Temporal, K8s Jobs, plain SSH): * * ```ts * import { - * plan, - * renderChunk, - * assemble, + * planV2, + * renderChunkV2, + * assembleV2, * } from "@hyperframes/producer/distributed"; * - * // Controller-side: produce a self-contained planDir + content-addressed planHash. - * const planResult = await plan(projectDir, config, planDir); + * // Controller-side: publish a content-addressed Plan v2 manifest + CAS. + * const planResult = await planV2(projectDir, config, planV2Dir); * * // Worker-side: render one chunk. Byte-identical retries on the same - * // (planDir, chunkIndex) — Temporal / Step Functions retry policies are + * // (planV2Dir, chunkIndex) — Temporal / Step Functions retry policies are * // safe to point at this. - * const chunk = await renderChunk(planDir, chunkIndex, outputChunkPath); + * const chunk = await renderChunkV2(planV2Dir, chunkIndex, outputChunkPath); * * // Controller-side: stitch chunks into the final deliverable. - * await assemble(planDir, chunkPaths, audioPath, outputPath); + * await assembleV2(planV2Dir, chunkPaths, outputPath); * ``` * * No networking, no AWS SDK, no Temporal SDK — those live in adapter @@ -56,11 +56,14 @@ export { // ── Plan v2 content-addressed transport ──────────────────────────────────── export { + createPlanV2FromExecutionPlan, createPlanV2FromV1, + getPlanV2ExecutionPlanHash, listPlanV2ArtifactsForTarget, materializePlanV2Target, planV2, planV2WithPublisher, + publishPlanV2FromExecutionPlan, publishPlanV2FromV1, readPlanV2Manifest, validatePlanV2MaterializedTarget, @@ -123,6 +126,7 @@ export { getDistributedRenderCapabilities, PLAN_ARTIFACT_LAYOUT, PLAN_HASH_SCHEMA, + PLAN_PROTOCOL_V1, PLAN_PROTOCOL_V2, PLAN_PROTOCOL_UNSUPPORTED, PLAN_SCHEMA_VERSION, diff --git a/packages/producer/src/index.ts b/packages/producer/src/index.ts index 93e1329794..4a91156f14 100644 --- a/packages/producer/src/index.ts +++ b/packages/producer/src/index.ts @@ -145,6 +145,7 @@ export { getDistributedRenderCapabilities, PLAN_ARTIFACT_LAYOUT, PLAN_HASH_SCHEMA, + PLAN_PROTOCOL_V1, PLAN_PROTOCOL_V2, PLAN_PROTOCOL_UNSUPPORTED, PLAN_SCHEMA_VERSION, @@ -153,7 +154,9 @@ export { PLAN_V2_INTEGRITY_UNRECOVERABLE, PLAN_V2_MATERIALIZATION_MARKER, PLAN_V2_SCHEMA_VERSION, + createPlanV2FromExecutionPlan, createPlanV2FromV1, + getPlanV2ExecutionPlanHash, listPlanV2ArtifactsForTarget, materializePlanV2Target, plan, @@ -164,6 +167,7 @@ export { readPlanProtocol, readPlanProtocolV1, readPlanV2Manifest, + publishPlanV2FromExecutionPlan, publishPlanV2FromV1, renderChunk, renderChunkV2, diff --git a/packages/producer/src/services/distributed/plan.test.ts b/packages/producer/src/services/distributed/plan.test.ts index 027bcc120b..9680c53b60 100644 --- a/packages/producer/src/services/distributed/plan.test.ts +++ b/packages/producer/src/services/distributed/plan.test.ts @@ -25,6 +25,7 @@ import { RenderQualityError } from "../renderOrchestrator.js"; import { CURRENT_PLAN_PROTOCOL } from "./planProtocol.js"; import { applyDistributedAudioWarningPolicy, + buildLocalExecutionPlan, buildChunkSlices, DEFAULT_CHUNK_SIZE, DEFAULT_MAX_PARALLEL_CHUNKS, @@ -569,7 +570,7 @@ describe("plan() — golden planDir + planHash determinism", () => { ); it( - "produces a byte-identical planHash on a second invocation", + "shares one byte-identical execution plan between the builder and legacy v1 wrapper", async () => { const planDirA = join(runRoot, "plan-determinism-a"); const planDirB = join(runRoot, "plan-determinism-b"); @@ -577,12 +578,26 @@ describe("plan() — golden planDir + planHash determinism", () => { mkdirSync(planDirB, { recursive: true }); const config = { fps: 30 as const, width: 320, height: 240, format: "mp4" as const }; - const a = await plan(projectDir, config, planDirA); + const a = await buildLocalExecutionPlan(projectDir, config, planDirA); const b = await plan(projectDir, config, planDirB); - expect(a.planHash).toBe(b.planHash); + expect(a.executionPlanDir).toBe(planDirA); + expect(a.executionPlanHash).toBe(b.planHash); expect(a.chunkCount).toBe(b.chunkCount); expect(a.totalFrames).toBe(b.totalFrames); + expect(Object.keys(b)).toEqual([ + "planDir", + "planProtocol", + "planHash", + "chunkCount", + "totalFrames", + "fps", + "width", + "height", + "format", + "ffmpegVersion", + "producerVersion", + ]); // Encoder JSON must be byte-identical — its bytes feed planHash, so any // drift here would silently change the hash framing. diff --git a/packages/producer/src/services/distributed/plan.ts b/packages/producer/src/services/distributed/plan.ts index ef989c3676..7712e995be 100644 --- a/packages/producer/src/services/distributed/plan.ts +++ b/packages/producer/src/services/distributed/plan.ts @@ -1,9 +1,10 @@ /** * Activity A of the distributed render pipeline. * - * `plan(projectDir, config, planDir)` composes the existing render stages - * (compile → probe → extract videos → audio → freeze) into a self-contained - * `/` directory tree that downstream chunk workers consume: + * `buildLocalExecutionPlan(projectDir, config, executionPlanDir)` composes the + * existing render stages (compile → probe → extract videos → audio → freeze) + * into a self-contained local execution directory that downstream chunk + * workers consume: * * / * ├── plan.json @@ -16,8 +17,9 @@ * └── chunks.json * * Pure function over local paths. No networking. Two invocations with the - * same inputs produce the same `planHash` — adapters use that contract to - * short-circuit `plan()` on workflow replay. + * same inputs produce the same execution-plan hash. Transport adapters use + * that representation either through the legacy v1 `plan()` wrapper or the + * v2 manifest/CAS publisher. * * Banned configurations (GPU encode, hardware browser GL, system primary * fonts) are rejected at plan time via `planValidation.ts` so chunk workers @@ -77,7 +79,7 @@ import { readFfmpegVersion, readProducerVersion, } from "./shared.js"; -import { CURRENT_PLAN_PROTOCOL, type PlanProtocolV1Descriptor } from "./planProtocol.js"; +import { PLAN_PROTOCOL_V1, type PlanProtocolV1Descriptor } from "./planProtocol.js"; import { measurePlanSizeBreakdown, type PlanSizeBreakdown, @@ -253,9 +255,35 @@ export interface DistributedRenderConfig { variables?: Record; } +/** Shared local representation consumed by both distributed plan transports. */ +export interface LocalExecutionPlan { + executionPlanDir: string; + executionPlanHash: string; + chunkCount: number; + totalFrames: number; + fps: 24 | 30 | 60; + width: number; + height: number; + format: DistributedFormat; + ffmpegVersion: string; + producerVersion: string; +} + +export interface BuildLocalExecutionPlanOptions { + /** + * Transport-specific size ceiling for the local execution representation. + * Legacy v1 uses the monolithic plan-directory limit; v2 disables that + * transport cap because it publishes role-scoped content-addressed objects. + */ + readonly executionPlanSizeLimitBytes?: number; +} + /** - * Result of {@link plan}. The `planHash` is the content-addressed identifier - * that adapters key replay short-circuits off of. + * Result of the legacy v1 {@link plan} wrapper. The `planHash` is the + * content-addressed identifier that adapters key replay short-circuits off of. + * + * @deprecated Use `planV2()` or `planV2WithPublisher()` for new integrations. + * The v1 result remains supported for existing transports. */ export interface PlanResult { planDir: string; @@ -798,15 +826,16 @@ export function resolveDistributedEngineConfig(config: DistributedRenderConfig): } /** - * Activity A of the distributed render pipeline. Produces a self-contained - * `/` from a project + config. See module docstring for the - * directory layout. + * Build the shared local execution representation used by both transport + * protocols. See the module docstring for the directory layout. */ -export async function plan( +export async function buildLocalExecutionPlan( projectDir: string, config: DistributedRenderConfig, - planDir: string, -): Promise { + executionPlanDir: string, + options: Readonly = {}, +): Promise { + const planDir = executionPlanDir; // Plan-time validation. Rejections here surface as typed errors with // non-retryable codes so workflow adapters don't waste retry budget on // banned configs. Runs BEFORE any directory creation so a banned input @@ -820,7 +849,10 @@ export async function plan( if (!existsSync(planDir)) mkdirSync(planDir, { recursive: true }); const log = config.logger ?? defaultLogger; - const sizeLimitBytes = config.planDirSizeLimitBytes ?? PLAN_DIR_SIZE_LIMIT_BYTES; + const sizeLimitBytes = + options.executionPlanSizeLimitBytes ?? + config.planDirSizeLimitBytes ?? + PLAN_DIR_SIZE_LIMIT_BYTES; const abortSignal = config.abortSignal; const assertNotAborted = (): void => { if (abortSignal?.aborted) { @@ -1193,9 +1225,8 @@ export async function plan( }); return { - planDir, - planProtocol: CURRENT_PLAN_PROTOCOL, - planHash, + executionPlanDir: planDir, + executionPlanHash: planHash, chunkCount, totalFrames, fps: config.fps, @@ -1206,3 +1237,30 @@ export async function plan( producerVersion, }; } + +/** + * Legacy v1 transport wrapper around {@link buildLocalExecutionPlan}. + * + * @deprecated Use `planV2()` or `planV2WithPublisher()` for new integrations. + * This wrapper, its layout, and its result shape remain supported. + */ +export async function plan( + projectDir: string, + config: DistributedRenderConfig, + planDir: string, +): Promise { + const executionPlan = await buildLocalExecutionPlan(projectDir, config, planDir); + return { + planDir: executionPlan.executionPlanDir, + planProtocol: PLAN_PROTOCOL_V1, + planHash: executionPlan.executionPlanHash, + chunkCount: executionPlan.chunkCount, + totalFrames: executionPlan.totalFrames, + fps: executionPlan.fps, + width: executionPlan.width, + height: executionPlan.height, + format: executionPlan.format, + ffmpegVersion: executionPlan.ffmpegVersion, + producerVersion: executionPlan.producerVersion, + }; +} diff --git a/packages/producer/src/services/distributed/planProtocol.test.ts b/packages/producer/src/services/distributed/planProtocol.test.ts index 699c895c95..c72e200eb0 100644 --- a/packages/producer/src/services/distributed/planProtocol.test.ts +++ b/packages/producer/src/services/distributed/planProtocol.test.ts @@ -13,6 +13,7 @@ import { getDistributedRenderCapabilities, PLAN_ARTIFACT_LAYOUT, PLAN_HASH_SCHEMA, + PLAN_PROTOCOL_V1, PLAN_PROTOCOL_V2, PLAN_PROTOCOL_UNSUPPORTED, PLAN_SCHEMA_VERSION, @@ -95,6 +96,13 @@ function createReaderPlan(options: { } describe("readPlanProtocol()", () => { + it("keeps the v1 descriptor byte-for-byte and identity-compatible", () => { + expect(PLAN_PROTOCOL_V1).toBe(CURRENT_PLAN_PROTOCOL); + expect(JSON.stringify(PLAN_PROTOCOL_V1)).toBe( + '{"schemaVersion":1,"artifactLayout":"plan-dir-v1","hashSchema":"hyperframes-plan-hash-v1"}', + ); + }); + it("treats an absent descriptor as legacy v1", () => { expect(readPlanProtocol({ planHash: "legacy" })).toBe(CURRENT_PLAN_PROTOCOL); }); diff --git a/packages/producer/src/services/distributed/planProtocol.ts b/packages/producer/src/services/distributed/planProtocol.ts index 2503d72a96..61895d5f7f 100644 --- a/packages/producer/src/services/distributed/planProtocol.ts +++ b/packages/producer/src/services/distributed/planProtocol.ts @@ -36,13 +36,21 @@ export interface PlanProtocolV2Descriptor extends PlanProtocolDescriptor { export type SupportedPlanProtocolDescriptor = PlanProtocolV1Descriptor | PlanProtocolV2Descriptor; -/** Descriptor written by the current producer and accepted by v1 workers. */ -export const CURRENT_PLAN_PROTOCOL: Readonly = Object.freeze({ +/** Descriptor for the legacy v1 execution-directory transport. */ +export const PLAN_PROTOCOL_V1: Readonly = Object.freeze({ schemaVersion: PLAN_SCHEMA_VERSION, artifactLayout: PLAN_ARTIFACT_LAYOUT, hashSchema: PLAN_HASH_SCHEMA, }); +/** + * Descriptor written by the legacy v1 planner and accepted by v1 workers. + * + * @deprecated Use {@link PLAN_PROTOCOL_V1}. Kept as an identity-preserving + * alias for existing integrations. + */ +export const CURRENT_PLAN_PROTOCOL: Readonly = PLAN_PROTOCOL_V1; + /** Explicit opt-in descriptor for the content-addressed v2 transport layout. */ export const PLAN_PROTOCOL_V2: Readonly = Object.freeze({ schemaVersion: PLAN_V2_SCHEMA_VERSION, @@ -70,14 +78,14 @@ export const DISTRIBUTED_RENDER_CAPABILITIES: Readonly { const protocol = readPlanProtocol(planJson, capabilities); - if (protocol !== CURRENT_PLAN_PROTOCOL) { + if (protocol !== PLAN_PROTOCOL_V1) { throw new PlanProtocolUnsupportedError( "content-addressed v2 plan must be materialized before v1 layout access", ); } - return CURRENT_PLAN_PROTOCOL; + return PLAN_PROTOCOL_V1; } diff --git a/packages/producer/src/services/distributed/planSizeCap.test.ts b/packages/producer/src/services/distributed/planSizeCap.test.ts index b23edc5bdb..de1b389c2d 100644 --- a/packages/producer/src/services/distributed/planSizeCap.test.ts +++ b/packages/producer/src/services/distributed/planSizeCap.test.ts @@ -25,7 +25,7 @@ import { PlanTooLargeError, plan, } from "./plan.js"; -import { planV2, readPlanV2Manifest } from "./planV2.js"; +import { getPlanV2ExecutionPlanHash, planV2, readPlanV2Manifest } from "./planV2.js"; import { measurePlanSizeBreakdown } from "./planSize.js"; import { DISTRIBUTED_DURATION_OUT_OF_RANGE } from "../render/planValidation.js"; @@ -262,7 +262,8 @@ describe("plan() under size cap", () => { const manifest = readPlanV2Manifest(v2.planDir); expect(v2.planProtocol.schemaVersion).toBe(2); expect(v2.planHash).toBe(manifest.planHash); - expect(v2.sourcePlanV1Hash).toBe(manifest.sourcePlanV1Hash); + expect(getPlanV2ExecutionPlanHash(v2)).toBe(manifest.sourcePlanV1Hash); + expect(Object.hasOwn(v2, "executionPlanHash")).toBe(false); expect(manifest.artifacts.length).toBeGreaterThan(0); }, TIMEOUT_MS, diff --git a/packages/producer/src/services/distributed/planV2.test.ts b/packages/producer/src/services/distributed/planV2.test.ts index fe02ac7c20..fee951f078 100644 --- a/packages/producer/src/services/distributed/planV2.test.ts +++ b/packages/producer/src/services/distributed/planV2.test.ts @@ -18,11 +18,14 @@ import { CURRENT_PLAN_PROTOCOL } from "./planProtocol.js"; import { FFMPEG_VERSION_MISMATCH, renderChunk, RenderChunkValidationError } from "./renderChunk.js"; import { createPlanV2FromV1, + createPlanV2FromExecutionPlan, + getPlanV2ExecutionPlanHash, listPlanV2ArtifactsForTarget, materializePlanV2Target, PLAN_V2_INTEGRITY_UNRECOVERABLE, PlanV2IntegrityError, publishPlanV2FromV1, + publishPlanV2FromExecutionPlan, readPlanV2Manifest, validatePlanV2MaterializedTarget, } from "./planV2.js"; @@ -164,19 +167,67 @@ function createV1Plan( } describe("Plan v2 manifest", () => { - it("is deterministic and keeps the v1 transport opt-in", () => { + it("is deterministic without changing the manifest or result wire shapes", () => { const root = tempPath("hf-plan-v2-determinism-"); const v1 = createV1Plan(root, { audio: true }); - const first = createPlanV2FromV1(v1, join(root, "v2-a")); - const second = createPlanV2FromV1(v1, join(root, "v2-b")); + const first = createPlanV2FromExecutionPlan(v1, join(root, "v2-a")); + const second = createPlanV2FromExecutionPlan(v1, join(root, "v2-b")); + const serializedManifest = readFileSync(first.manifestPath, "utf-8"); + const manifest = JSON.parse(serializedManifest) as Record; expect(first.planHash).toBe(second.planHash); - expect(readFileSync(first.manifestPath, "utf-8")).toBe( - readFileSync(second.manifestPath, "utf-8"), - ); + expect(serializedManifest).toBe(readFileSync(second.manifestPath, "utf-8")); expect(first.planHash).not.toBe(first.sourcePlanV1Hash); + expect(getPlanV2ExecutionPlanHash(first)).toBe(first.sourcePlanV1Hash); expect(first.planProtocol.schemaVersion).toBe(2); expect(first.limitations.videoDependencyMode).toBe("exact-rendered-frames"); + expect(Object.keys(first)).toEqual([ + "planDir", + "manifestPath", + "planProtocol", + "planHash", + "sourcePlanV1Hash", + "chunkCount", + "totalFrames", + "fps", + "width", + "height", + "format", + "ffmpegVersion", + "producerVersion", + "limitations", + ]); + expect(Object.keys(manifest)).toEqual([ + "artifacts", + "chunkCount", + "ffmpegVersion", + "format", + "fps", + "height", + "limitations", + "planHash", + "producerVersion", + "protocol", + "sourcePlanV1Hash", + "totalFrames", + "width", + ]); + expect(Object.hasOwn(first, "executionPlanHash")).toBe(false); + expect(Object.hasOwn(manifest, "executionPlanHash")).toBe(false); + }); + + it("keeps legacy conversion names as byte-identical compatibility aliases", async () => { + const root = tempPath("hf-plan-v2-compat-aliases-"); + const executionPlanDir = createV1Plan(root, { audio: true }); + const canonical = createPlanV2FromExecutionPlan(executionPlanDir, join(root, "canonical")); + const compatibility = createPlanV2FromV1(executionPlanDir, join(root, "compatibility")); + const publisher = new LocalPlanV2ArtifactPublisher(join(root, "published")); + const published = await publishPlanV2FromExecutionPlan(executionPlanDir, publisher); + + expect(readFileSync(canonical.manifestPath)).toEqual(readFileSync(compatibility.manifestPath)); + expect(getPlanV2ExecutionPlanHash(published)).toBe(getPlanV2ExecutionPlanHash(canonical)); + expect(published.sourcePlanV1Hash).toBe(canonical.sourcePlanV1Hash); + expect(Object.hasOwn(published, "executionPlanHash")).toBe(false); }); it("accepts and materializes the same bounded timing produced for v1", () => { @@ -220,7 +271,7 @@ describe("Plan v2 manifest", () => { expect(caught).toHaveProperty("code", PLAN_V2_INTEGRITY_UNRECOVERABLE); expect(caught).toHaveProperty( "message", - expect.stringMatching(/v1 plan content fingerprint does not match/), + expect.stringMatching(/execution plan content fingerprint does not match/), ); expect(existsSync(destination)).toBe(false); }); @@ -584,6 +635,16 @@ describe("Plan v2 manifest", () => { result.planHash, ); expect(chunk.sourcePlanV1Hash).toBe(result.sourcePlanV1Hash); + expect(Object.keys(chunk)).toEqual([ + "planDir", + "target", + "planHash", + "sourcePlanV1Hash", + "artifactCount", + "sizeBytes", + "audioPath", + ]); + expect(Object.hasOwn(chunk, "executionPlanHash")).toBe(false); }); it("uses v2 subset integrity instead of the whole-v1 plan hash", async () => { @@ -689,7 +750,7 @@ describe("Plan v2 artifact publisher", () => { const v1 = createV1Plan(root, { audio: true }); const destination = join(root, "v2"); const publisher = new LocalPlanV2ArtifactPublisher(destination); - const manifest = await publishPlanV2FromV1(v1, publisher); + const manifest = await publishPlanV2FromExecutionPlan(v1, publisher); const artifact = manifest.artifacts.find( (candidate) => candidate.path === "compiled/asset.txt", ); diff --git a/packages/producer/src/services/distributed/planV2.ts b/packages/producer/src/services/distributed/planV2.ts index faf768528a..4131a58f2c 100644 --- a/packages/producer/src/services/distributed/planV2.ts +++ b/packages/producer/src/services/distributed/planV2.ts @@ -4,7 +4,7 @@ * V2 deliberately separates transport from execution. The transport root is * a small immutable `plan.json` manifest plus sha256-addressed blobs. Workers * select and materialize only the dependencies for their role, then invoke - * the existing v1 execution functions on the verified local layout. + * the shared execution functions on the verified local layout. * * Video-frame dependencies are derived by evaluating the engine's own * FrameLookupTable at every captured global frame. If legacy video metadata @@ -33,7 +33,7 @@ import { dirname, isAbsolute, join, relative, resolve, sep } from "node:path"; import { createFrameLookupTable, type ExtractedFrames } from "@hyperframes/engine"; import { recomputePlanHashFromPlanDir, type ChunkSliceJson } from "../render/stages/freezePlan.js"; import { canonicalJsonStringify, sha256Hex } from "../render/stages/planHash.js"; -import { type DistributedRenderConfig, plan } from "./plan.js"; +import { buildLocalExecutionPlan, type DistributedRenderConfig } from "./plan.js"; import { PLAN_PROTOCOL_V2, readPlanProtocolV1, @@ -69,7 +69,7 @@ export type PlanV2MaterializationTarget = | Readonly<{ role: "assembler" }>; export interface PlanV2Artifact { - /** POSIX path in the materialized v1 execution directory. */ + /** POSIX path in the materialized local execution directory. */ readonly path: string; readonly sha256: string; readonly sizeBytes: number; @@ -80,9 +80,14 @@ export interface PlanV2Artifact { export interface PlanV2Manifest { readonly protocol: Readonly; - /** V2 manifest digest; intentionally distinct from the v1 execution hash. */ + /** V2 manifest digest; intentionally distinct from the local execution-plan hash. */ readonly planHash: string; - /** Original execution-plan hash retained for output/replay correlation. */ + /** + * Local execution-plan hash retained for output/replay correlation. + * + * @deprecated This is the compatibility wire name. Use + * {@link getPlanV2ExecutionPlanHash} in new code. + */ readonly sourcePlanV1Hash: string; readonly chunkCount: number; readonly totalFrames: number; @@ -105,6 +110,12 @@ export interface PlanV2Result { readonly manifestPath: string; readonly planProtocol: Readonly; readonly planHash: string; + /** + * Hash of the shared local execution representation. + * + * @deprecated This is the compatibility wire name. Use + * {@link getPlanV2ExecutionPlanHash} in new code. + */ readonly sourcePlanV1Hash: string; readonly chunkCount: number; readonly totalFrames: number; @@ -119,7 +130,7 @@ export interface PlanV2Result { export interface PlanV2WithPublisherOptions { /** - * Parent for the planner-private v1 staging directory. Remote adapters + * Parent for the planner-private local execution directory. Remote adapters * normally leave this unset and use the OS temp directory. The local * compatibility wrapper keeps staging beside its destination so hard links * can avoid a second set of data blocks. @@ -131,6 +142,12 @@ export interface PlanV2MaterializationResult { readonly planDir: string; readonly target: PlanV2MaterializationTarget; readonly planHash: string; + /** + * Hash of the shared local execution representation. + * + * @deprecated This is the compatibility wire name. Use + * {@link getPlanV2ExecutionPlanHash} in new code. + */ readonly sourcePlanV1Hash: string; readonly artifactCount: number; readonly sizeBytes: number; @@ -198,7 +215,7 @@ function listFiles(root: string): Array<{ path: string; absolutePath: string }> return files.sort((a, b) => a.path.localeCompare(b.path)); } -/** Hash large plan artifacts with bounded memory (important above the v1 2 GiB cap). */ +/** Hash large plan artifacts with bounded memory (important above the legacy 2 GiB cap). */ function sha256File(path: string): string { const hash = createHash("sha256"); const buffer = Buffer.allocUnsafe(1024 * 1024); @@ -226,8 +243,8 @@ function assertValidExtractionCacheCompleteSentinel(path: string): void { } } -function validateExtractionCacheCompleteSentinels(planV1Dir: string): void { - const videoRoot = join(planV1Dir, "video-frames"); +function validateExtractionCacheCompleteSentinels(executionPlanDir: string): void { + const videoRoot = join(executionPlanDir, "video-frames"); if (!existsSync(videoRoot)) return; for (const videoEntry of readdirSync(videoRoot, { withFileTypes: true })) { @@ -259,7 +276,7 @@ function resolveExtractedVideoOutputDir(planDir: string, videoId: string): strin return outputDir; } -// This is the fail-safe policy table for every v1 artifact class. Keeping the +// This is the fail-safe policy table for every local execution artifact class. Keeping the // branches together makes new artifact classes visibly fall through to both roles. // fallow-ignore-next-line complexity function artifactTargets( @@ -283,14 +300,14 @@ function artifactTargets( assembler: false, }; } - // Unknown future v1 files go to both roles. Over-including is safe; + // Unknown future execution files go to both roles. Over-including is safe; // silently omitting a new execution dependency is not. return { chunks: "all", assembler: true }; } -function listVideoFramePaths(planV1Dir: string, videos: PlanVideosJson): ExtractedFrames[] { +function listVideoFramePaths(executionPlanDir: string, videos: PlanVideosJson): ExtractedFrames[] { return videos.extracted.map((video) => { - const outputDir = resolveExtractedVideoOutputDir(planV1Dir, video.videoId); + const outputDir = resolveExtractedVideoOutputDir(executionPlanDir, video.videoId); const frameNames = readdirSync(outputDir).sort(); const framePaths = new Map(); for (const frameName of frameNames) { @@ -370,26 +387,26 @@ function parseChunkSlices(value: unknown): ChunkSliceJson[] { } function buildVideoChunkDependencies( - planV1Dir: string, + executionPlanDir: string, dimensions: Record, ): { mode: "exact-rendered-frames" | "full-source-pack"; dependencies: ReadonlyMap | null; } { - const videoRoot = join(planV1Dir, "video-frames"); + const videoRoot = join(executionPlanDir, "video-frames"); const hasExtractedFrames = existsSync(videoRoot) && listFiles(videoRoot).length > 0; - const videosPath = join(planV1Dir, PLAN_VIDEOS_META_RELATIVE_PATH); + const videosPath = join(executionPlanDir, PLAN_VIDEOS_META_RELATIVE_PATH); if (!existsSync(videosPath)) { return hasExtractedFrames ? { mode: "full-source-pack", dependencies: null } : { mode: "exact-rendered-frames", dependencies: new Map() }; } - const chunksPath = join(planV1Dir, "meta", "chunks.json"); + const chunksPath = join(executionPlanDir, "meta", "chunks.json"); const videos = readJsonFile(videosPath, PLAN_VIDEOS_META_RELATIVE_PATH); const chunks = readJsonFile(chunksPath, "meta/chunks.json"); const parsedVideos = parsePlanVideosJson(videos); const parsedChunks = parseChunkSlices(chunks); - const extracted = listVideoFramePaths(planV1Dir, parsedVideos); + const extracted = listVideoFramePaths(executionPlanDir, parsedVideos); const table = createFrameLookupTable(parsedVideos.videos, extracted); const fpsNum = readPositiveInteger(dimensions.fpsNum, "dimensions.fpsNum"); const fpsDen = readPositiveInteger(dimensions.fpsDen, "dimensions.fpsDen"); @@ -399,7 +416,7 @@ function buildVideoChunkDependencies( for (let frame = chunk.startFrame; frame < chunk.endFrame; frame++) { const globalTime = (frame * fpsDen) / fpsNum; for (const payload of table.getActiveFramePayloads(globalTime).values()) { - const path = relative(resolve(planV1Dir), payload.framePath).split(sep).join("/"); + const path = relative(resolve(executionPlanDir), payload.framePath).split(sep).join("/"); assertSafeRelativePath(path); const indexes = mutable.get(path) ?? new Set(); indexes.add(chunk.index); @@ -445,38 +462,40 @@ interface PlanV2Publication { // Artifact classification intentionally keeps every fail-safe branch together. // fallow-ignore-next-line complexity -function buildPlanV2Publication(planV1Dir: string): PlanV2Publication { - const v1PlanPath = join(planV1Dir, "plan.json"); - if (!existsSync(v1PlanPath)) { - throw new PlanV2IntegrityError(`v1 plan is missing plan.json: ${v1PlanPath}`); - } - const v1PlanValue = readJsonFile(v1PlanPath, "v1 plan.json"); - if (!isRecord(v1PlanValue)) { - throw new PlanV2IntegrityError("v1 plan.json must be an object"); - } - const v1Plan = v1PlanValue; - readPlanProtocolV1(v1Plan); - const sourcePlanV1Hash = v1Plan.planHash; - if (!isSha256(sourcePlanV1Hash)) { - throw new PlanV2IntegrityError("v1 plan.json.planHash must be a sha256 digest"); - } - const recomputedSourcePlanV1Hash = recomputePlanHashFromPlanDir(planV1Dir); - if (recomputedSourcePlanV1Hash !== sourcePlanV1Hash) { +function buildPlanV2Publication(executionPlanDir: string): PlanV2Publication { + const executionPlanPath = join(executionPlanDir, "plan.json"); + if (!existsSync(executionPlanPath)) { + throw new PlanV2IntegrityError(`execution plan is missing plan.json: ${executionPlanPath}`); + } + const executionPlanValue = readJsonFile(executionPlanPath, "execution plan.json"); + if (!isRecord(executionPlanValue)) { + throw new PlanV2IntegrityError("execution plan.json must be an object"); + } + const executionPlan = executionPlanValue; + // The shared local representation intentionally retains the v1-compatible + // descriptor while both legacy readers and v2 materialization are supported. + readPlanProtocolV1(executionPlan); + const executionPlanHash = executionPlan.planHash; + if (!isSha256(executionPlanHash)) { + throw new PlanV2IntegrityError("execution plan.json.planHash must be a sha256 digest"); + } + const recomputedExecutionPlanHash = recomputePlanHashFromPlanDir(executionPlanDir); + if (recomputedExecutionPlanHash !== executionPlanHash) { throw new PlanV2IntegrityError( - `v1 plan content fingerprint does not match plan.json.planHash: ` + - `expected ${sourcePlanV1Hash}, recomputed ${recomputedSourcePlanV1Hash}`, + `execution plan content fingerprint does not match plan.json.planHash: ` + + `expected ${executionPlanHash}, recomputed ${recomputedExecutionPlanHash}`, ); } const artifacts: PlanV2Artifact[] = []; const blobs = new Map(); - const dimensions = v1Plan.dimensions; + const dimensions = executionPlan.dimensions; if (!isRecord(dimensions)) { - throw new PlanV2IntegrityError("v1 plan.json.dimensions must be an object"); + throw new PlanV2IntegrityError("execution plan.json.dimensions must be an object"); } - validateExtractionCacheCompleteSentinels(planV1Dir); - const videoDependencyPlan = buildVideoChunkDependencies(planV1Dir, dimensions); - for (const file of listFiles(planV1Dir)) { + validateExtractionCacheCompleteSentinels(executionPlanDir); + const videoDependencyPlan = buildVideoChunkDependencies(executionPlanDir, dimensions); + for (const file of listFiles(executionPlanDir)) { if (isExtractionCacheCompleteSentinelPath(file.path)) continue; const targets = artifactTargets(file.path, videoDependencyPlan.dependencies); if ( @@ -502,15 +521,17 @@ function buildPlanV2Publication(planV1Dir: string): PlanV2Publication { const base: Omit = { protocol: PLAN_PROTOCOL_V2, - sourcePlanV1Hash, - chunkCount: readPositiveInteger(v1Plan.chunkCount, "chunkCount"), - totalFrames: readPositiveInteger(v1Plan.totalFrames, "totalFrames"), - fps: readV1PlanFps(dimensions), + // Retain the established manifest key byte-for-byte. It now acts as the + // wire alias for the neutral local execution-plan hash. + sourcePlanV1Hash: executionPlanHash, + chunkCount: readPositiveInteger(executionPlan.chunkCount, "chunkCount"), + totalFrames: readPositiveInteger(executionPlan.totalFrames, "totalFrames"), + fps: readExecutionPlanFps(dimensions), width: readPositiveInteger(dimensions.width, "dimensions.width"), height: readPositiveInteger(dimensions.height, "dimensions.height"), format: readDistributedFormat(dimensions.format), - ffmpegVersion: readString(v1Plan.ffmpegVersion, "ffmpegVersion"), - producerVersion: readString(v1Plan.producerVersion, "producerVersion"), + ffmpegVersion: readString(executionPlan.ffmpegVersion, "ffmpegVersion"), + producerVersion: readString(executionPlan.producerVersion, "producerVersion"), limitations: { videoDependencyMode: videoDependencyPlan.mode }, artifacts, }; @@ -523,12 +544,15 @@ function buildPlanV2Publication(planV1Dir: string): PlanV2Publication { }; } -/** Convert a frozen v1 execution directory into the immutable v2 transport. */ -export function createPlanV2FromV1(planV1Dir: string, planV2Dir: string): PlanV2Result { +/** Publish a frozen local execution directory into an immutable local v2 transport. */ +export function createPlanV2FromExecutionPlan( + executionPlanDir: string, + planV2Dir: string, +): PlanV2Result { if (existsSync(planV2Dir)) { throw new PlanV2IntegrityError(`output directory already exists: ${planV2Dir}`); } - const publication = buildPlanV2Publication(planV1Dir); + const publication = buildPlanV2Publication(executionPlanDir); mkdirSync(dirname(planV2Dir), { recursive: true }); const tempDir = mkdtempSync(join(dirname(planV2Dir), ".plan-v2-build-")); try { @@ -548,12 +572,22 @@ export function createPlanV2FromV1(planV1Dir: string, planV2Dir: string): PlanV2 } } -export async function publishPlanV2FromV1( - planV1Dir: string, +/** + * Compatibility alias for {@link createPlanV2FromExecutionPlan}. + * + * @deprecated Use `createPlanV2FromExecutionPlan`. + */ +export function createPlanV2FromV1(executionPlanDir: string, planV2Dir: string): PlanV2Result { + return createPlanV2FromExecutionPlan(executionPlanDir, planV2Dir); +} + +/** Publish a frozen local execution directory through a storage-neutral v2 publisher. */ +export async function publishPlanV2FromExecutionPlan( + executionPlanDir: string, publisher: PlanV2ArtifactPublisher, ): Promise { try { - const publication = buildPlanV2Publication(planV1Dir); + const publication = buildPlanV2Publication(executionPlanDir); const concurrency = 16; for (let offset = 0; offset < publication.blobs.length; offset += concurrency) { const batch = publication.blobs.slice(offset, offset + concurrency); @@ -574,6 +608,18 @@ export async function publishPlanV2FromV1( } } +/** + * Compatibility alias for {@link publishPlanV2FromExecutionPlan}. + * + * @deprecated Use `publishPlanV2FromExecutionPlan`. + */ +export async function publishPlanV2FromV1( + executionPlanDir: string, + publisher: PlanV2ArtifactPublisher, +): Promise { + return publishPlanV2FromExecutionPlan(executionPlanDir, publisher); +} + /** * Plan into a storage-neutral publisher. Implementations may write to a local * directory, S3, GCS, or another durable CAS. Only the planner's private @@ -586,14 +632,12 @@ export async function planV2WithPublisher( publisher: PlanV2ArtifactPublisher, options: Readonly = {}, ): Promise { - const stagingRoot = mkdtempSync(join(options.stagingParentDir ?? tmpdir(), ".plan-v2-source-")); + const stagingRoot = mkdtempSync(join(options.stagingParentDir ?? tmpdir(), ".execution-plan-")); try { - await plan( - projectDir, - { ...config, planDirSizeLimitBytes: Number.MAX_SAFE_INTEGER }, - stagingRoot, - ); - return await publishPlanV2FromV1(stagingRoot, publisher); + await buildLocalExecutionPlan(projectDir, config, stagingRoot, { + executionPlanSizeLimitBytes: Number.MAX_SAFE_INTEGER, + }); + return await publishPlanV2FromExecutionPlan(stagingRoot, publisher); } catch (error) { try { await publisher.abort(); @@ -607,9 +651,9 @@ export async function planV2WithPublisher( } /** - * Plan directly into v2. The large v1 directory exists only as local staging; - * its historical 2 GiB transport cap is disabled because no monolithic - * archive is emitted. + * Plan directly into v2. The shared local execution representation exists + * only as planner-private staging; the legacy 2 GiB transport cap is disabled + * because no monolithic archive is emitted. */ export async function planV2( projectDir: string, @@ -632,7 +676,7 @@ function resultFromManifest(planV2Dir: string, manifest: PlanV2Manifest): PlanV2 manifestPath: join(planV2Dir, "plan.json"), planProtocol: PLAN_PROTOCOL_V2, planHash: manifest.planHash, - sourcePlanV1Hash: manifest.sourcePlanV1Hash, + sourcePlanV1Hash: getPlanV2ExecutionPlanHash(manifest), chunkCount: manifest.chunkCount, totalFrames: manifest.totalFrames, fps: manifest.fps, @@ -645,6 +689,16 @@ function resultFromManifest(planV2Dir: string, manifest: PlanV2Manifest): PlanV2 }; } +/** + * Return the shared local execution-plan hash from a v2 manifest while + * preserving the established `sourcePlanV1Hash` wire key. + */ +export function getPlanV2ExecutionPlanHash( + plan: Readonly>, +): string { + return plan.sourcePlanV1Hash; +} + function readString(value: unknown, field: string): string { if (typeof value !== "string" || value.length === 0) { throw new PlanV2IntegrityError(`${field} must be a non-empty string`); @@ -673,7 +727,7 @@ function readSupportedFps(value: unknown): 24 | 30 | 60 { return value; } -function readV1PlanFps(dimensions: Record): 24 | 30 | 60 { +function readExecutionPlanFps(dimensions: Record): 24 | 30 | 60 { const fpsDen = readPositiveInteger(dimensions.fpsDen, "dimensions.fpsDen"); if (fpsDen !== 1) { throw new PlanV2IntegrityError("dimensions.fpsDen must be 1 for plan v2"); @@ -744,7 +798,7 @@ function parsePlanV2Manifest(value: unknown): Readonly { paths.add(artifact.path); } if (!paths.has("plan.json")) - throw new PlanV2IntegrityError("manifest must include the v1 plan.json artifact"); + throw new PlanV2IntegrityError("manifest must include the local execution plan.json artifact"); if ( !isRecord(value.limitations) || (value.limitations.videoDependencyMode !== "exact-rendered-frames" && @@ -830,7 +884,7 @@ function verifyBlob(planV2Dir: string, artifact: Readonly): stri } /** - * Verify every selected blob first, then atomically publish a v1-compatible + * Verify every selected blob first, then atomically publish the shared local * execution directory. The marker lets execution functions revalidate the * selected subset without requiring assembler-only audio in chunk workers. */ @@ -858,7 +912,7 @@ export function materializePlanV2Target( } // Plan v2 transports only files, so a chunk where a video is inactive has // no selected frame artifact from which to create its per-video directory. - // renderChunk consumes a v1-compatible layout and intentionally validates + // renderChunk consumes the shared local layout and intentionally validates // every extracted-video directory from meta/videos.json. Recreate those // zero-byte structural directories without downloading unused frame data. if (target.role === "chunk") materializeExtractedVideoDirectories(tempDir); @@ -876,7 +930,7 @@ export function materializePlanV2Target( planDir: destinationDir, target, planHash: manifest.planHash, - sourcePlanV1Hash: manifest.sourcePlanV1Hash, + sourcePlanV1Hash: getPlanV2ExecutionPlanHash(manifest), artifactCount: artifacts.length, sizeBytes: artifacts.reduce((sum, artifact) => sum + artifact.sizeBytes, 0), audioPath: diff --git a/packages/producer/src/services/distributed/publicExports.test.ts b/packages/producer/src/services/distributed/publicExports.test.ts index d634ec6d24..cce59624fb 100644 --- a/packages/producer/src/services/distributed/publicExports.test.ts +++ b/packages/producer/src/services/distributed/publicExports.test.ts @@ -83,6 +83,7 @@ describe("@hyperframes/producer/distributed (subpath)", () => { artifactLayout: "plan-dir-v1", hashSchema: "hyperframes-plan-hash-v1", }); + expect(distributedSubpath.PLAN_PROTOCOL_V1).toBe(distributedSubpath.CURRENT_PLAN_PROTOCOL); expect(distributedSubpath.DISTRIBUTED_RENDER_CAPABILITIES.roles).toEqual({ planner: { produces: [distributedSubpath.CURRENT_PLAN_PROTOCOL, distributedSubpath.PLAN_PROTOCOL_V2], @@ -100,6 +101,11 @@ describe("@hyperframes/producer/distributed (subpath)", () => { expect(typeof distributedSubpath.readPlanProtocol).toBe("function"); expect(typeof distributedSubpath.planV2).toBe("function"); expect(typeof distributedSubpath.planV2WithPublisher).toBe("function"); + expect(typeof distributedSubpath.createPlanV2FromExecutionPlan).toBe("function"); + expect(typeof distributedSubpath.publishPlanV2FromExecutionPlan).toBe("function"); + expect(typeof distributedSubpath.getPlanV2ExecutionPlanHash).toBe("function"); + // Deprecated compatibility aliases remain available. + expect(typeof distributedSubpath.createPlanV2FromV1).toBe("function"); expect(typeof distributedSubpath.publishPlanV2FromV1).toBe("function"); expect(typeof distributedSubpath.LocalPlanV2ArtifactPublisher).toBe("function"); expect(typeof distributedSubpath.renderChunkV2).toBe("function"); @@ -119,6 +125,7 @@ describe("@hyperframes/producer (main entry)", () => { it("re-exports the plan protocol contract", () => { expect(producerIndex.CURRENT_PLAN_PROTOCOL).toBe(distributedSubpath.CURRENT_PLAN_PROTOCOL); + expect(producerIndex.PLAN_PROTOCOL_V1).toBe(distributedSubpath.PLAN_PROTOCOL_V1); expect(producerIndex.DISTRIBUTED_RENDER_CAPABILITIES).toBe( distributedSubpath.DISTRIBUTED_RENDER_CAPABILITIES, ); @@ -127,6 +134,11 @@ describe("@hyperframes/producer (main entry)", () => { expect(producerIndex.PLAN_V2_INTEGRITY_UNRECOVERABLE).toBe("PLAN_V2_INTEGRITY_UNRECOVERABLE"); expect(typeof producerIndex.readPlanProtocol).toBe("function"); expect(typeof producerIndex.planV2WithPublisher).toBe("function"); + expect(typeof producerIndex.createPlanV2FromExecutionPlan).toBe("function"); + expect(typeof producerIndex.publishPlanV2FromExecutionPlan).toBe("function"); + expect(typeof producerIndex.getPlanV2ExecutionPlanHash).toBe("function"); + // Deprecated compatibility aliases remain available. + expect(typeof producerIndex.createPlanV2FromV1).toBe("function"); expect(typeof producerIndex.publishPlanV2FromV1).toBe("function"); expect(typeof producerIndex.PlanV2IntegrityError).toBe("function"); expect(typeof producerIndex.PlanProtocolUnsupportedError).toBe("function"); diff --git a/packages/producer/src/services/render/stages/freezePlan.ts b/packages/producer/src/services/render/stages/freezePlan.ts index 5bf8a738e3..389cdbe1f4 100644 --- a/packages/producer/src/services/render/stages/freezePlan.ts +++ b/packages/producer/src/services/render/stages/freezePlan.ts @@ -14,7 +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 { PLAN_PROTOCOL_V1 } from "../../distributed/planProtocol.js"; import { canonicalJsonStringify, computePlanHash, @@ -356,7 +356,7 @@ export async function freezePlan(input: FreezePlanInput): Promise