fix(webapp): treat Phase 2 batch-stream retries as idempotent (TRI-9944)

When the SDK created a batch and then streamed its items (Phase 2 of
the 2-phase batch API), a lost response would trigger the SDK's
network-retry path. For small, fast-completing batches the original
request had already enqueued every item, sealed the batch, and the
runs flipped the batch to PROCESSING or COMPLETED by the time the
retry arrived. The retry then failed the pre-loop check at
streamBatchItems.server.ts:109 with a 422 — surfacing a
customer-visible BatchTriggerError for a batch whose runs had
actually succeeded.

StreamBatchItemsService.call now returns the standard sealed:true
success response (itemsAccepted: 0, itemsDeduplicated: 0,
runCount: batch.runCount) when the batch is already sealed or in
PROCESSING/COMPLETED, matching the idempotency already applied at
the two post-loop race-condition branches in the same file (lines
226 and 306). ABORTED and other unexpected non-PENDING states still
throw.

Tests:
- Rewrote the existing "already sealed" race test from expecting a
  throw to expecting sealed:true (Phase 2 retry idempotency).
- Added a COMPLETED-pre-loop test mirroring the exact customer
  scenario (single-item batch, status=COMPLETED, sealed=false —
  tryCompleteBatch sets status without setting sealed).
- Added a negative ABORTED test to lock in that terminal-failure
  states still surface as errors.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: matt-aitken

.server-changes/batch-stream-phase2-retry-idempotency.md

@@ -0,0 +1,24 @@
+---
+area: webapp
+type: fix
+---
+
+Treat Phase 2 batch-stream retries as idempotent when the batch has
+already been sealed or moved past `PENDING` (TRI-9944).
+
+When the SDK created a batch and then streamed its items (Phase 2 of
+the 2-phase batch API), a lost response would trigger the SDK's
+network-retry path. For small, fast-completing batches the original
+request had already enqueued every item, sealed the batch, and let the
+runs flip the batch to `PROCESSING` or even `COMPLETED` by the time the
+retry arrived. The retry then failed the pre-loop check at
+`apps/webapp/app/runEngine/services/streamBatchItems.server.ts:109`
+with a 422 — surfacing a customer-visible `BatchTriggerError` for a
+batch whose runs had actually succeeded.
+
+`StreamBatchItemsService.call` now returns the standard `sealed: true`
+success response (with `itemsAccepted: 0`, `itemsDeduplicated: 0`,
+`runCount: batch.runCount`) when the batch is already sealed or in
+`PROCESSING`/`COMPLETED`, matching the idempotency already applied at
+the two post-loop race-condition branches in the same file.
+`ABORTED` and other unexpected non-`PENDING` states still throw.

apps/webapp/app/runEngine/services/streamBatchItems.server.ts

@@ -100,13 +100,29 @@ export class StreamBatchItemsService extends WithRunEngine {
           throw new ServiceValidationError(`Batch ${batchFriendlyId} not found`);
         }
 
-        if (batch.sealed) {
-          throw new ServiceValidationError(
-            `Batch ${batchFriendlyId} is already sealed and cannot accept more items`
-          );
+        // Phase 2 retry idempotency (TRI-9944): if the batch is already sealed
+        // or has moved past PENDING into PROCESSING/COMPLETED, this is a retry
+        // of a request whose response was lost — the original successful request
+        // already enqueued every item and sealed the batch. Returning sealed:true
+        // makes the SDK stop retrying instead of throwing a customer-visible 422.
+        if (batch.sealed || batch.status === "PROCESSING" || batch.status === "COMPLETED") {
+          logger.info("Batch already sealed/completed - treating Phase 2 retry as success", {
+            batchId: batchFriendlyId,
+            batchSealed: batch.sealed,
+            batchStatus: batch.status,
+          });
+
+          return {
+            id: batchFriendlyId,
+            itemsAccepted: 0,
+            itemsDeduplicated: 0,
+            sealed: true,
+            runCount: batch.runCount,
+          };
         }
 
         if (batch.status !== "PENDING") {
+          // ABORTED or any other unexpected non-PENDING state — surface as an error.
           throw new ServiceValidationError(
             `Batch ${batchFriendlyId} is not in PENDING status (current: ${batch.status})`
           );

apps/webapp/test/engine/streamBatchItems.test.ts

@@ -174,7 +174,7 @@ describe("StreamBatchItemsService", () => {
   );
 
   containerTest(
-    "should handle race condition when batch already sealed by another request",
+    "should return sealed=true when batch is already sealed and PROCESSING (Phase 2 retry idempotency)",
     async ({ prisma, redisOptions }) => {
       const engine = new RunEngine({
         prisma,
@@ -211,43 +211,160 @@ describe("StreamBatchItemsService", () => {
 
       const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
 
-      // Create a batch that is already sealed and PROCESSING (simulating another request won the race)
+      // Simulate the SDK retrying Phase 2 after the original request succeeded:
+      // the original request already sealed the batch and moved it to PROCESSING.
       const batch = await createBatch(prisma, authenticatedEnvironment.id, {
         runCount: 2,
         status: "PROCESSING",
         sealed: true,
       });
 
-      // Initialize the batch in Redis with full count
-      await engine.initializeBatch({
-        batchId: batch.id,
-        friendlyId: batch.friendlyId,
-        environmentId: authenticatedEnvironment.id,
-        environmentType: authenticatedEnvironment.type,
-        organizationId: authenticatedEnvironment.organizationId,
-        projectId: authenticatedEnvironment.projectId,
-        runCount: 2,
-        processingConcurrency: 10,
+      const service = new StreamBatchItemsService({
+        prisma,
+        engine,
       });
 
-      // Enqueue items directly
-      await engine.enqueueBatchItem(batch.id, authenticatedEnvironment.id, 0, {
-        task: "test-task",
-        payload: JSON.stringify({ data: "item1" }),
-        payloadType: "application/json",
+      const result = await service.call(
+        authenticatedEnvironment,
+        batch.friendlyId,
+        itemsToAsyncIterable([]),
+        {
+          maxItemBytes: 1024 * 1024,
+        }
+      );
+
+      // The retry should be treated as success — the original request already
+      // enqueued every item, so the SDK should stop retrying.
+      expect(result.sealed).toBe(true);
+      expect(result.id).toBe(batch.friendlyId);
+      expect(result.itemsAccepted).toBe(0);
+      expect(result.itemsDeduplicated).toBe(0);
+      expect(result.runCount).toBe(2);
+
+      await engine.quit();
+    }
+  );
+
+  containerTest(
+    "should return sealed=true when batch is COMPLETED before Phase 2 retry arrives (TRI-9944)",
+    async ({ prisma, redisOptions }) => {
+      const engine = new RunEngine({
+        prisma,
+        worker: {
+          redis: redisOptions,
+          workers: 1,
+          tasksPerWorker: 10,
+          pollIntervalMs: 100,
+          disabled: true,
+        },
+        queue: {
+          redis: redisOptions,
+        },
+        runLock: {
+          redis: redisOptions,
+        },
+        machines: {
+          defaultMachine: "small-1x",
+          machines: {
+            "small-1x": {
+              name: "small-1x" as const,
+              cpu: 0.5,
+              memory: 0.5,
+              centsPerMs: 0.0001,
+            },
+          },
+          baseCostInCents: 0.0005,
+        },
+        batchQueue: {
+          redis: redisOptions,
+        },
+        tracer: trace.getTracer("test", "0.0.0"),
       });
-      await engine.enqueueBatchItem(batch.id, authenticatedEnvironment.id, 1, {
-        task: "test-task",
-        payload: JSON.stringify({ data: "item2" }),
-        payloadType: "application/json",
+
+      const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
+
+      // The customer-reported scenario: single-item batch where the original
+      // Phase 2 request succeeded server-side, the run executed fast, the batch
+      // flipped to COMPLETED, then the lost-response SDK retry hits us.
+      // Note: tryCompleteBatch sets status=COMPLETED but does NOT set sealed=true.
+      const batch = await createBatch(prisma, authenticatedEnvironment.id, {
+        runCount: 1,
+        status: "COMPLETED",
+        sealed: false,
+      });
+
+      const service = new StreamBatchItemsService({
+        prisma,
+        engine,
+      });
+
+      const result = await service.call(
+        authenticatedEnvironment,
+        batch.friendlyId,
+        itemsToAsyncIterable([]),
+        {
+          maxItemBytes: 1024 * 1024,
+        }
+      );
+
+      expect(result.sealed).toBe(true);
+      expect(result.id).toBe(batch.friendlyId);
+      expect(result.itemsAccepted).toBe(0);
+      expect(result.itemsDeduplicated).toBe(0);
+
+      await engine.quit();
+    }
+  );
+
+  containerTest(
+    "should throw when batch is in ABORTED status",
+    async ({ prisma, redisOptions }) => {
+      const engine = new RunEngine({
+        prisma,
+        worker: {
+          redis: redisOptions,
+          workers: 1,
+          tasksPerWorker: 10,
+          pollIntervalMs: 100,
+          disabled: true,
+        },
+        queue: {
+          redis: redisOptions,
+        },
+        runLock: {
+          redis: redisOptions,
+        },
+        machines: {
+          defaultMachine: "small-1x",
+          machines: {
+            "small-1x": {
+              name: "small-1x" as const,
+              cpu: 0.5,
+              memory: 0.5,
+              centsPerMs: 0.0001,
+            },
+          },
+          baseCostInCents: 0.0005,
+        },
+        batchQueue: {
+          redis: redisOptions,
+        },
+        tracer: trace.getTracer("test", "0.0.0"),
+      });
+
+      const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
+
+      const batch = await createBatch(prisma, authenticatedEnvironment.id, {
+        runCount: 2,
+        status: "ABORTED",
+        sealed: false,
       });
 
       const service = new StreamBatchItemsService({
         prisma,
         engine,
       });
 
-      // This should fail because the batch is already sealed
       await expect(
         service.call(authenticatedEnvironment, batch.friendlyId, itemsToAsyncIterable([]), {
           maxItemBytes: 1024 * 1024,