fix(guardrails): chunk + time-bound internal PII mask requests

- chunk maskPIIBatchViaHttp by count (2000) and bytes (256KB) so large
  executions split across requests and never hit the contract's 100k cap
- add AbortSignal.timeout(45s) per request so a slow/unreachable app container
  aborts and the caller scrubs, instead of hanging the trigger.dev job
- catch maskPIIBatch failures in the route: log and return a structured 500
  (broken venv fails loudly server-side; caller still scrubs, no leak)
- add mask-client tests (order across chunks, count split, non-2xx, empty)

Author: TheodoreSpeaks

apps/sim/app/api/guardrails/mask-batch/route.ts

@@ -1,4 +1,5 @@
 import { createLogger } from '@sim/logger'
+import { getErrorMessage } from '@sim/utils/errors'
 import { type NextRequest, NextResponse } from 'next/server'
 import { guardrailsMaskBatchContract } from '@/lib/api/contracts'
 import { parseRequest } from '@/lib/api/server'
@@ -25,7 +26,20 @@ export const POST = withRouteHandler(async (request: NextRequest) => {
 
   const { texts, entityTypes, language } = parsed.data.body
 
-  const masked = await maskPIIBatch(texts, entityTypes, language)
-  logger.info('Masked PII batch', { count: texts.length })
-  return NextResponse.json({ masked })
+  try {
+    const masked = await maskPIIBatch(texts, entityTypes, language)
+    logger.info('Masked PII batch', { count: texts.length })
+    return NextResponse.json({ masked })
+  } catch (error) {
+    // A broken/absent venv makes maskPIIBatch throw; fail loudly here (the
+    // caller scrubs to REDACTION_FAILED, so PII is never leaked).
+    logger.error('PII batch masking failed', {
+      error: getErrorMessage(error),
+      count: texts.length,
+    })
+    return NextResponse.json(
+      { error: getErrorMessage(error, 'PII masking failed') },
+      { status: 500 }
+    )
+  }
 })

apps/sim/lib/guardrails/mask-client.test.ts

@@ -0,0 +1,68 @@
+/**
+ * @vitest-environment node
+ */
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+
+const { mockToken, mockBaseUrl } = vi.hoisted(() => ({
+  mockToken: vi.fn(),
+  mockBaseUrl: vi.fn(),
+}))
+
+vi.mock('@/lib/auth/internal', () => ({ generateInternalToken: mockToken }))
+vi.mock('@/lib/core/utils/urls', () => ({ getInternalApiBaseUrl: mockBaseUrl }))
+
+import { maskPIIBatchViaHttp } from '@/lib/guardrails/mask-client'
+
+describe('maskPIIBatchViaHttp', () => {
+  let fetchMock: ReturnType<typeof vi.fn>
+
+  beforeEach(() => {
+    vi.clearAllMocks()
+    mockToken.mockResolvedValue('tok')
+    mockBaseUrl.mockReturnValue('http://app.internal:3000')
+    fetchMock = vi.fn(async (_url: string, init: { body: string }) => {
+      const { texts } = JSON.parse(init.body) as { texts: string[] }
+      return new Response(JSON.stringify({ masked: texts.map((t) => `M(${t})`) }), {
+        status: 200,
+        headers: { 'content-type': 'application/json' },
+      })
+    })
+    vi.stubGlobal('fetch', fetchMock)
+  })
+
+  afterEach(() => {
+    vi.unstubAllGlobals()
+  })
+
+  it('masks a small batch in a single request, with an abort timeout', async () => {
+    const out = await maskPIIBatchViaHttp(['a', 'b', 'c'], ['EMAIL_ADDRESS'])
+
+    expect(out).toEqual(['M(a)', 'M(b)', 'M(c)'])
+    expect(fetchMock).toHaveBeenCalledTimes(1)
+    expect(fetchMock.mock.calls[0][1].signal).toBeInstanceOf(AbortSignal)
+  })
+
+  it('splits by count into multiple requests, preserving global order', async () => {
+    const texts = Array.from({ length: 5000 }, (_, i) => `t${i}`)
+
+    const out = await maskPIIBatchViaHttp(texts, [])
+
+    expect(out).toHaveLength(5000)
+    expect(out[0]).toBe('M(t0)')
+    expect(out[4999]).toBe('M(t4999)')
+    expect(fetchMock).toHaveBeenCalledTimes(3) // 2000-per-request cap
+  })
+
+  it('throws on a non-2xx response so the caller can scrub', async () => {
+    fetchMock.mockResolvedValueOnce(new Response('boom', { status: 500 }))
+
+    await expect(maskPIIBatchViaHttp(['a'], [])).rejects.toThrow(/mask-batch request failed/)
+  })
+
+  it('returns [] without any request for empty input', async () => {
+    const out = await maskPIIBatchViaHttp([], [])
+
+    expect(out).toEqual([])
+    expect(fetchMock).not.toHaveBeenCalled()
+  })
+})

apps/sim/lib/guardrails/mask-client.ts

@@ -2,25 +2,77 @@ import type { GuardrailsMaskBatchResult } from '@/lib/api/contracts'
 import { generateInternalToken } from '@/lib/auth/internal'
 import { getInternalApiBaseUrl } from '@/lib/core/utils/urls'
 
+/**
+ * Per-request limits. A chunk is flushed when it hits either bound, keeping each
+ * request small enough for one short Presidio pass under a tight timeout and far
+ * below the contract's 100k-entry cap — so large executions split across
+ * requests instead of failing validation.
+ */
+const REQUEST_MAX_BYTES = 256 * 1024
+const REQUEST_MAX_COUNT = 2_000
+/** Slightly above the 30s Python subprocess timeout so a hung app container aborts gracefully. */
+const REQUEST_TIMEOUT_MS = 45_000
+
 /**
  * Mask PII across many strings via the internal app-container endpoint.
  *
  * Presidio (a Python venv) only exists in the app container, but the
  * log-redaction persist path also runs inside the trigger.dev runtime — so
  * redaction always routes through HTTP, the same way the guardrails tool does.
- * Order is preserved: the returned array matches `texts` length.
+ * Strings are grouped into byte/count-budgeted chunks; order is preserved, so
+ * the returned array matches `texts` length.
  *
- * Rejects on any non-2xx or shape mismatch so the caller can apply its own
- * fail-safe (scrubbing rather than leaking).
+ * Rejects on any non-2xx, timeout, or shape mismatch so the caller can apply
+ * its own fail-safe (scrubbing rather than leaking).
  */
 export async function maskPIIBatchViaHttp(
   texts: string[],
   entityTypes: string[],
   language?: string
 ): Promise<string[]> {
+  if (texts.length === 0) return []
+
   const token = await generateInternalToken()
   const url = `${getInternalApiBaseUrl()}/api/guardrails/mask-batch`
 
+  const masked: string[] = []
+  let batch: string[] = []
+  let batchBytes = 0
+
+  const flush = async () => {
+    if (batch.length === 0) return
+    const out = await postChunk(url, token, batch, entityTypes, language)
+    if (out.length !== batch.length) {
+      throw new Error('PII mask-batch returned an unexpected result')
+    }
+    for (const item of out) masked.push(item)
+    batch = []
+    batchBytes = 0
+  }
+
+  for (const text of texts) {
+    const bytes = Buffer.byteLength(text, 'utf8')
+    if (
+      batch.length > 0 &&
+      (batch.length >= REQUEST_MAX_COUNT || batchBytes + bytes > REQUEST_MAX_BYTES)
+    ) {
+      await flush()
+    }
+    batch.push(text)
+    batchBytes += bytes
+  }
+  await flush()
+
+  return masked
+}
+
+async function postChunk(
+  url: string,
+  token: string,
+  texts: string[],
+  entityTypes: string[],
+  language: string | undefined
+): Promise<string[]> {
   // boundary-raw-fetch: internal server-to-server call to the app container (internal JWT auth, configurable base URL)
   const response = await fetch(url, {
     method: 'POST',
@@ -29,6 +81,7 @@ export async function maskPIIBatchViaHttp(
       authorization: `Bearer ${token}`,
     },
     body: JSON.stringify({ texts, entityTypes, language }),
+    signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS),
   })
 
   if (!response.ok) {
@@ -37,7 +90,7 @@ export async function maskPIIBatchViaHttp(
   }
 
   const data = (await response.json()) as GuardrailsMaskBatchResult
-  if (!Array.isArray(data.masked) || data.masked.length !== texts.length) {
+  if (!Array.isArray(data.masked)) {
     throw new Error('PII mask-batch returned an unexpected result')
   }
   return data.masked