fix(webapp): address CodeRabbit review on streaming batch ingest

- Enforce positive STREAMING_BATCH_INGEST_CONCURRENCY in the env schema
  (.int().positive()) — p-map requires concurrency >= 1, so 0/negative would
  throw at runtime.
- Apply the same out-of-range index guard to oversized-item markers as normal
  items, so an oversized item with index >= runCount returns a 4xx instead of
  creating a stray pre-failed run. Covered by a new test.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: matt-aitken

apps/webapp/app/env.server.ts

@@ -677,7 +677,7 @@ const EnvironmentSchema = z
     // Number of streamed batch items ingested concurrently in Phase 2. Peak
     // in-flight memory per request ≈ this × STREAMING_BATCH_ITEM_MAXIMUM_SIZE,
     // so raise with care. Set to 1 for fully sequential ingestion.
-    STREAMING_BATCH_INGEST_CONCURRENCY: z.coerce.number().int().default(10),
+    STREAMING_BATCH_INGEST_CONCURRENCY: z.coerce.number().int().positive().default(10),
     BATCH_RATE_LIMIT_REFILL_RATE: z.coerce.number().int().default(100),
     BATCH_RATE_LIMIT_MAX: z.coerce.number().int().default(1200),
     BATCH_RATE_LIMIT_REFILL_INTERVAL: z.string().default("10s"),

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

@@ -406,6 +406,14 @@ export class StreamBatchItemsService extends WithRunEngine {
     if (rawItem && typeof rawItem === "object" && "__batchItemError" in rawItem) {
       const marker = rawItem as OversizedItemMarker;
 
+      // Same out-of-range guard as normal items: an oversized item with an
+      // out-of-range index must 4xx rather than create a stray pre-failed run.
+      if (marker.index >= runCount) {
+        throw new ServiceValidationError(
+          `Item index ${marker.index} exceeds batch runCount ${runCount}`
+        );
+      }
+
       const errorMessage = `Batch item payload is too large (${(marker.actualSize / 1024).toFixed(
         1
       )} KB). Maximum allowed size is ${(marker.maxSize / 1024).toFixed(

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

@@ -1753,6 +1753,54 @@ describe("StreamBatchItemsService", () => {
       await engine.quit();
     }
   );
+
+  containerTest(
+    "rejects an oversized item whose index exceeds runCount",
+    async ({ prisma, redisOptions }) => {
+      const engine = buildEngine(prisma, redisOptions);
+      const environment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
+
+      const batch = await createBatch(prisma, environment.id, {
+        runCount: 2,
+        status: "PENDING",
+        sealed: false,
+      });
+
+      await engine.initializeBatch({
+        batchId: batch.id,
+        friendlyId: batch.friendlyId,
+        environmentId: environment.id,
+        environmentType: environment.type,
+        organizationId: environment.organizationId,
+        projectId: environment.projectId,
+        runCount: 2,
+        processingConcurrency: 10,
+      });
+
+      const service = new StreamBatchItemsService({ prisma, engine });
+
+      // An oversized marker must hit the same out-of-range guard as a normal
+      // item rather than slipping through to create a stray pre-failed run.
+      async function* oversizedOutOfRange() {
+        yield {
+          __batchItemError: "OVERSIZED",
+          index: 5,
+          task: "test-task",
+          actualSize: 9999,
+          maxSize: 1000,
+        };
+      }
+
+      await expect(
+        service.call(environment, batch.friendlyId, oversizedOutOfRange(), {
+          maxItemBytes: 1024,
+          concurrency: 4,
+        })
+      ).rejects.toThrow(ServiceValidationError);
+
+      await engine.quit();
+    }
+  );
 });
 
 describe("createNdjsonParserStream", () => {