@@ -6,12 +6,15 @@
 single PII_URL serves both.
 """
 
+import json
 import logging
 import time
 from typing import Any
 
 import regex as regex_module
-from fastapi import FastAPI, HTTPException
+from fastapi import FastAPI, HTTPException, Request
+from fastapi.exceptions import RequestValidationError
+from fastapi.responses import JSONResponse
 from presidio_analyzer import (
     AnalyzerEngine,
     BatchAnalyzerEngine,
@@ -46,7 +49,7 @@
 )
 from presidio_anonymizer import AnonymizerEngine
 from presidio_anonymizer.entities import OperatorConfig
-from pydantic import BaseModel
+from pydantic import BaseModel, model_validator
 
 # Languages served. Each needs its spaCy model installed in the image; the
 # es/it/pl/fi predefined recognizers (ES_NIF, IT_FISCAL_CODE, PL_PESEL, ...)
@@ -307,6 +310,41 @@ def _analyze_many(
 
 app = FastAPI(title="Sim Presidio", docs_url=None, redoc_url=None)
 
+
+@app.exception_handler(RequestValidationError)
+async def request_validation_error(_request: Request, exc: RequestValidationError) -> JSONResponse:
+    """422 without FastAPI's echo of each offending input value, which would copy
+    request text (possibly PII) into the error."""
+    return JSONResponse(
+        status_code=422,
+        content={
+            "detail": [
+                {
+                    "loc": list(error.get("loc", ())),
+                    "msg": error.get("msg", ""),
+                    "type": error.get("type", ""),
+                }
+                for error in exc.errors()
+            ]
+        },
+    )
+
+
+class RequestModel(BaseModel):
+    """Base for every request body. JSON can carry an escaped unpaired UTF-16
+    surrogate that decodes into a str with no UTF-8 encoding; echoing it in the
+    response would fail as an opaque 500, so reject it up front as a 422."""
+
+    @model_validator(mode="before")
+    @classmethod
+    def _require_utf8_encodable(cls, data: Any) -> Any:
+        try:
+            json.dumps(data, ensure_ascii=False).encode("utf-8")
+        except UnicodeEncodeError:
+            raise ValueError("body contains an unpaired surrogate") from None
+        return data
+
+
 # Internal entity id assigned to the i-th user-supplied custom pattern. Never
 # surfaced: the anonymizer maps it back to the pattern's chosen `replacement`, and
 # callers relabel any leftover CUSTOM_<i> span to the pattern's display name.
@@ -387,7 +425,7 @@ def resolve_entities(
     return list(req_entities) + custom_entity_ids
 
 
-class AnalyzeRequest(BaseModel):
+class AnalyzeRequest(RequestModel):
     text: str
     language: str = "en"
     entities: list[str] | None = None
@@ -396,15 +434,15 @@ class AnalyzeRequest(BaseModel):
     patterns: list[CustomPattern] | None = None
 
 
-class AnalyzeBatchRequest(BaseModel):
+class AnalyzeBatchRequest(RequestModel):
     texts: list[str]
     language: str = "en"
     entities: list[str] | None = None
     score_threshold: float | None = None
     patterns: list[CustomPattern] | None = None
 
 
-class AnonymizeRequest(BaseModel):
+class AnonymizeRequest(RequestModel):
     text: str
     analyzer_results: list[dict[str, Any]] = []
     anonymizers: dict[str, dict[str, Any]] | None = None
@@ -417,14 +455,14 @@ class AnonymizeBatchItem(BaseModel):
     analyzer_results: list[dict[str, Any]] = []
 
 
-class AnonymizeBatchRequest(BaseModel):
+class AnonymizeBatchRequest(RequestModel):
     items: list[AnonymizeBatchItem] = []
     anonymizers: dict[str, dict[str, Any]] | None = None
     operators: dict[str, dict[str, Any]] | None = None
     patterns: list[CustomPattern] | None = None
 
 
-class RedactRequest(BaseModel):
+class RedactRequest(RequestModel):
     text: str
     language: str = "en"
     entities: list[str] | None = None
@@ -434,7 +472,7 @@ class RedactRequest(BaseModel):
     patterns: list[CustomPattern] | None = None
 
 
-class RedactBatchRequest(BaseModel):
+class RedactBatchRequest(RequestModel):
     texts: list[str]
     language: str = "en"
     entities: list[str] | None = None
 
@@ -1,22 +1,17 @@
-import { createMockRequest, hybridAuthMockFns } from '@sim/testing'
+import { hybridAuthMockFns } from '@sim/testing/mocks/hybrid-auth.mock'
+import { createMockRequest } from '@sim/testing/mocks/request.mock'
 import { beforeEach, describe, expect, it, vi } from 'vitest'
-
-const { mockMaskPIIBatch } = vi.hoisted(() => ({
-  mockMaskPIIBatch: vi.fn(),
-}))
+import { POST } from '@/app/api/guardrails/mask-batch/route'
 
 const mockCheckInternalAuth = hybridAuthMockFns.mockCheckInternalAuth
 
-vi.mock('@/lib/guardrails/validate_pii', () => ({
-  maskPIIBatch: mockMaskPIIBatch,
-}))
-
-import { POST } from '@/app/api/guardrails/mask-batch/route'
-
 describe('POST /api/guardrails/mask-batch', () => {
+  let presidioFetch: ReturnType<typeof vi.fn>
+
   beforeEach(() => {
     mockCheckInternalAuth.mockResolvedValue({ success: true })
-    mockMaskPIIBatch.mockImplementation(async (texts: string[]) => texts.map((t) => `M(${t})`))
+    presidioFetch = vi.fn()
+    vi.stubGlobal('fetch', presidioFetch)
   })
 
   it('returns 401 without internal auth', async () => {
@@ -30,6 +25,30 @@ describe('POST /api/guardrails/mask-batch', () => {
     )
 
     expect(res.status).toBe(401)
-    expect(mockMaskPIIBatch).not.toHaveBeenCalled()
+    expect(presidioFetch).not.toHaveBeenCalled()
+  })
+
+  it.each([
+    [
+      '422 when Presidio rejects the input, so it is never resent',
+      422,
+      () => new Response('{}', { status: 422 }),
+    ],
+    [
+      '503 when Presidio is unreachable, so it reads as an outage',
+      503,
+      () => Promise.reject(new TypeError('fetch failed')),
+    ],
+    [
+      '500 for any other Presidio failure, retried on its own',
+      500,
+      () => new Response('boom', { status: 500 }),
+    ],
+  ] as const)('answers %s', async (_name, status, presidio) => {
+    presidioFetch.mockImplementationOnce(presidio)
+
+    const res = await POST(createMockRequest('POST', { texts: ['a'], entityTypes: [] }))
+
+    expect(res.status).toBe(status)
   })
 })
@@ -5,10 +5,25 @@ import { guardrailsMaskBatchContract } from '@/lib/api/contracts'
 import { parseRequest } from '@/lib/api/server'
 import { checkInternalAuth } from '@/lib/auth/hybrid'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
-import { maskPIIBatch } from '@/lib/guardrails/validate_pii'
+import {
+  maskPIIBatch,
+  PiiServiceRejectedError,
+  PiiServiceUnavailableError,
+} from '@/lib/guardrails/validate_pii'
 
 const logger = createLogger('GuardrailsMaskBatchAPI')
 
+/**
+ * Tells the mask client how to react to a failure: 422 = Presidio rejected this
+ * input (never resend it), 503 = Presidio is down (retry, and stop sending queued
+ * chunks), 500 = anything else (retry this chunk alone).
+ */
+function statusForMaskError(error: unknown): 422 | 503 | 500 {
+  if (error instanceof PiiServiceRejectedError) return 422
+  if (error instanceof PiiServiceUnavailableError) return 503
+  return 500
+}
+
 /**
  * Internal batch PII masking. The log-redaction persist path runs in both the
  * Next.js server and the trigger.dev runtime, but only the app task reaches the
@@ -36,15 +51,13 @@ export const POST = withRouteHandler(async (request: NextRequest) => {
     })
     return NextResponse.json({ masked })
   } catch (error) {
-    // An unreachable/misconfigured Presidio service makes maskPIIBatch throw; fail
-    // loudly here (the caller scrubs to REDACTION_FAILED, so PII is never leaked).
+    // Fail loudly; the caller scrubs to REDACTION_FAILED, so PII is never leaked.
+    const status = statusForMaskError(error)
     logger.error('PII batch masking failed', {
       error: getErrorMessage(error),
       count: texts.length,
+      status,
     })
-    return NextResponse.json(
-      { error: getErrorMessage(error, 'PII masking failed') },
-      { status: 500 }
-    )
+    return NextResponse.json({ error: getErrorMessage(error, 'PII masking failed') }, { status })
   }
 })
@@ -1,4 +1,5 @@
 import { loggerMock } from '@sim/testing'
+import { maskClientMock, maskClientMockFns } from '@sim/testing/mocks/mask-client.mock'
 import { permissionCheckMock } from '@sim/testing/mocks/permission-check.mock'
 import { storageServiceMockFns } from '@sim/testing/mocks/storage-service.mock'
 import { uploadsMock } from '@sim/testing/mocks/uploads.mock'
@@ -19,6 +20,7 @@ import { ResolvedSecretTraceRegistry } from '@/executor/utils/resolved-secret-tr
 import { VariableResolver } from '@/executor/variables/resolver'
 import type { SerializedBlock, SerializedWorkflow } from '@/serializer/types'
 
+const mockMaskBatch = maskClientMockFns.mockMaskPIIBatchViaHttp
 const mockUploadFile = storageServiceMockFns.mockUploadFile
 const mockDownloadFile = storageServiceMockFns.mockDownloadFile
 
@@ -29,17 +31,11 @@ const blockExecutorBaseLogger =
   loggerMock.createLogger.mock.results[blockExecutorLoggerCallIndex]?.value
 if (!blockExecutorBaseLogger) throw new Error('BlockExecutor logger mock was not initialized')
 
-const { mockMaskBatch } = vi.hoisted(() => ({
-  mockMaskBatch: vi.fn(),
-}))
-
 vi.mock('@/ee/access-control/utils/permission-check', () => permissionCheckMock)
 
 vi.mock('@/lib/uploads', () => uploadsMock)
 
-vi.mock('@/lib/guardrails/mask-client', () => ({
-  maskPIIBatchViaHttp: mockMaskBatch,
-}))
+vi.mock('@/lib/guardrails/mask-client', () => maskClientMock)
 
 vi.mock('@/lib/logs/execution/pii-redaction', async (importOriginal) => {
   const actual = await importOriginal<typeof import('@/lib/logs/execution/pii-redaction')>()
@@ -139,9 +135,10 @@ describe('BlockExecutor', () => {
     clearLargeValueCacheForTests()
     mockUploadFile.mockClear()
     mockDownloadFile.mockResolvedValue(Buffer.from(JSON.stringify(items)))
-    mockMaskBatch.mockImplementation(async (texts: string[]) =>
-      texts.map((text) => text.replaceAll('alice@example.com', '<EMAIL_ADDRESS>'))
-    )
+    mockMaskBatch.mockImplementation(async (texts: string[]) => ({
+      masked: texts.map((text) => text.replaceAll('alice@example.com', '<EMAIL_ADDRESS>')),
+      scrubbedCount: 0,
+    }))
     const block = createBlock()
     const workflow: SerializedWorkflow = {
       version: '1',
 
@@ -0,0 +1,12 @@
+import { describe, expect, it } from 'vitest'
+import { filterForDisplay } from '@/lib/core/utils/display-filters'
+
+describe('filterForDisplay', () => {
+  it('never truncates a string inside a surrogate pair', () => {
+    const value = `${'x'.repeat(14_999)}😀${'y'.repeat(10)}`
+
+    const shown: string = filterForDisplay({ value }).value
+
+    expect(shown).toBe(`${'x'.repeat(14_999)}... [truncated 12 chars]`)
+  })
+})
@@ -1,13 +1,17 @@
+import { truncateAtCodePoint } from '@sim/utils/string'
 import { filterUserFileForDisplay, isUserFile } from '@/lib/core/utils/user-file'
 
 const MAX_STRING_LENGTH = 15000
 const MAX_DEPTH = 50
 
+/**
+ * Never cuts inside a surrogate pair: a lone half has no UTF-8 encoding, and
+ * these values flow on to PII redaction, whose service rejects such text.
+ */
 function truncateString(value: string, maxLength = MAX_STRING_LENGTH): string {
-  if (value.length <= maxLength) {
-    return value
-  }
-  return `${value.substring(0, maxLength)}... [truncated ${value.length - maxLength} chars]`
+  if (value.length <= maxLength) return value
+  const kept = truncateAtCodePoint(value, maxLength, '')
+  return `${kept}... [truncated ${value.length - kept.length} chars]`
 }
 
 function filterUserFile(data: any): any {
 
@@ -35,7 +35,7 @@ describe('maskPIIBatchViaHttp', () => {
   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, [])
+    const { masked: out } = await maskPIIBatchViaHttp(texts, [])
 
     expect(out).toHaveLength(5000)
     expect(out[0]).toBe('M(t0)')
@@ -54,9 +54,9 @@ describe('maskPIIBatchViaHttp', () => {
   it('retries a transient 5xx with backoff and then succeeds', async () => {
     fetchMock.mockResolvedValueOnce(new Response('deploying', { status: 503 }))
 
-    const out = await maskPIIBatchViaHttp(['a'], [])
+    const { masked } = await maskPIIBatchViaHttp(['a'], [])
 
-    expect(out).toEqual(['M(a)'])
+    expect(masked).toEqual(['M(a)'])
     expect(fetchMock).toHaveBeenCalledTimes(2)
     expect(mockSleep).toHaveBeenCalledTimes(1)
   })
@@ -69,6 +69,47 @@ describe('maskPIIBatchViaHttp', () => {
     expect(mockSleep).toHaveBeenCalledTimes(7)
   })
 
+  describe('with a failed-chunk placeholder', () => {
+    const PLACEHOLDER = '[FAILED]'
+    const CHUNK_SIZE = 2000
+    const CONCURRENCY = 64
+    /**
+     * Fills every concurrent slot with chunks the service fails, then queues two more
+     * chunks (`'queued'`) that it would mask if they were ever sent.
+     */
+    const failingThenQueued = () => [
+      ...Array.from({ length: CHUNK_SIZE * CONCURRENCY }, () => 'failing'),
+      ...Array.from({ length: CHUNK_SIZE + 1 }, () => 'queued'),
+    ]
+    const serviceFailingWith = (status: number) =>
+      fetchMock.mockImplementation(async (_url: string, init: { body: string }) => {
+        const { texts } = JSON.parse(init.body) as { texts: string[] }
+        if (texts[0] !== 'queued') return new Response('failed', { status })
+        return Response.json({ masked: texts.map((t) => `M(${t})`) })
+      })
+
+    it('scrubs queued chunks without sending them once an outage outlives its retries', async () => {
+      serviceFailingWith(503)
+
+      const { masked } = await maskPIIBatchViaHttp(failingThenQueued(), [], undefined, undefined, {
+        failedChunkPlaceholder: PLACEHOLDER,
+      })
+
+      expect(masked.every((value) => value === PLACEHOLDER)).toBe(true)
+    })
+
+    it('still masks queued chunks after a persistent 500, which is not an outage', async () => {
+      serviceFailingWith(500)
+
+      const { masked } = await maskPIIBatchViaHttp(failingThenQueued(), [], undefined, undefined, {
+        failedChunkPlaceholder: PLACEHOLDER,
+      })
+
+      expect(masked.slice(0, CHUNK_SIZE * CONCURRENCY).every((v) => v === PLACEHOLDER)).toBe(true)
+      expect(masked.slice(CHUNK_SIZE * CONCURRENCY).every((v) => v === 'M(queued)')).toBe(true)
+    })
+  })
+
   it('does not retry a null 200 body (deterministic, not a transient TypeError)', async () => {
     fetchMock.mockResolvedValueOnce(
       new Response('null', { status: 200, headers: { 'content-type': 'application/json' } })