@@ -118,7 +118,7 @@ Redaction applies to text strings at the configured workflow stages. It does not
 
 If log redaction fails, affected strings are replaced with `[REDACTION_FAILED]` rather than stored unmasked. Failure at Workflow input or Block outputs stops the run instead of passing unredacted values downstream.
 
-Base64-encoded file content, such as a `data:` URL or a long base64 string, can't be checked for PII as text. In stored logs it is replaced with `[BINARY_CONTENT_REDACTED]`. At Workflow input and Block outputs it passes through unchanged, so later blocks can still read the file. **Run from block** and re-running with a previous run's input read that run's log, so they see the marker in place of that content, the same way they see masked PII.
+Base64-encoded file content, such as a `data:` URL or a long base64 string, can't be checked for PII as text, because any PII is inside the encoded file. In stored logs it is replaced with `[BINARY_CONTENT_REDACTED]`. At Workflow input and Block outputs it is masked like any other string. **Run from block** and re-running with a previous run's input read that run's log, so they see the marker in place of that content, the same way they see masked PII.
 
 ## Self-hosted setup
 
 
@@ -228,30 +228,93 @@ def _regex_only(entities: list[str] | None, score_threshold: float | None) -> bo
     )
 
 
-def _require_nlp_length(
-    texts: list[str],
+# spaCy refuses (error E088) any text longer than its model's `max_length`, so a
+# longer text runs NER in windows of this many characters, following spaCy's
+# guidance to segment long documents (it sizes NER memory at roughly 1GB per
+# 100,000 characters of one text).
+NER_WINDOW_CHARS = 100_000
+# How much of the previous window each window repeats. Far longer than any entity
+# spaCy tags, so an entity cut by one window's edge lies whole inside the next.
+NER_WINDOW_OVERLAP = 1_000
+
+
+def _exceeds_nlp_length(text: str, language: str) -> bool:
+    limit = _NLP_MAX_LENGTH.get(language)
+    return limit is not None and len(text) > limit
+
+
+def _ner_window_entities(entities: list[str] | None, language: str) -> list[str]:
+    """The requested entities only spaCy NER produces and this language serves, or
+    all of them when `entities` is empty (detect-all)."""
+    ner = NER_ENTITIES & set(analyzer.get_supported_entities(language))
+    return sorted(ner if not entities else ner.intersection(entities))
+
+
+def _analyze_long(
+    text: str,
     language: str,
     entities: list[str] | None,
     score_threshold: float | None,
-) -> None:
-    """Refuse, as a 422, any text longer than the language model's spaCy
-    `max_length` when the request runs the NLP pass. spaCy raises ValueError E088
-    for such a text, which FastAPI answers as an opaque 500 that callers retry as
-    a transient failure; the same text fails the same way every time, so callers
-    must split it rather than resend it. Regex-only requests skip spaCy and are
-    not limited."""
-    if _regex_only(entities, score_threshold):
-        return
-    limit = _NLP_MAX_LENGTH.get(language)
-    if limit is None:
-        return
-    for index, text in enumerate(texts):
-        if len(text) > limit:
-            raise HTTPException(
-                status_code=422,
-                detail=f"Text {index} has {len(text)} characters; NER accepts at most "
-                f"{limit} per text. Split it before sending.",
+    return_decision_process: bool = False,
+    ad_hoc_recognizers: list[PatternRecognizer] | None = None,
+) -> list[RecognizerResult]:
+    """Analyze a text longer than spaCy's `max_length`.
+
+    Pattern, checksum, and custom recognizers have no length limit, so they run
+    once over the whole text (against blank NLP artifacts, as the regex-only path
+    does) and no match can straddle a cut. Only NER runs in overlapping windows
+    (see NER_WINDOW_CHARS), asked for just its own entities. A span touching a
+    window's far edge may be cut short, so it is left to the next window, which
+    sees it whole; a span ending inside the overlap before that edge was already
+    found by the previous window and is skipped. The rest are shifted to
+    whole-text offsets, and overlapping spans are left to the anonymizer, which
+    merges them. Context words cannot raise a pattern match's score here, which
+    matters only when a score_threshold is set."""
+    entities = entities or None
+    results = analyzer.analyze(
+        text=text,
+        language=language,
+        entities=entities,
+        score_threshold=score_threshold,
+        return_decision_process=return_decision_process,
+        nlp_artifacts=_BLANK_ARTIFACTS.get(language),
+        ad_hoc_recognizers=ad_hoc_recognizers or None,
+    )
+    window_entities = _ner_window_entities(entities, language)
+    if not window_entities:
+        return results
+    step = NER_WINDOW_CHARS - NER_WINDOW_OVERLAP
+    for start in range(0, len(text), step):
+        window = text[start : start + NER_WINDOW_CHARS]
+        is_last = start + NER_WINDOW_CHARS >= len(text)
+        seen_through = NER_WINDOW_OVERLAP if start else 0
+        for result in analyzer.analyze(
+            text=window,
+            language=language,
+            entities=window_entities,
+            score_threshold=score_threshold,
+            return_decision_process=return_decision_process,
+        ):
+            if result.end < seen_through or (not is_last and result.end >= len(window)):
+                continue
+            results.append(
+                RecognizerResult(
+                    result.entity_type,
+                    result.start + start,
+                    result.end + start,
+                    result.score,
+                    result.analysis_explanation,
+                    result.recognition_metadata,
+                )
             )
+        if is_last:
+            break
+    unique: dict[tuple[str, int, int], RecognizerResult] = {}
+    for result in results:
+        key = (result.entity_type, result.start, result.end)
+        if key not in unique or result.score > unique[key].score:
+            unique[key] = result
+    return list(unique.values())
 
 
 def _analyze_one(
@@ -262,13 +325,15 @@ def _analyze_one(
     return_decision_process: bool = False,
     ad_hoc_recognizers: list[PatternRecognizer] | None = None,
 ):
-    _require_nlp_length([text], language, entities, score_threshold)
+    regex_only = _regex_only(entities, score_threshold)
+    if not regex_only and _exceeds_nlp_length(text, language):
+        return _analyze_long(
+            text, language, entities, score_threshold, return_decision_process, ad_hoc_recognizers
+        )
     # Regex-only requests reuse a blank NlpArtifacts to skip the spaCy NLP pass;
     # otherwise analyze() computes artifacts (runs spaCy) as usual. Custom-pattern
     # recognizers are regex-based, so they run fine against the blank artifacts.
-    nlp_artifacts = (
-        _BLANK_ARTIFACTS.get(language) if _regex_only(entities, score_threshold) else None
-    )
+    nlp_artifacts = _BLANK_ARTIFACTS.get(language) if regex_only else None
     return analyzer.analyze(
         text=text,
         language=language,
@@ -292,8 +357,9 @@ def _analyze_many(
     score_threshold: float | None,
     ad_hoc_recognizers: list[PatternRecognizer] | None = None,
 ):
-    """Analyze many texts, skipping the spaCy pass for regex-only requests."""
-    _require_nlp_length(texts, language, entities, score_threshold)
+    """Analyze many texts, skipping the spaCy pass for regex-only requests and
+    analyzing any text longer than spaCy's `max_length` on its own (see
+    `_analyze_long`)."""
     if _regex_only(entities, score_threshold):
         blank = _BLANK_ARTIFACTS.get(language)
         return [
@@ -307,16 +373,24 @@ def _analyze_many(
             )
             for text in texts
         ]
-    return list(
-        batch_analyzer.analyze_iterator(
-            texts=texts,
-            language=language,
-            batch_size=NLP_BATCH_SIZE,
-            entities=entities or None,
-            score_threshold=score_threshold,
-            ad_hoc_recognizers=ad_hoc_recognizers or None,
-        )
+    long_indices = {i for i, text in enumerate(texts) if _exceeds_nlp_length(text, language)}
+    short_indices = [i for i in range(len(texts)) if i not in long_indices]
+    results: list[list[RecognizerResult]] = [[] for _ in texts]
+    short_results = batch_analyzer.analyze_iterator(
+        texts=[texts[i] for i in short_indices],
+        language=language,
+        batch_size=NLP_BATCH_SIZE,
+        entities=entities or None,
+        score_threshold=score_threshold,
+        ad_hoc_recognizers=ad_hoc_recognizers or None,
     )
+    for i, per_text in zip(short_indices, short_results):
+        results[i] = list(per_text)
+    for i in sorted(long_indices):
+        results[i] = _analyze_long(
+            texts[i], language, entities, score_threshold, ad_hoc_recognizers=ad_hoc_recognizers
+        )
+    return results
 
 
 @asynccontextmanager
 
@@ -1,7 +1,6 @@
 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'
-import { PII_MAX_TEXT_CHARS } from '@/lib/guardrails/pii-batching'
 import { POST } from '@/app/api/guardrails/mask-batch/route'
 
 const mockCheckInternalAuth = hybridAuthMockFns.mockCheckInternalAuth
@@ -52,20 +51,4 @@ describe('POST /api/guardrails/mask-batch', () => {
 
     expect(res.status).toBe(status)
   })
-
-  it('splits a text too long for one NER pass before calling Presidio, then rejoins it', async () => {
-    presidioFetch.mockImplementation(async (_url: string, init: { body: string }) => {
-      const { texts } = JSON.parse(init.body) as { texts: string[] }
-      return Response.json({ texts: texts.map((t) => t.replaceAll('Ada', '<PERSON>')) })
-    })
-    const text = 'Ada wrote this line.\n'.repeat(60_000)
-
-    const res = await POST(createMockRequest('POST', { texts: [text], entityTypes: ['PERSON'] }))
-
-    expect(res.status).toBe(200)
-    expect(await res.json()).toEqual({ masked: [text.replaceAll('Ada', '<PERSON>')] })
-    const sent = presidioFetch.mock.calls.flatMap(([, init]) => JSON.parse(init.body).texts)
-    expect(sent.length).toBeGreaterThan(1)
-    for (const piece of sent) expect(piece.length).toBeLessThanOrEqual(PII_MAX_TEXT_CHARS)
-  })
 })
@@ -16,7 +16,6 @@ vi.mock('@/lib/auth/internal', () => authInternalMock)
 vi.mock('@sim/utils/helpers', () => utilsHelpersMock)
 
 import { maskPIIBatchViaHttp } from '@/lib/guardrails/mask-client'
-import { PII_MAX_TEXT_CHARS } from '@/lib/guardrails/pii-batching'
 import { POST as maskBatchRoute } from '@/app/api/guardrails/mask-batch/route'
 
 const mockSleep = utilsHelpersMockFns.mockSleep
@@ -82,79 +81,6 @@ describe('maskPIIBatchViaHttp', () => {
     expect(mockSleep).toHaveBeenCalledTimes(2)
   })
 
-  describe('with a text longer than one Presidio pass allows', () => {
-    /** Each request masks `Ada` to `<PERSON>`, so pieces rejoin to the masked whole. */
-    const maskingAda = () =>
-      fetchMock.mockImplementation(async (_url: string, init: { body: string }) => {
-        const { texts } = JSON.parse(init.body) as { texts: string[] }
-        return Response.json({ masked: texts.map((t) => t.replaceAll('Ada', '<PERSON>')) })
-      })
-    const sentPieces = () =>
-      fetchMock.mock.calls.flatMap(([, init]) => JSON.parse(init.body).texts as string[])
-
-    it('sends newline-aligned pieces within the limit and rejoins the masked result', async () => {
-      maskingAda()
-      const text = Array.from({ length: 60_000 }, (_, i) => `row ${i} owner Ada\n`).join('')
-
-      const { masked } = await maskPIIBatchViaHttp([text, 'short Ada'], ['PERSON'])
-
-      expect(masked).toEqual([text.replaceAll('Ada', '<PERSON>'), 'short <PERSON>'])
-      const pieces = sentPieces().filter((p) => p !== 'short Ada')
-      expect(pieces.length).toBeGreaterThan(1)
-      expect(pieces.join('')).toBe(text)
-      for (const piece of pieces) expect(piece.length).toBeLessThanOrEqual(PII_MAX_TEXT_CHARS)
-      for (const piece of pieces.slice(0, -1)) expect(piece.endsWith('\n')).toBe(true)
-    })
-
-    it('cuts minified JSON at punctuation so no email is split across pieces', async () => {
-      maskingAda()
-      const record = '{"email":"ada.lovelace@example.com","id":1234}'
-      const text = `[${Array.from({ length: 25_000 }, () => record).join(',')}]`
-
-      await maskPIIBatchViaHttp([text], ['EMAIL_ADDRESS'])
-
-      const pieces = sentPieces()
-      expect(pieces.length).toBeGreaterThan(1)
-      const emailsSent = pieces.reduce(
-        (count, piece) => count + piece.split('ada.lovelace@example.com').length - 1,
-        0
-      )
-      expect(emailsSent).toBe(25_000)
-    })
-
-    it('never splits a surrogate pair when the text has nothing to cut at', async () => {
-      // One leading unit puts a high surrogate exactly where a hard cut would land.
-      const text = `a${'\u{1F600}'.repeat(PII_MAX_TEXT_CHARS)}`
-
-      const { masked } = await maskPIIBatchViaHttp([text], [])
-
-      const pieces = sentPieces()
-      expect(pieces.length).toBeGreaterThan(1)
-      expect(pieces.join('')).toBe(text)
-      expect(masked).toEqual([`M(${pieces.join(')M(')})`])
-    })
-
-    it('replaces the whole text when any one of its pieces fails', async () => {
-      const text = `${'safe line\n'.repeat(120_000)}poison`
-      fetchMock.mockImplementation(async (_url: string, init: { body: string }) => {
-        const { texts } = JSON.parse(init.body) as { texts: string[] }
-        if (texts.some((t) => t.endsWith('poison'))) return new Response('no', { status: 422 })
-        return Response.json({ masked: texts })
-      })
-
-      const { masked, scrubbedCount } = await maskPIIBatchViaHttp(
-        [text],
-        [],
-        undefined,
-        undefined,
-        { failedChunkPlaceholder: '[FAILED]' }
-      )
-
-      expect(masked).toEqual(['[FAILED]'])
-      expect(scrubbedCount).toBe(1)
-    })
-  })
-
   describe('with a failed-chunk placeholder', () => {
     const PLACEHOLDER = '[FAILED]'
     const CHUNK_SIZE = 2000
 
@@ -8,7 +8,7 @@ import { generateInternalToken } from '@/lib/auth/internal'
 import { env } from '@/lib/core/config/env'
 import { mapWithConcurrency } from '@/lib/core/utils/concurrency'
 import { getInternalApiBaseUrl } from '@/lib/core/utils/urls'
-import { chunkIndicesByBudget, joinSegments, segmentTexts } from '@/lib/guardrails/pii-batching'
+import { chunkIndicesByBudget } from '@/lib/guardrails/pii-batching'
 import type { CustomPiiPattern } from '@/lib/guardrails/pii-entities'
 
 const logger = createLogger('PiiMaskClient')
@@ -106,9 +106,9 @@ function isOutageError(error: unknown): boolean {
 
 interface MaskPIIBatchOptions {
   /**
-   * When set, a chunk that cannot be masked has every string with a piece in it
-   * replaced with this value instead of rejecting the call, so one bad chunk costs
-   * only its own strings. After an outage (see {@link OUTAGE_STATUSES}) exhausts a chunk's
+   * When set, a chunk that cannot be masked has each of its strings replaced with
+   * this value instead of rejecting the call, so one bad chunk costs only its own
+   * strings. After an outage (see {@link OUTAGE_STATUSES}) exhausts a chunk's
    * retries, chunks not yet sent are replaced the same way.
    */
   failedChunkPlaceholder?: string
@@ -127,9 +127,7 @@ interface MaskPIIBatchResult {
  * Only the app task reaches the Presidio service (it holds `PII_URL`), 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.
- * A string too long for one NER pass is cut into pieces first (see
- * {@link segmentTexts}), so no request carries a text Presidio would refuse.
- * Pieces are grouped into byte/count-budgeted chunks (keeping each request far
+ * Strings are grouped into byte/count-budgeted chunks (keeping each request far
  * under the 10MB Next body limit) and the chunks are sent with bounded
  * concurrency, so a large payload fans out rather than serializing; order is
  * preserved, so the returned `masked` array matches `texts` length.
@@ -152,47 +150,38 @@ export async function maskPIIBatchViaHttp(
 
   const url = `${getInternalApiBaseUrl()}/api/guardrails/mask-batch`
   const placeholder = options.failedChunkPlaceholder
-  const { segments, owners } = segmentTexts(texts)
-  const maskedSegments = new Array<string>(segments.length)
-  const scrubbedTexts = new Set<number>()
+  const masked = new Array<string>(texts.length)
   let serviceUnavailable = false
+  let scrubbedCount = 0
   let firstFailure: unknown
 
-  const scrubChunk = (indices: number[]) => {
-    for (const k of indices) scrubbedTexts.add(owners[k])
+  const scrubChunk = (indices: number[], value: string) => {
+    for (const i of indices) masked[i] = value
+    scrubbedCount += indices.length
   }
 
-  await mapWithConcurrency(chunkIndicesByBudget(segments), CHUNK_CONCURRENCY, async (indices) => {
+  await mapWithConcurrency(chunkIndicesByBudget(texts), CHUNK_CONCURRENCY, async (indices) => {
     if (serviceUnavailable && placeholder !== undefined) {
-      scrubChunk(indices)
+      scrubChunk(indices, placeholder)
       return
     }
     try {
-      const chunk = indices.map((k) => segments[k])
+      const chunk = indices.map((i) => texts[i])
       const out = await postChunk(url, chunk, entityTypes, language, customPatterns)
       if (out.length !== chunk.length) {
         throw new Error('PII mask-batch returned an unexpected result')
       }
-      indices.forEach((segmentIndex, k) => {
-        maskedSegments[segmentIndex] = out[k]
+      indices.forEach((originalIndex, k) => {
+        masked[originalIndex] = out[k]
       })
     } catch (error) {
       if (placeholder === undefined) throw error
       if (isOutageError(error)) serviceUnavailable = true
       firstFailure ??= error
-      scrubChunk(indices)
+      scrubChunk(indices, placeholder)
     }
   })
 
-  const joined = joinSegments(maskedSegments, owners, texts.length)
-  // A text with any unmasked piece is replaced whole: its other pieces are masked,
-  // but keeping them would persist a fragment around the hole.
-  const masked =
-    placeholder === undefined
-      ? joined
-      : joined.map((text, i) => (scrubbedTexts.has(i) ? placeholder : text))
-  const scrubbedCount = scrubbedTexts.size
-
   if (scrubbedCount > 0) {
     logger.error('PII masking failed for some chunks; replaced their strings', {
       error: getErrorMessage(firstFailure),