feat(producer): add services/distributed/assemble.ts

[jrusso1020]

What

Phase 3 of the distributed rendering plan: the third public primitive. Adds assemble(planDir, chunkPaths, audioPath, outputPath) and its supporting types at packages/producer/src/services/distributed/assemble.ts. See DISTRIBUTED-RENDERING-PLAN.md §11 Phase 3.

Why

plan() (#808) and renderChunk() (#809) produce the planDir and the per-chunk outputs respectively. assemble() is what every distributed fan-out workflow runs last: it stitches the chunks into the final deliverable. Without it, the planDir → chunks chain stops at a list of files; nothing produces the user-facing mp4/mov/png-sequence.

How

assemble() branches on the planDir's encoder format:

mp4 / mov: ffmpeg -f concat -c copy over the ordered chunk paths. Each chunk's first frame is an IDR keyframe (PR 3.2 set lockGopForChunkConcat: true), so concat-copy round-trips losslessly. The concatenated output is then:

1. Passed through padOrTrimAudioToVideoFrameCount (PR 2.7 surface) when audioPath is non-null, so audio length is exactly frameCount / fps rather than the audio mixer's original-duration output.
2. Muxed with the normalized audio via the engine's muxVideoWithAudio (same helper the in-process renderer's assembleStage uses).
3. Passed through applyFaststart so the moov atom moves to the file's start.

When no audio is present, the concat output skips mux and goes straight to applyFaststart.

png-sequence: chunks are directories of frame_NNNNNN.png files numbered locally per chunk. assemble() merges them with a continuous global index so chunk 0's frame_000000.png lands at frame_000001.png in the output, chunk 1's first frame becomes frame_(N+1), etc. When audioPath is non-null we copy it alongside as audio.aac so callers who need to re-mux later have it.

Validation

Both branches assert chunkPaths.length === chunks.length (the value read from meta/chunks.json) and that each chunk path exists. A missing or mismatched manifest trips a typed error before any ffmpeg invocation.

What did NOT change

No engine helpers, no Phase 1 stages, no in-process orchestrator. assemble() reuses muxVideoWithAudio / applyFaststart / runFfmpeg / padOrTrimAudioToVideoFrameCount exactly as they exist — the in-process runAssembleStage is intentionally not called because it operates on a RenderJob and emits updateJobStatus payloads, neither of which the distributed activity has.

Test plan

• Unit tests added — packages/producer/src/services/distributed/assemble.test.ts. 5 cases:
  • Concat-copies two mp4 chunks and applies faststart (ffprobe asserts codec, frame count, atom order).
  • Muxes audio with frame-count-derived duration when audio.aac is present (ffprobe asserts audio duration within 50ms of totalFrames / fps).
  • Merges png-sequence chunk directories with continuous global numbering (asserts filenames frame_000001..frame_000007).
  • Rejects mismatched chunkPaths.length vs chunks.json.length.
  • Rejects a planDir missing plan.json.
• bun test packages/producer/src/services/distributed/ — 18 pass (PRs 3.1 + 3.2 + 3.3).
• bun run --filter @hyperframes/producer typecheck — clean.
• bunx oxlint + bunx oxfmt --check — clean on changed files.
• Producer Docker regression harness — pending CI. executeRenderJob is unchanged; PSNR baselines should hold.

The mp4 fixture pre-renders test inputs via raw ffmpeg (testsrc filter + closed-GOP libx264) rather than going through the Chrome capture pipeline. This isolates concat-copy + mux + faststart from the renderChunk path that PRs 3.1/3.2 already cover, and avoids the chrome-headless-shell smoke-test gating that PR 3.2 needed.

This is PR 3 of a 6-PR Phase 3 stack:

• 3.1 — services/distributed/plan.ts (feat(producer): add services/distributed/plan.ts #808)
• 3.2 — services/distributed/renderChunk.ts (feat(producer): add services/distributed/renderChunk.ts #809)
• 3.3 (this PR) — services/distributed/assemble.ts
• 3.4 — planDir size cap (PLAN_TOO_LARGE)
• 3.5 — distributed format banlist (webm + HDR mp4)
• 3.6 — public exports + @hyperframes/producer/distributed subpath

🤖 Generated with Claude Code

[jrusso1020]

• feat(producer): public exports for distributed render primitives #816
• feat(producer): refuse distributed-unsupported formats (webm + HDR mp4) #815
• feat(producer): enforce planDir size cap (PLAN_TOO_LARGE) #814
• feat(producer): add services/distributed/assemble.ts #813 👈 (View in Graphite)
• feat(producer): add services/distributed/renderChunk.ts #809
• feat(producer): add services/distributed/plan.ts #808
• main

This stack of pull requests is managed by Graphite. Learn more about stacking.

[miguel-heygen]

Review: assemble.ts — Distributed Chunk Assembly

Clean implementation. Successfully avoids the re-render bottleneck — audio arrives pre-rendered via audioPath, padOrTrim normalizes to video length, mux stream-copies video + re-encodes audio at 192k AAC. Concat demuxer usage is textbook.

Test coverage is comprehensive: mp4 concat + faststart byte-level verification, audio mux with duration assertion (50ms tolerance for AAC frame quantization), png-sequence merge with global renumbering, and two rejection cases.

Notes

1. Silent audio skip (~line 104-106): When audioPath is non-null but the file doesn't exist, it silently skips audio. This could hide an upstream audio render failure. Consider at minimum logging a warning.

2. Double faststart for mp4 with audio: muxVideoWithAudio adds +faststart internally, then assemble.ts:186 calls applyFaststart again. Idempotent but wastes one ffmpeg invocation. Consider skipping applyFaststart when audio was muxed.

3. console.warn vs structured log: mergePngFrameDirs frame count mismatch uses console.warn instead of the structured log parameter used elsewhere in the module.

All minor — none blocking. LGTM.

[vanceingalls]

Phase 3 Activity C — assemble stitches chunks into the final deliverable. Concat-copy on mp4/mov is the right call given the chunk encoder's lockGopForChunkConcat: true + gopSize === framesInChunk from #809, and the png-sequence directory merge with global re-numbering is clean. A few real concerns around audio handling and recoverability.

Strengths

• assemble.ts:500 — path.replace(/'/g, "'\\''") is the correct ffmpeg-concat-demuxer single-quote escape. Spelled out so it survives the next refactor.
• assemble.ts:526-543 — running padOrTrimAudioToVideoFrameCount BEFORE mux is the right ordering. Audio drift at the end of long renders is exactly the failure mode the helper exists to catch, and feeding the concat output's actual frame count (rather than plan.totalFrames) keeps the normalization grounded in what was actually rendered.
• assemble.ts:599-647 — the png-sequence merge does global re-numbering across chunk boundaries (globalIdx + 1) so consumers see one continuous sequence rather than per-chunk numbering. Matches the in-process renderer's convention.

Findings

important — assemble accepts a null audioPath even when plan.hasAudio === true (assemble.ts:447-543)

The caller passes audioPath: string | null. If the composition has audio (plan.hasAudio === true) but the caller passes null, the function silently skips the mux and ships a video with no audio track — no warning, no error, no diagnostic. The mismatch is exactly the kind of contract violation assemble should catch at its trust boundary:

if (plan.hasAudio && audioPath === null) {
  throw new Error("[assemble] planDir has audio but caller passed audioPath=null");
}
if (!plan.hasAudio && audioPath !== null) {
  // Warn at minimum — caller is passing audio for a no-audio plan,
  // which suggests a stale planDir / mismatched caller config.
}

Without this, a workflow adapter bug (e.g., forgetting to download audio.aac from S3 before calling assemble) ships a silently-broken deliverable. The five-test coverage in assemble.test.ts doesn't exercise this case.

important — mergePngFrameDirs warns on frame-count mismatch but proceeds (assemble.ts:629-639)

if (globalIdx !== totalFrames) {
  console.warn(
    `[assemble] png-sequence frame count mismatch: merged ${globalIdx} frames vs ` +
      `plan.totalFrames=${totalFrames}. Using on-disk count.`,
  );
}

The comment says "consumers should rely on the on-disk count" but the result struct returns framesEncoded: globalIdx (the on-disk count). So the structured return is correct but the console.warn is unstructured — it won't show up in workflow adapter logs, won't surface to the user, won't trip the workflow's failure path.

The justification ("can differ by ±1 across hosts in edge cases") is fair for small drifts, but ±1 is also exactly the value that masks "one of the chunks silently produced N-1 frames because the encoder choked." The drift threshold matters. Two cleaner shapes:

1. Tight: throw on any mismatch. The byte-identical-retry contract from #809 makes this safe — if a chunk produces the wrong frame count it's deterministic, so the mismatch is a real bug worth surfacing.
2. Tolerant: allow Math.abs(globalIdx - totalFrames) <= 1, throw otherwise. Document the ±1 tolerance and which paths produce it.

I'd take (1) and let real edge cases surface via test failures so the cause can be diagnosed instead of buried in a warning.

important — the mp4/mov finally block on line 568-577 doesn't clean up outputPath on partial failure

The finally removes workDir (good), but if the faststart step fails after the concat + mux succeeded, outputPath is left as either a partial file (if applyFaststart was mid-write) or the mux output (if faststart hadn't started). Callers reading outputPath after a failed assemble() get garbage they think is valid. Either:

• Atomic write: faststart to a .tmp path, rename to outputPath on success.
• Cleanup outputPath in the same finally when the outer try throws.

The "callers handle this" answer is fine if documented; it's not currently documented.

nit — assemble.ts:566 — applyFaststart is documented as "no-op for .mov" but the chained call still writes through

Comment at line 562: "applyFaststart is a no-op for .mov (it copies the input to output)". A copy is not free — it's a full read of muxOutputPath and a full write of outputPath. For a 4K SDR 1-minute mov at ProRes 4444 that's ~10 GB. If there's no functional reason to copy on .mov, skip the call and renameSync(muxOutputPath, outputPath) instead — same outcome at no cost.

nit — assemble doesn't verify chunkPaths point at outputs from THIS planDir

assemble.ts:476-480 checks paths exist but doesn't check the perf sidecars (which carry planHash) match plan.planHash. A caller pointing at chunks from a different planDir produces undefined output. Read the perf sidecars and assert planHash matches; cheap (it's already on disk per #809).

Verdict: APPROVE
Reasoning: The concat-copy + mux + faststart sequence is correct and well-grounded in the chunk encoder's GOP discipline. The findings above are real but recoverable as follow-ups — none corrupt the happy-path output, they just leave silent-failure surfaces around it. Worth landing now and fixing the trust-boundary gaps in a focused follow-up.

— Vai

[vanceingalls]

Re-review

PR rebased onto the new #808/#809 bases (which shifted the shared.ts helpers and added the env-restore + planHash-validation contracts to renderChunk). The assemble.ts content itself didn't change between SHAs 5f0239a → 24f0ffea; the diff is identical to what I previously approved.

Net-new findings

Note (catch from a fresh read — not blocking): assemble.ts:491 — const workDir = \${outputPath}.assemble-work`;has the same race I flagged onrenderChunk(concurrent invocations on the sameoutputPathwouldrmSynceach other's work). Assemble is controller-side so the realistic risk is lower than forrenderChunk workers, but the same suffix pattern (pid + randomBytes`) would be cheap to apply if the controller ever fan-outs more than once. Nit, not gating — flagging for parity with the renderChunk fix.

Verdict: APPROVE
Reasoning: No content change since prior approval; prior findings non-existent. Catch on the workDir suffix is a nit, easily a follow-up. CI required checks green.

— Vai

The base branch was changed.

[miguel-heygen]

Re-stamping after rebase. LGTM.

[jrusso1020]

Merge activity

• May 14, 2:38 AM UTC: @jrusso1020 merged this pull request with Graphite.