refactor(producer): add remote-ready plan v2 publisher (#2792)

packages/producer/src/distributed.ts

@@ -60,6 +60,8 @@ export {
   listPlanV2ArtifactsForTarget,
   materializePlanV2Target,
   planV2,
+  planV2WithPublisher,
+  publishPlanV2FromV1,
   readPlanV2Manifest,
   validatePlanV2MaterializedTarget,
   PLAN_V2_INTEGRITY_UNRECOVERABLE,
@@ -71,7 +73,14 @@ export {
   type PlanV2MaterializationResult,
   type PlanV2MaterializationTarget,
   type PlanV2Result,
+  type PlanV2WithPublisherOptions,
 } from "./services/distributed/planV2.js";
+export {
+  LocalPlanV2ArtifactPublisher,
+  type LocalPlanV2ArtifactPublisherOptions,
+  type PlanV2ArtifactPublisher,
+  type PlanV2PublishBlob,
+} from "./services/distributed/planV2Publisher.js";
 export { assembleV2, renderChunkV2 } from "./services/distributed/planV2Execution.js";
 
 // ── RenderChunk (Activity B) ────────────────────────────────────────────────

packages/producer/src/index.ts

@@ -152,11 +152,13 @@ export {
   materializePlanV2Target,
   plan,
   planV2,
+  planV2WithPublisher,
   PlanV2IntegrityError,
   PlanProtocolUnsupportedError,
   readPlanProtocol,
   readPlanProtocolV1,
   readPlanV2Manifest,
+  publishPlanV2FromV1,
   renderChunk,
   renderChunkV2,
   validatePlanV2MaterializedTarget,
@@ -175,5 +177,8 @@ export {
   type PlanV2MaterializationResult,
   type PlanV2MaterializationTarget,
   type PlanV2Result,
+  type PlanV2WithPublisherOptions,
+  type PlanV2ArtifactPublisher,
+  type PlanV2PublishBlob,
   type SupportedPlanProtocolDescriptor,
 } from "./distributed.js";

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

@@ -5,6 +5,7 @@ import {
   mkdtempSync,
   readFileSync,
   rmSync,
+  statSync,
   symlinkSync,
   writeFileSync,
 } from "node:fs";
@@ -20,9 +21,11 @@ import {
   materializePlanV2Target,
   PLAN_V2_INTEGRITY_UNRECOVERABLE,
   PlanV2IntegrityError,
+  publishPlanV2FromV1,
   readPlanV2Manifest,
   validatePlanV2MaterializedTarget,
 } from "./planV2.js";
+import { LocalPlanV2ArtifactPublisher, type PlanV2ArtifactPublisher } from "./planV2Publisher.js";
 
 const tempDirs: string[] = [];
 
@@ -399,6 +402,149 @@ describe("Plan v2 manifest", () => {
   });
 });
 
+describe("Plan v2 artifact publisher", () => {
+  it("publishes the manifest last and hard-links local immutable blobs", async () => {
+    const root = tempPath("hf-plan-v2-publisher-");
+    const v1 = createV1Plan(root, { audio: true });
+    const destination = join(root, "v2");
+    const publisher = new LocalPlanV2ArtifactPublisher(destination);
+    const manifest = await publishPlanV2FromV1(v1, publisher);
+    const artifact = manifest.artifacts.find(
+      (candidate) => candidate.path === "compiled/asset.txt",
+    );
+    if (artifact === undefined) throw new Error("test fixture is missing compiled/asset.txt");
+    const sourceStat = statSync(join(v1, artifact.path));
+    const blobStat = statSync(
+      join(destination, "artifacts", "sha256", artifact.sha256.slice(0, 2), artifact.sha256),
+    );
+
+    expect(readPlanV2Manifest(destination)).toEqual(manifest);
+    expect({ dev: blobStat.dev, ino: blobStat.ino }).toEqual({
+      dev: sourceStat.dev,
+      ino: sourceStat.ino,
+    });
+  });
+
+  it("falls back to an atomic copy when hard-linking is unavailable", async () => {
+    const root = tempPath("hf-plan-v2-publisher-copy-");
+    const v1 = createV1Plan(root);
+    const destination = join(root, "v2");
+    const publisher = new LocalPlanV2ArtifactPublisher(destination, {
+      linkFile() {
+        throw Object.assign(new Error("cross-device link"), { code: "EXDEV" });
+      },
+    });
+    const manifest = await publishPlanV2FromV1(v1, publisher);
+    const artifact = manifest.artifacts.find(
+      (candidate) => candidate.path === "compiled/asset.txt",
+    );
+    if (artifact === undefined) throw new Error("test fixture is missing compiled/asset.txt");
+    const sourcePath = join(v1, artifact.path);
+    const blobPath = join(
+      destination,
+      "artifacts",
+      "sha256",
+      artifact.sha256.slice(0, 2),
+      artifact.sha256,
+    );
+
+    expect(readFileSync(blobPath)).toEqual(readFileSync(sourcePath));
+    expect({ dev: statSync(blobPath).dev, ino: statSync(blobPath).ino }).not.toEqual({
+      dev: statSync(sourcePath).dev,
+      ino: statSync(sourcePath).ino,
+    });
+  });
+
+  it("produces byte-identical local CAS output through both publication paths", async () => {
+    const root = tempPath("hf-plan-v2-publisher-parity-");
+    const v1 = createV1Plan(root, { audio: true });
+    const directDir = join(root, "direct");
+    const publishedDir = join(root, "published");
+    createPlanV2FromV1(v1, directDir);
+    const publisher = new LocalPlanV2ArtifactPublisher(publishedDir);
+    const manifest = await publishPlanV2FromV1(v1, publisher);
+
+    expect(readFileSync(join(publishedDir, "plan.json"))).toEqual(
+      readFileSync(join(directDir, "plan.json")),
+    );
+    for (const artifact of manifest.artifacts) {
+      const suffix = join("artifacts", "sha256", artifact.sha256.slice(0, 2), artifact.sha256);
+      expect(readFileSync(join(publishedDir, suffix))).toEqual(
+        readFileSync(join(directDir, suffix)),
+      );
+    }
+  });
+
+  it("supports a remote publisher contract with no shared destination filesystem", async () => {
+    const root = tempPath("hf-plan-v2-remote-publisher-");
+    const v1 = createV1Plan(root, { audio: true });
+    const blobs = new Map<string, Buffer>();
+    let committedManifest: string | undefined;
+    const publisher: PlanV2ArtifactPublisher = {
+      async putBlob(blob) {
+        blobs.set(blob.sha256, readFileSync(blob.sourcePath));
+      },
+      async commitManifest(manifestBytes) {
+        committedManifest = manifestBytes;
+      },
+      async abort() {},
+    };
+
+    const manifest = await publishPlanV2FromV1(v1, publisher);
+    expect(committedManifest).toBe(canonicalJsonStringify(manifest));
+    expect(blobs.size).toBe(new Set(manifest.artifacts.map((artifact) => artifact.sha256)).size);
+    for (const artifact of manifest.artifacts) {
+      expect(blobs.get(artifact.sha256)?.byteLength).toBe(artifact.sizeBytes);
+    }
+  });
+
+  it("rejects malformed digests before constructing a local CAS path", async () => {
+    const root = tempPath("hf-plan-v2-publisher-digest-");
+    const sourcePath = join(root, "source");
+    writeFileSync(sourcePath, "bytes");
+    const publisher = new LocalPlanV2ArtifactPublisher(join(root, "v2"));
+
+    await expect(
+      publisher.putBlob({ sourcePath, sha256: "../escape", sizeBytes: 5 }),
+    ).rejects.toThrow("must be a lowercase sha256 digest");
+    await publisher.abort();
+  });
+
+  it("refuses to commit a manifest until every referenced blob is durable", async () => {
+    const root = tempPath("hf-plan-v2-publisher-incomplete-");
+    const publisher = new LocalPlanV2ArtifactPublisher(join(root, "v2"));
+    const digest = "a".repeat(64);
+
+    await expect(
+      publisher.commitManifest(JSON.stringify({ artifacts: [{ sha256: digest }] })),
+    ).rejects.toThrow("cannot commit manifest before referenced blob is durable");
+    await publisher.abort();
+  });
+
+  it("aborts without committing a manifest when a blob publish fails", async () => {
+    const root = tempPath("hf-plan-v2-publisher-failure-");
+    const calls: string[] = [];
+    const publisher: PlanV2ArtifactPublisher = {
+      async putBlob(blob) {
+        calls.push(`blob:${blob.sha256}`);
+        throw new Error("injected blob failure");
+      },
+      async commitManifest() {
+        calls.push("manifest");
+      },
+      async abort() {
+        calls.push("abort");
+      },
+    };
+
+    await expect(publishPlanV2FromV1(createV1Plan(root), publisher)).rejects.toThrow(
+      "injected blob failure",
+    );
+    expect(calls.at(-1)).toBe("abort");
+    expect(calls).not.toContain("manifest");
+  });
+});
+
 describe("Plan v2 hash schema", () => {
   it("does not reuse a raw artifact digest as its manifest hash", () => {
     const root = tempPath("hf-plan-v2-hash-");

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

@@ -27,6 +27,7 @@ import {
   writeFileSync,
 } from "node:fs";
 import { createHash } from "node:crypto";
+import { tmpdir } from "node:os";
 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";
@@ -37,6 +38,13 @@ import {
   readPlanProtocolV1,
   type PlanProtocolV2Descriptor,
 } from "./planProtocol.js";
+import {
+  LocalPlanV2ArtifactPublisher,
+  type PlanV2ArtifactPublisher,
+  type PlanV2PublishBlob,
+} from "./planV2Publisher.js";
+import { PLAN_V2_INTEGRITY_UNRECOVERABLE, PlanV2IntegrityError } from "./planV2Errors.js";
+import { planV2BlobPath } from "./planV2Layout.js";
 import {
   PLAN_AUDIO_RELATIVE_PATH,
   PLAN_VIDEOS_META_RELATIVE_PATH,
@@ -46,23 +54,7 @@ import {
 
 const PLAN_V2_HASH_PREFIX = "hyperframes-plan-manifest-hash-v2\x00";
 export const PLAN_V2_MATERIALIZATION_MARKER = ".hyperframes-plan-v2.json";
-export const PLAN_V2_INTEGRITY_UNRECOVERABLE = "PLAN_V2_INTEGRITY_UNRECOVERABLE" as const;
-
-/**
- * A deterministic plan-v2 validation or integrity failure. Distributed
- * adapters classify both the class name and code as terminal so immutable
- * corruption does not consume the retry budget.
- */
-export class PlanV2IntegrityError 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_V2_INTEGRITY_UNRECOVERABLE = PLAN_V2_INTEGRITY_UNRECOVERABLE;
-
-  constructor(message: string) {
-    super(`[planV2] ${message}`);
-    this.name = "PlanV2IntegrityError";
-  }
-}
+export { PLAN_V2_INTEGRITY_UNRECOVERABLE, PlanV2IntegrityError };
 
 export type PlanV2MaterializationTarget =
   | Readonly<{ role: "chunk"; chunkIndex: number }>
@@ -117,6 +109,16 @@ export interface PlanV2Result {
   readonly limitations: Readonly<PlanV2Limitations>;
 }
 
+export interface PlanV2WithPublisherOptions {
+  /**
+   * Parent for the planner-private v1 staging 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.
+   */
+  readonly stagingParentDir?: string;
+}
+
 export interface PlanV2MaterializationResult {
   readonly planDir: string;
   readonly target: PlanV2MaterializationTarget;
@@ -188,10 +190,6 @@ function listFiles(root: string): Array<{ path: string; absolutePath: string }>
   return files.sort((a, b) => a.path.localeCompare(b.path));
 }
 
-function blobPath(planV2Dir: string, sha256: string): string {
-  return join(planV2Dir, "artifacts", "sha256", sha256.slice(0, 2), sha256);
-}
-
 /** Hash large plan artifacts with bounded memory (important above the v1 2 GiB cap). */
 function sha256File(path: string): string {
   const hash = createHash("sha256");
@@ -447,11 +445,14 @@ function writeBlob(sourcePath: string, destinationPath: string): void {
   }
 }
 
-/** Convert a frozen v1 execution directory into the immutable v2 transport. */
-export function createPlanV2FromV1(planV1Dir: string, planV2Dir: string): PlanV2Result {
-  if (existsSync(planV2Dir)) {
-    throw new PlanV2IntegrityError(`output directory already exists: ${planV2Dir}`);
-  }
+interface PlanV2Publication {
+  readonly manifest: PlanV2Manifest;
+  readonly blobs: readonly Readonly<PlanV2PublishBlob>[];
+}
+
+// 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}`);
@@ -474,63 +475,142 @@ export function createPlanV2FromV1(planV1Dir: string, planV2Dir: string): PlanV2
     );
   }
 
+  const artifacts: PlanV2Artifact[] = [];
+  const blobs = new Map<string, PlanV2PublishBlob>();
+  const dimensions = v1Plan.dimensions;
+  if (!isRecord(dimensions)) {
+    throw new PlanV2IntegrityError("v1 plan.json.dimensions must be an object");
+  }
+  const videoDependencyPlan = buildVideoChunkDependencies(planV1Dir, dimensions);
+  for (const file of listFiles(planV1Dir)) {
+    const targets = artifactTargets(file.path, videoDependencyPlan.dependencies);
+    if (
+      file.path.startsWith("video-frames/") &&
+      targets.chunks !== "all" &&
+      targets.chunks.length === 0 &&
+      !targets.assembler
+    ) {
+      continue;
+    }
+    const sha256 = sha256File(file.absolutePath);
+    const sizeBytes = statSync(file.absolutePath).size;
+    if (!blobs.has(sha256)) {
+      blobs.set(sha256, { sourcePath: file.absolutePath, sha256, sizeBytes });
+    }
+    artifacts.push({
+      path: file.path,
+      sha256,
+      sizeBytes,
+      ...targets,
+    });
+  }
+
+  const base: Omit<PlanV2Manifest, "planHash"> = {
+    protocol: PLAN_PROTOCOL_V2,
+    sourcePlanV1Hash,
+    chunkCount: readPositiveInteger(v1Plan.chunkCount, "chunkCount"),
+    totalFrames: readPositiveInteger(v1Plan.totalFrames, "totalFrames"),
+    fps: readV1PlanFps(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"),
+    limitations: { videoDependencyMode: videoDependencyPlan.mode },
+    artifacts,
+  };
+  return {
+    manifest: {
+      ...base,
+      planHash: computeManifestHash(base),
+    },
+    blobs: [...blobs.values()],
+  };
+}
+
+/** Convert a frozen v1 execution directory into the immutable v2 transport. */
+export function createPlanV2FromV1(planV1Dir: string, planV2Dir: string): PlanV2Result {
+  if (existsSync(planV2Dir)) {
+    throw new PlanV2IntegrityError(`output directory already exists: ${planV2Dir}`);
+  }
+  const publication = buildPlanV2Publication(planV1Dir);
   mkdirSync(dirname(planV2Dir), { recursive: true });
   const tempDir = mkdtempSync(join(dirname(planV2Dir), ".plan-v2-build-"));
   try {
-    const artifacts: PlanV2Artifact[] = [];
-    const dimensions = v1Plan.dimensions;
-    if (!isRecord(dimensions)) {
-      throw new PlanV2IntegrityError("v1 plan.json.dimensions must be an object");
+    for (const blob of publication.blobs) {
+      writeBlob(blob.sourcePath, planV2BlobPath(tempDir, blob.sha256));
     }
-    const videoDependencyPlan = buildVideoChunkDependencies(planV1Dir, dimensions);
-    for (const file of listFiles(planV1Dir)) {
-      const targets = artifactTargets(file.path, videoDependencyPlan.dependencies);
-      if (
-        file.path.startsWith("video-frames/") &&
-        targets.chunks !== "all" &&
-        targets.chunks.length === 0 &&
-        !targets.assembler
-      ) {
-        continue;
-      }
-      const sha256 = sha256File(file.absolutePath);
-      const sizeBytes = statSync(file.absolutePath).size;
-      writeBlob(file.absolutePath, blobPath(tempDir, sha256));
-      artifacts.push({
-        path: file.path,
-        sha256,
-        sizeBytes,
-        ...targets,
-      });
-    }
-
-    const base: Omit<PlanV2Manifest, "planHash"> = {
-      protocol: PLAN_PROTOCOL_V2,
-      sourcePlanV1Hash,
-      chunkCount: readPositiveInteger(v1Plan.chunkCount, "chunkCount"),
-      totalFrames: readPositiveInteger(v1Plan.totalFrames, "totalFrames"),
-      fps: readV1PlanFps(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"),
-      limitations: { videoDependencyMode: videoDependencyPlan.mode },
-      artifacts,
-    };
-    const manifest: PlanV2Manifest = {
-      ...base,
-      planHash: computeManifestHash(base),
-    };
-    writeFileSync(join(tempDir, "plan.json"), canonicalJsonStringify(manifest), "utf-8");
+    writeFileSync(
+      join(tempDir, "plan.json"),
+      canonicalJsonStringify(publication.manifest),
+      "utf-8",
+    );
     renameSync(tempDir, planV2Dir);
-    return resultFromManifest(planV2Dir, manifest);
+    return resultFromManifest(planV2Dir, publication.manifest);
   } catch (error) {
     rmSync(tempDir, { recursive: true, force: true });
     throw error;
   }
 }
 
+export async function publishPlanV2FromV1(
+  planV1Dir: string,
+  publisher: PlanV2ArtifactPublisher,
+): Promise<PlanV2Manifest> {
+  try {
+    const publication = buildPlanV2Publication(planV1Dir);
+    const concurrency = 16;
+    for (let offset = 0; offset < publication.blobs.length; offset += concurrency) {
+      const batch = publication.blobs.slice(offset, offset + concurrency);
+      const results = await Promise.allSettled(batch.map((blob) => publisher.putBlob(blob)));
+      for (const result of results) {
+        if (result.status === "rejected") throw result.reason;
+      }
+    }
+    await publisher.commitManifest(canonicalJsonStringify(publication.manifest));
+    return publication.manifest;
+  } catch (error) {
+    try {
+      await publisher.abort();
+    } catch {
+      // Preserve the publication failure; cleanup is best-effort.
+    }
+    throw error;
+  }
+}
+
+/**
+ * Plan into a storage-neutral publisher. Implementations may write to a local
+ * directory, S3, GCS, or another durable CAS. Only the planner's private
+ * staging directory is local; distributed workers receive adapter-owned
+ * manifest and artifact locators.
+ */
+export async function planV2WithPublisher(
+  projectDir: string,
+  config: DistributedRenderConfig,
+  publisher: PlanV2ArtifactPublisher,
+  options: Readonly<PlanV2WithPublisherOptions> = {},
+): Promise<PlanV2Manifest> {
+  const stagingRoot = mkdtempSync(join(options.stagingParentDir ?? tmpdir(), ".plan-v2-source-"));
+  try {
+    await plan(
+      projectDir,
+      { ...config, planDirSizeLimitBytes: Number.MAX_SAFE_INTEGER },
+      stagingRoot,
+    );
+    return await publishPlanV2FromV1(stagingRoot, publisher);
+  } catch (error) {
+    try {
+      await publisher.abort();
+    } catch {
+      // Preserve the planning/publication failure; cleanup is best-effort.
+    }
+    throw error;
+  } finally {
+    rmSync(stagingRoot, { recursive: true, force: true });
+  }
+}
+
 /**
  * Plan directly into v2. The large v1 directory exists only as local staging;
  * its historical 2 GiB transport cap is disabled because no monolithic
@@ -544,18 +624,11 @@ export async function planV2(
   if (existsSync(planV2Dir)) {
     throw new PlanV2IntegrityError(`output directory already exists: ${planV2Dir}`);
   }
-  mkdirSync(dirname(planV2Dir), { recursive: true });
-  const stagingDir = mkdtempSync(join(dirname(planV2Dir), ".plan-v2-source-"));
-  try {
-    await plan(
-      projectDir,
-      { ...config, planDirSizeLimitBytes: Number.MAX_SAFE_INTEGER },
-      stagingDir,
-    );
-    return createPlanV2FromV1(stagingDir, planV2Dir);
-  } finally {
-    rmSync(stagingDir, { recursive: true, force: true });
-  }
+  const publisher = new LocalPlanV2ArtifactPublisher(planV2Dir);
+  const manifest = await planV2WithPublisher(projectDir, config, publisher, {
+    stagingParentDir: dirname(planV2Dir),
+  });
+  return resultFromManifest(planV2Dir, manifest);
 }
 
 function resultFromManifest(planV2Dir: string, manifest: PlanV2Manifest): PlanV2Result {
@@ -747,7 +820,7 @@ export function listPlanV2ArtifactsForTarget(
 }
 
 function verifyBlob(planV2Dir: string, artifact: Readonly<PlanV2Artifact>): string {
-  const sourcePath = blobPath(planV2Dir, artifact.sha256);
+  const sourcePath = planV2BlobPath(planV2Dir, artifact.sha256);
   if (!existsSync(sourcePath)) {
     throw new PlanV2IntegrityError(`missing content-addressed artifact ${artifact.sha256}`);
   }

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

@@ -0,0 +1,17 @@
+export const PLAN_V2_INTEGRITY_UNRECOVERABLE = "PLAN_V2_INTEGRITY_UNRECOVERABLE" as const;
+
+/**
+ * A deterministic plan-v2 validation or integrity failure. Distributed
+ * adapters classify both the class name and code as terminal so immutable
+ * corruption does not consume the retry budget.
+ */
+export class PlanV2IntegrityError 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_V2_INTEGRITY_UNRECOVERABLE = PLAN_V2_INTEGRITY_UNRECOVERABLE;
+
+  constructor(message: string) {
+    super(`[planV2] ${message}`);
+    this.name = "PlanV2IntegrityError";
+  }
+}

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

@@ -0,0 +1,16 @@
+import { join } from "node:path";
+import { PlanV2IntegrityError } from "./planV2Errors.js";
+
+const SHA256_PATTERN = /^[0-9a-f]{64}$/;
+
+export function assertPlanV2Sha256(value: unknown, field: string): string {
+  if (typeof value !== "string" || !SHA256_PATTERN.test(value)) {
+    throw new PlanV2IntegrityError(`${field} must be a lowercase sha256 digest`);
+  }
+  return value;
+}
+
+export function planV2BlobPath(root: string, sha256: string): string {
+  const digest = assertPlanV2Sha256(sha256, "artifact sha256");
+  return join(root, "artifacts", "sha256", digest.slice(0, 2), digest);
+}

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

@@ -0,0 +1,137 @@
+import {
+  copyFileSync,
+  existsSync,
+  linkSync,
+  mkdirSync,
+  mkdtempSync,
+  renameSync,
+  rmSync,
+  statSync,
+  writeFileSync,
+} from "node:fs";
+import { dirname, join } from "node:path";
+import { PlanV2IntegrityError } from "./planV2Errors.js";
+import { assertPlanV2Sha256, planV2BlobPath } from "./planV2Layout.js";
+
+export interface PlanV2PublishBlob {
+  readonly sourcePath: string;
+  readonly sha256: string;
+  readonly sizeBytes: number;
+}
+
+export interface PlanV2ArtifactPublisher {
+  /**
+   * Durably publish one immutable digest before resolving. `sourcePath` is
+   * planner-local and valid only for the lifetime of this call; remote
+   * adapters must finish reading/uploading it before the promise resolves.
+   */
+  putBlob(blob: Readonly<PlanV2PublishBlob>): Promise<void>;
+  /** Commit the manifest only after every referenced blob is durable. */
+  commitManifest(manifestBytes: string): Promise<void>;
+  /** Idempotent best-effort cleanup for an unpublished or partial plan. */
+  abort(): Promise<void>;
+}
+
+function canFallbackToCopy(error: unknown): boolean {
+  if (!(error instanceof Error) || !("code" in error)) return false;
+  const code = error.code;
+  return code === "EXDEV" || code === "EPERM" || code === "EACCES" || code === "ENOTSUP";
+}
+
+function isRecord(value: unknown): value is Record<string, unknown> {
+  return typeof value === "object" && value !== null && !Array.isArray(value);
+}
+
+function manifestDigests(manifestBytes: string): readonly string[] {
+  let value: unknown;
+  try {
+    value = JSON.parse(manifestBytes);
+  } catch {
+    throw new PlanV2IntegrityError("manifest publisher received invalid JSON");
+  }
+  if (!isRecord(value) || !Array.isArray(value.artifacts)) {
+    throw new PlanV2IntegrityError("manifest publisher requires an artifacts array");
+  }
+  return value.artifacts.map((artifact, index) => {
+    if (!isRecord(artifact)) {
+      throw new PlanV2IntegrityError(`manifest artifacts[${index}] must be an object`);
+    }
+    return assertPlanV2Sha256(artifact.sha256, `manifest artifacts[${index}].sha256`);
+  });
+}
+
+export interface LocalPlanV2ArtifactPublisherOptions {
+  /** Test seam for exercising cross-device/restricted-filesystem fallback. */
+  readonly linkFile?: typeof linkSync;
+}
+
+/**
+ * Planner-local manifest-last publisher.
+ *
+ * Same-filesystem blobs are hard-linked from the frozen source plan, so the
+ * staging tree and local CAS do not consume a second set of data blocks.
+ * Cross-device and restricted filesystems fall back to an atomic copy. This
+ * implementation is a compatibility adapter for local callers; distributed
+ * nodes exchange remote manifest/CAS locators through their cloud adapters.
+ */
+export class LocalPlanV2ArtifactPublisher implements PlanV2ArtifactPublisher {
+  readonly destinationDir: string;
+  readonly temporaryDir: string;
+  readonly #linkFile: typeof linkSync;
+  #committed = false;
+
+  constructor(destinationDir: string, options: Readonly<LocalPlanV2ArtifactPublisherOptions> = {}) {
+    if (existsSync(destinationDir)) {
+      throw new PlanV2IntegrityError(`output directory already exists: ${destinationDir}`);
+    }
+    this.destinationDir = destinationDir;
+    this.#linkFile = options.linkFile ?? linkSync;
+    mkdirSync(dirname(destinationDir), { recursive: true });
+    this.temporaryDir = mkdtempSync(join(dirname(destinationDir), ".plan-v2-publish-"));
+  }
+
+  async putBlob(blob: Readonly<PlanV2PublishBlob>): Promise<void> {
+    const digest = assertPlanV2Sha256(blob.sha256, "published blob sha256");
+    const sourceSize = statSync(blob.sourcePath).size;
+    if (sourceSize !== blob.sizeBytes) {
+      throw new PlanV2IntegrityError(
+        `published blob size changed for ${digest}: expected ${blob.sizeBytes}, got ${sourceSize}`,
+      );
+    }
+    const destinationPath = planV2BlobPath(this.temporaryDir, digest);
+    if (existsSync(destinationPath)) return;
+    mkdirSync(dirname(destinationPath), { recursive: true });
+    const stagingDir = mkdtempSync(join(dirname(destinationPath), ".plan-v2-blob-"));
+    const temporaryPath = join(stagingDir, "blob");
+    try {
+      try {
+        this.#linkFile(blob.sourcePath, temporaryPath);
+      } catch (error) {
+        if (!canFallbackToCopy(error)) throw error;
+        copyFileSync(blob.sourcePath, temporaryPath);
+      }
+      renameSync(temporaryPath, destinationPath);
+    } finally {
+      rmSync(stagingDir, { recursive: true, force: true });
+    }
+  }
+
+  async commitManifest(manifestBytes: string): Promise<void> {
+    for (const digest of new Set(manifestDigests(manifestBytes))) {
+      if (!existsSync(planV2BlobPath(this.temporaryDir, digest))) {
+        throw new PlanV2IntegrityError(
+          `cannot commit manifest before referenced blob is durable: ${digest}`,
+        );
+      }
+    }
+    writeFileSync(join(this.temporaryDir, "plan.json"), manifestBytes, "utf-8");
+    renameSync(this.temporaryDir, this.destinationDir);
+    this.#committed = true;
+  }
+
+  async abort(): Promise<void> {
+    if (!this.#committed) {
+      rmSync(this.temporaryDir, { recursive: true, force: true });
+    }
+  }
+}

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

@@ -98,6 +98,9 @@ describe("@hyperframes/producer/distributed (subpath)", () => {
     expect(typeof distributedSubpath.getDistributedRenderCapabilities).toBe("function");
     expect(typeof distributedSubpath.readPlanProtocol).toBe("function");
     expect(typeof distributedSubpath.planV2).toBe("function");
+    expect(typeof distributedSubpath.planV2WithPublisher).toBe("function");
+    expect(typeof distributedSubpath.publishPlanV2FromV1).toBe("function");
+    expect(typeof distributedSubpath.LocalPlanV2ArtifactPublisher).toBe("function");
     expect(typeof distributedSubpath.renderChunkV2).toBe("function");
     expect(typeof distributedSubpath.assembleV2).toBe("function");
     expect(typeof distributedSubpath.readPlanV2Manifest).toBe("function");
@@ -122,6 +125,8 @@ describe("@hyperframes/producer (main entry)", () => {
     expect(producerIndex.PLAN_PROTOCOL_UNSUPPORTED).toBe("PLAN_PROTOCOL_UNSUPPORTED");
     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.publishPlanV2FromV1).toBe("function");
     expect(typeof producerIndex.PlanV2IntegrityError).toBe("function");
     expect(typeof producerIndex.PlanProtocolUnsupportedError).toBe("function");
   });