From b0e8eff4d111f5e76aa41740ad7e5799165fe981 Mon Sep 17 00:00:00 2001
From: Vance Ingalls <vance@heygen.com>
Date: Mon, 26 Oct 2026 21:35:01 +0000
Subject: [PATCH] fix(producer): detach the risky probe before closing it in
 the segmented session factory
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

With several workers, openSegmentWorkersTogether calls create() on the shared
default factory concurrently. The probe was only cleared after its close was
awaited, so a second worker still saw it and closed it again. Clear the
binding first, then close.

The reuse branch has no such window: `probeSession ?? await open()` does not
await when the probe is present, and the binding is cleared synchronously.

— Via

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 .../stages/captureSegmentedStage.test.ts      | 24 +++++++++++++++++++
 .../render/stages/captureSegmentedStage.ts    |  3 ++-
 2 files changed, 26 insertions(+), 1 deletion(-)

diff --git a/packages/producer/src/services/render/stages/captureSegmentedStage.test.ts b/packages/producer/src/services/render/stages/captureSegmentedStage.test.ts
index 359852ae46..04e1f39b1d 100644
--- a/packages/producer/src/services/render/stages/captureSegmentedStage.test.ts
+++ b/packages/producer/src/services/render/stages/captureSegmentedStage.test.ts
@@ -346,6 +346,30 @@ describe("runCaptureSegmentedStage", () => {
     expect(closeCaptureSession.mock.calls.map((c) => c[0])).toContain(probeSession);
   });
 
+  it("closes a risky probe once when several workers open sessions together", async () => {
+    closeCaptureSession.mockClear();
+    const probeSession = {
+      ...fakeStageInput({ totalFrames: 1 }).probeSession,
+      deRiskFrames: new Set([2]),
+    };
+    const result = await runCaptureSegmentedStage({
+      ...fakeStageInput({ totalFrames: 6 }),
+      probeSession: probeSession as never,
+      segmentFrames: 3,
+      workerCount: 2,
+      segmentDir: join(fixtureRoot, "risk-probe-workers"),
+      deps: {
+        spawnEncoder: mock(async () => okEncoder()),
+        captureFrame: mock(async () => ({ buffer: Buffer.alloc(1) })),
+        concat: mock(async () => ({ success: true as const })),
+        closeSession: mock(async () => {}),
+        removeFile: () => {},
+      },
+    });
+    expect(result.success).toBe(true);
+    expect(closeCaptureSession.mock.calls.filter((c) => c[0] === probeSession)).toHaveLength(1);
+  });
+
   it("skips completed segments and still concats all of them in order", async () => {
     const stableDir = join(fixtureRoot, "stable");
     const spawned: string[] = [];
diff --git a/packages/producer/src/services/render/stages/captureSegmentedStage.ts b/packages/producer/src/services/render/stages/captureSegmentedStage.ts
index a4aa740b01..2b4f073ef1 100644
--- a/packages/producer/src/services/render/stages/captureSegmentedStage.ts
+++ b/packages/producer/src/services/render/stages/captureSegmentedStage.ts
@@ -293,8 +293,9 @@ function defaultSessionFactory(input: CaptureSegmentedStageInput): SessionFactor
     create: async () => {
       const probeCannotCaptureEveryFrame = Boolean(probeSession?.deRiskFrames?.size);
       if (probeSession && probeCannotCaptureEveryFrame) {
-        await closeCaptureSession(probeSession).catch(() => {});
+        const riskyProbe = probeSession;
         probeSession = null;
+        await closeCaptureSession(riskyProbe).catch(() => {});
       }
       const session = probeSession ?? (await openSegmentedSession(input));
       if (probeSession) {