diff --git a/apps/docs/content/docs/platform/enterprise/data-retention.mdx b/apps/docs/content/docs/platform/enterprise/data-retention.mdx
index b12a72e6054..c3252f45023 100644
--- a/apps/docs/content/docs/platform/enterprise/data-retention.mdx
+++ b/apps/docs/content/docs/platform/enterprise/data-retention.mdx
@@ -118,6 +118,8 @@ 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, 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
 
 ### Enable cleanup and schedule it
diff --git a/apps/pii/server.py b/apps/pii/server.py
index 6934daf8355..405d96e13bc 100644
--- a/apps/pii/server.py
+++ b/apps/pii/server.py
@@ -184,6 +184,7 @@ def build_analyzer() -> AnalyzerEngine:
 anonymizer: AnonymizerEngine
 NER_ENTITIES: frozenset
 _BLANK_ARTIFACTS: dict
+_NLP_MAX_LENGTH: dict
 
 # Every entity the spaCy NER recognizers can actually produce. A request touching
 # any of these must run spaCy; a request naming only non-NER (regex/checksum)
@@ -227,6 +228,96 @@ def _regex_only(entities: list[str] | None, score_threshold: float | None) -> bo
     )
 
 
+# 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 _window_entities(
+    entities: list[str] | None, language: str, score_threshold: float | None
+) -> list[str]:
+    """The requested built-in entities this language serves that windows must find,
+    or all of them when `entities` is empty (detect-all). Without a score_threshold
+    that is only what spaCy NER produces; with one it also includes the pattern
+    entities, whose context-boosted score can decide whether a match clears the
+    threshold and needs the real NLP artifacts only a window has."""
+    served = set(analyzer.get_supported_entities(language))
+    if score_threshold is None or score_threshold <= 0:
+        served &= NER_ENTITIES
+    return sorted(served if not entities else served.intersection(entities))
+
+
+def _analyze_long(
+    text: str,
+    language: str,
+    entities: list[str] | None,
+    score_threshold: float | None,
+    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, plus the pattern
+    entities when a score_threshold makes context scoring matter (see
+    `_window_entities`). Every window's spans are kept, shifted to whole-text
+    offsets: NER reads context, so a window can find a name its neighbour missed
+    or cut short. A span found more than once keeps its highest score, and
+    overlapping spans of one type are merged by the anonymizer."""
+    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 = _window_entities(entities, language, score_threshold)
+    if not window_entities:
+        return results
+    step = NER_WINDOW_CHARS - NER_WINDOW_OVERLAP
+    for start in range(0, len(text), step):
+        for result in analyzer.analyze(
+            text=text[start : start + NER_WINDOW_CHARS],
+            language=language,
+            entities=window_entities,
+            score_threshold=score_threshold,
+            return_decision_process=return_decision_process,
+        ):
+            results.append(
+                RecognizerResult(
+                    result.entity_type,
+                    result.start + start,
+                    result.end + start,
+                    result.score,
+                    result.analysis_explanation,
+                    result.recognition_metadata,
+                )
+            )
+        if start + NER_WINDOW_CHARS >= len(text):
+            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(
     text: str,
     language: str,
@@ -235,12 +326,15 @@ def _analyze_one(
     return_decision_process: bool = False,
     ad_hoc_recognizers: list[PatternRecognizer] | None = None,
 ):
+    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,
@@ -264,7 +358,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."""
+    """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 [
@@ -278,21 +374,29 @@ 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
 async def lifespan(_app: FastAPI) -> AsyncIterator[None]:
-    global analyzer, batch_analyzer, anonymizer, NER_ENTITIES, _BLANK_ARTIFACTS
+    global analyzer, batch_analyzer, anonymizer, NER_ENTITIES, _BLANK_ARTIFACTS, _NLP_MAX_LENGTH
     logger.info("building analyzer (spacy)")
     analyzer = build_analyzer()
     batch_analyzer = BatchAnalyzerEngine(analyzer_engine=analyzer)
@@ -319,6 +423,10 @@ async def lifespan(_app: FastAPI) -> AsyncIterator[None]:
         language: analyzer.nlp_engine.process_text("", language)
         for language in SUPPORTED_LANGUAGES
     }
+    _NLP_MAX_LENGTH = {
+        language: nlp.max_length
+        for language, nlp in getattr(analyzer.nlp_engine, "nlp", {}).items()
+    }
     yield
 
 
diff --git a/apps/sim/lib/guardrails/mask-client.test.ts b/apps/sim/lib/guardrails/mask-client.test.ts
index 48ed3feedde..9da95d11d06 100644
--- a/apps/sim/lib/guardrails/mask-client.test.ts
+++ b/apps/sim/lib/guardrails/mask-client.test.ts
@@ -73,6 +73,14 @@ describe('maskPIIBatchViaHttp', () => {
     expect(mockSleep).toHaveBeenCalledTimes(7)
   })
 
+  it('stops after three attempts on a persistent 500, which can be a deterministic input failure', async () => {
+    fetchMock.mockImplementation(async () => new Response('failed', { status: 500 }))
+
+    await expect(maskPIIBatchViaHttp(['a'], [])).rejects.toThrow(/mask-batch request failed/)
+    expect(fetchMock).toHaveBeenCalledTimes(3)
+    expect(mockSleep).toHaveBeenCalledTimes(2)
+  })
+
   describe('with a failed-chunk placeholder', () => {
     const PLACEHOLDER = '[FAILED]'
     const CHUNK_SIZE = 2000
diff --git a/apps/sim/lib/guardrails/mask-client.ts b/apps/sim/lib/guardrails/mask-client.ts
index 07725cbe3d7..ddd9199b08d 100644
--- a/apps/sim/lib/guardrails/mask-client.ts
+++ b/apps/sim/lib/guardrails/mask-client.ts
@@ -36,6 +36,15 @@ const CHUNK_CONCURRENCY = env.PII_MASK_CHUNK_CONCURRENCY ?? 64
  */
 const MAX_CHUNK_ATTEMPTS = 8
 
+/**
+ * Attempts for a chunk the route answers with a 500. The route keeps 503 for a
+ * Presidio outage and 422 for input Presidio refused, so a 500 is an unclassified
+ * failure that may be deterministic for this input: two retries cover a one-off
+ * crash without spending the full outage budget resending input that fails the
+ * same way every time.
+ */
+const MAX_ATTEMPTS_AFTER_SERVER_ERROR = 3
+
 const RETRYABLE_STATUSES = new Set([408, 429, 500, 502, 503, 504])
 
 class MaskChunkHttpError extends Error {
@@ -196,7 +205,11 @@ async function postChunk(
     try {
       return await postChunkOnce(url, texts, entityTypes, language, customPatterns)
     } catch (error) {
-      if (attempt >= MAX_CHUNK_ATTEMPTS || !isRetryableChunkError(error)) {
+      const maxAttempts =
+        error instanceof MaskChunkHttpError && error.status === 500
+          ? MAX_ATTEMPTS_AFTER_SERVER_ERROR
+          : MAX_CHUNK_ATTEMPTS
+      if (attempt >= maxAttempts || !isRetryableChunkError(error)) {
         throw error
       }
       const retryAfterMs = error instanceof MaskChunkHttpError ? error.retryAfterMs : null
diff --git a/apps/sim/lib/logs/execution/pii-large-values.test.ts b/apps/sim/lib/logs/execution/pii-large-values.test.ts
index 90f7f9ba227..ad917046339 100644
--- a/apps/sim/lib/logs/execution/pii-large-values.test.ts
+++ b/apps/sim/lib/logs/execution/pii-large-values.test.ts
@@ -289,6 +289,37 @@ describe('redactManifest — chunk-wise', () => {
   })
 })
 
+describe('encoded file bodies inside large values', () => {
+  const encodedFile = Buffer.from(
+    Array.from({ length: 48 * 1024 }, (_, i) => (i * 7919 + 13) % 256)
+  ).toString('base64')
+
+  it('replaces them in the execution log', async () => {
+    chunkData.set(REF.id, { file: encodedFile, note: 'contact bob' })
+
+    const result = await redactLargeValueRefs(
+      { finalOutput: REF },
+      { entityTypes: ['PERSON'], language: 'en', store: STORE }
+    )
+
+    expect(result.finalOutput).toEqual({
+      file: '[BINARY_CONTENT_REDACTED]',
+      note: 'MASKED(contact bob)',
+    })
+  })
+
+  it('masks them as text in live block state', async () => {
+    chunkData.set(REF.id, { file: encodedFile })
+
+    const result = await redactLargeValueRefsInValue(
+      { 'block-1': { output: REF } },
+      { entityTypes: ['PERSON'], language: 'en', store: STORE, onFailure: 'throw' }
+    )
+
+    expect(result).toEqual({ 'block-1': { output: { file: `MASKED(${encodedFile})` } } })
+  })
+})
+
 describe('redactLargeValueRefsInValue (arbitrary blockStates)', () => {
   it('hydrates + re-stores a ref nested in a non-RedactablePayload shape', async () => {
     chunkData.set(REF.id, { note: 'contact bob' })
diff --git a/apps/sim/lib/logs/execution/pii-large-values.ts b/apps/sim/lib/logs/execution/pii-large-values.ts
index bcefb5381ca..98f241a17c8 100644
--- a/apps/sim/lib/logs/execution/pii-large-values.ts
+++ b/apps/sim/lib/logs/execution/pii-large-values.ts
@@ -43,6 +43,12 @@ export interface RedactLargeValueRefsOptions {
    * than feeding a marker (or leaking raw bytes) downstream.
    */
   onFailure?: PiiRedactionFailureMode
+  /**
+   * @internal Set by {@link redactLargeValueRefs}, the execution log's entry point,
+   * so encoded file bodies inside its large values are replaced like inline ones
+   * (see `PiiRedactionOptions.replaceEncodedBinary`). Callers never set it.
+   */
+  replaceEncodedBinary?: boolean
   /**
    * @internal Shared across every nested invocation spawned from one entry-point
    * call, so oversized hydrations serialize globally — not just within their own
@@ -100,7 +106,7 @@ export async function redactLargeValueRefs(
   payload: RedactablePayload,
   options: RedactLargeValueRefsOptions
 ): Promise<RedactablePayload> {
-  const gatedOptions = withOversizedGate(options)
+  const gatedOptions = withOversizedGate({ ...options, replaceEncodedBinary: true })
   // Collect refs across the WHOLE payload first (shared `seen`), so every ref is
   // hydrated+masked+re-stored in one bounded-concurrency pass instead of one
   // sequential pass per key. A ref shared across keys is walked/masked once.
@@ -269,6 +275,7 @@ async function maskMaterializedValue(
     language: options.language,
     customPatterns: options.customPatterns,
     onFailure: options.onFailure ?? 'scrub',
+    replaceEncodedBinary: options.replaceEncodedBinary,
   })
   return redactValueRefs(masked, options)
 }
diff --git a/apps/sim/lib/logs/execution/pii-redaction.test.ts b/apps/sim/lib/logs/execution/pii-redaction.test.ts
index b32e4c5c85e..4c314539b7e 100644
--- a/apps/sim/lib/logs/execution/pii-redaction.test.ts
+++ b/apps/sim/lib/logs/execution/pii-redaction.test.ts
@@ -5,6 +5,7 @@ import { afterAll, beforeEach, describe, expect, it, vi } from 'vitest'
 vi.mock('@/lib/auth/internal', () => authInternalMock)
 
 import {
+  BINARY_CONTENT_REDACTED_MARKER,
   PiiRedactionError,
   REDACTION_FAILED_MARKER,
   redactObjectStrings,
@@ -92,7 +93,7 @@ describe('redactPIIFromExecution', () => {
   })
 
   it('masks large strings too (never left unredacted)', async () => {
-    const big = 'x'.repeat(200 * 1024)
+    const big = 'lorem ipsum '.repeat(17_000)
     const payload = { finalOutput: { big, small: 'pii' } }
 
     const result = await redactPIIFromExecution(payload, { entityTypes: [] })
@@ -129,6 +130,70 @@ describe('redactPIIFromExecution', () => {
   })
 })
 
+/** Base64 of 48KB of non-text bytes, the shape a file body takes in a payload. */
+const encodedFile = Buffer.from(
+  Array.from({ length: 48 * 1024 }, (_, i) => (i * 7919 + 13) % 256)
+).toString('base64')
+
+describe('encoded binary content', () => {
+  it('replaces encoded file bodies in a persisted log without sending them to Presidio', async () => {
+    const payload = {
+      traceSpans: [{ blockId: 'write', input: { content: encodedFile, name: 'Ada Lovelace' } }],
+      workflowInput: { file: `data:application/pdf;base64,${encodedFile}` },
+    }
+
+    const result = await redactPIIFromExecution(payload, { entityTypes: ['PERSON'] })
+
+    const [span] = result.traceSpans as { input: { content: string; name: string } }[]
+    expect(span.input.content).toBe(BINARY_CONTENT_REDACTED_MARKER)
+    expect((result.workflowInput as { file: string }).file).toBe(BINARY_CONTENT_REDACTED_MARKER)
+    expect(span.input.name).toBe('MASKED(Ada Lovelace)')
+    expect(sentTexts()).toEqual(['Ada Lovelace'])
+  })
+
+  it('keeps live data on the masking path, so no string skips redaction by its shape', async () => {
+    const shapedLikeAFile = `${'Ab1_'.repeat(256)}123-45-6789`
+    const value = { content: encodedFile, id: shapedLikeAFile }
+
+    const result = await redactObjectStrings(value, { entityTypes: [], onFailure: 'throw' })
+
+    expect(result).toEqual({ content: `MASKED(${encodedFile})`, id: `MASKED(${shapedLikeAFile})` })
+    expect(sentTexts()).toEqual([encodedFile, shapedLikeAFile])
+  })
+
+  /** `encodedFile` wrapped at `width` columns, as MIME (76) and PEM (64) encoders emit it. */
+  const wrapped = (width: number, newline: string) =>
+    (encodedFile.match(new RegExp(`.{1,${width}}`, 'g')) ?? []).join(newline)
+
+  it.each([
+    ['with a trailing newline', `${encodedFile}\n`],
+    ['wrapped at 76 columns', wrapped(76, '\n')],
+    ['wrapped at 64 columns with CRLF', `${wrapped(64, '\r\n')}\r\n`],
+    ['as a wrapped data URL', `data:application/pdf;base64,${wrapped(76, '\n')}`],
+    ['as a data URL with an uppercase scheme', `DATA:application/pdf;base64,${encodedFile}`],
+    [
+      'as a data URL with long metadata',
+      `data:application/pdf;name=${'quarterly-report-'.repeat(40)}.pdf;base64,${encodedFile}`,
+    ],
+  ])('replaces an encoded file body %s in a persisted log', async (_label, body) => {
+    const result = await redactPIIFromExecution({ finalOutput: body }, { entityTypes: [] })
+
+    expect(result.finalOutput).toBe(BINARY_CONTENT_REDACTED_MARKER)
+    expect(sentTexts()).toEqual([])
+  })
+
+  it.each([
+    ['a newline-separated list of usernames', 'AdaLovelace1815\n'.repeat(100)],
+    ['a short base64 token', 'QWRhIExvdmVsYWNl'],
+    ['a long space-separated sentence', 'Ada Lovelace wrote code in 1843 '.repeat(150)],
+  ])('still masks %s in a persisted log, which is not an encoded file', async (_label, text) => {
+    const result = await redactPIIFromExecution({ finalOutput: text }, { entityTypes: [] })
+
+    expect(result.finalOutput).toBe(`MASKED(${text})`)
+    expect(sentTexts()).toEqual([text])
+  })
+})
+
 describe('redactObjectStrings', () => {
   it('throws PiiRedactionError on masking failure when onFailure is throw', async () => {
     maskRoute.mockResolvedValueOnce(rejected())
diff --git a/apps/sim/lib/logs/execution/pii-redaction.ts b/apps/sim/lib/logs/execution/pii-redaction.ts
index 34d791e0491..c72e6790590 100644
--- a/apps/sim/lib/logs/execution/pii-redaction.ts
+++ b/apps/sim/lib/logs/execution/pii-redaction.ts
@@ -10,6 +10,69 @@ const logger = createLogger('PiiRedaction')
 /** Replaces text we could not safely mask, so PII is never persisted on failure. */
 export const REDACTION_FAILED_MARKER = '[REDACTION_FAILED]'
 
+/** Replaces an encoded file body in the execution log (see {@link isEncodedBinary}). */
+export const BINARY_CONTENT_REDACTED_MARKER = '[BINARY_CONTENT_REDACTED]'
+
+/**
+ * Shortest bare base64 string treated as an encoded file body. Far longer than any
+ * word, identifier, hash, or token, so only encoded payloads reach it.
+ */
+const MIN_ENCODED_BINARY_CHARS = 1024
+
+const STANDARD_BASE64 = /^[A-Za-z0-9+/]+={0,2}$/
+const URL_SAFE_BASE64 = /^[A-Za-z0-9_-]+={0,2}$/
+/** A `data:` URL's header through its comma, declaring base64 (RFC 2397). */
+const BASE64_DATA_URL_HEADER = /^data:(?:[\w.+-]+\/[\w.+-]+)?(?:;[\w.+-]+=[^;,]*)*;base64,$/i
+
+/** Narrowest and widest lines encoders wrap base64 at: 60 (common), 64 (PEM), 76 (MIME). */
+const MIN_WRAP_WIDTH = 60
+const MAX_WRAP_WIDTH = 76
+
+function isBase64(value: string): boolean {
+  return STANDARD_BASE64.test(value) || URL_SAFE_BASE64.test(value)
+}
+
+/**
+ * The base64 in `value` with trailing whitespace dropped and, when it is wrapped
+ * the way encoders wrap it (every line but the last a fixed width, a multiple of 4
+ * between {@link MIN_WRAP_WIDTH} and {@link MAX_WRAP_WIDTH}), its line breaks
+ * removed. Returns `null` when `value` is not base64.
+ */
+function base64Body(value: string): string | null {
+  const body = value.trimEnd()
+  const firstBreak = body.indexOf('\n')
+  if (firstBreak === -1) return isBase64(body) ? body : null
+  const width = body[firstBreak - 1] === '\r' ? firstBreak - 1 : firstBreak
+  if (width < MIN_WRAP_WIDTH || width > MAX_WRAP_WIDTH || width % 4 !== 0) return null
+  if (!isBase64(body.slice(0, width))) return null
+  const lines = body.split(/\r?\n/)
+  const last = lines[lines.length - 1]
+  const fixedWidth = lines.every((line, i) => i === lines.length - 1 || line.length === width)
+  if (!fixedWidth || last.length === 0 || last.length > width) return null
+  const joined = lines.join('')
+  return isBase64(joined) ? joined : null
+}
+
+/**
+ * Whether `value` is an encoded file body: a `data:` URL declaring base64, or a
+ * bare base64 string at least {@link MIN_ENCODED_BINARY_CHARS} long, either one
+ * optionally wrapped the way encoders wrap it (see {@link base64Body}). Text
+ * masking cannot check such a value for PII, which lives inside the decoded file.
+ */
+function isEncodedBinary(value: string): boolean {
+  if (/^data:/i.test(value)) {
+    const comma = value.indexOf(',')
+    return (
+      comma !== -1 &&
+      BASE64_DATA_URL_HEADER.test(value.slice(0, comma + 1)) &&
+      base64Body(value.slice(comma + 1)) !== null
+    )
+  }
+  if (value.length < MIN_ENCODED_BINARY_CHARS) return false
+  const body = base64Body(value)
+  return body !== null && body.length >= MIN_ENCODED_BINARY_CHARS
+}
+
 /**
  * How to handle a masking failure (Presidio error or over-ceiling payload):
  * - `'scrub'` (default): replace eligible strings with {@link REDACTION_FAILED_MARKER}.
@@ -28,6 +91,16 @@ export interface PiiRedactionOptions {
   customPatterns?: CustomPiiPattern[]
   /** Failure handling. Defaults to `'scrub'`. */
   onFailure?: PiiRedactionFailureMode
+  /**
+   * Replace encoded file bodies (see {@link isEncodedBinary}) with
+   * {@link BINARY_CONTENT_REDACTED_MARKER} instead of masking them as text. Only
+   * for the execution log: masking cannot see PII inside a file, so the log drops
+   * the file rather than keep content it never checked. Replacing can only remove
+   * content, so a string misread as a file costs log detail, never PII. Live data
+   * leaves this off, keeping every string on the masking path so none can skip
+   * redaction by its shape, and keeping file bytes the next block reads.
+   */
+  replaceEncodedBinary?: boolean
 }
 
 /** Thrown when in-flight redaction (`onFailure: 'throw'`) cannot mask safely. */
@@ -79,14 +152,27 @@ function isEligibleString(value: string): boolean {
 }
 
 /**
- * Rebuild `value` replacing every eligible string leaf with `handle(leaf)`.
- * Used for both collection (handle records and returns the input) and
- * substitution (handle returns the masked value), so traversal order and
- * eligibility are guaranteed identical across the two passes.
+ * What one pass does with each string leaf: `text` for a string Presidio masks,
+ * and, when set, `encodedBinary` for an encoded file body it never sees (see
+ * {@link isEncodedBinary}). Without `encodedBinary` every string is text.
  */
-function transformStrings(value: unknown, handle: (s: string) => string): unknown {
+interface StringVisitor {
+  text: (s: string) => string
+  encodedBinary?: (s: string) => string
+}
+
+/**
+ * Rebuild `value` replacing every eligible string leaf with the visitor's result.
+ * Used for both collection (the visitor records and returns the input) and
+ * substitution (it returns the masked value), so traversal order, eligibility, and
+ * the text/binary split are guaranteed identical across the two passes.
+ */
+function transformStrings(value: unknown, visit: StringVisitor): unknown {
   if (typeof value === 'string') {
-    return isEligibleString(value) ? handle(value) : value
+    if (!isEligibleString(value)) return value
+    return visit.encodedBinary && isEncodedBinary(value)
+      ? visit.encodedBinary(value)
+      : visit.text(value)
   }
   // Treat offloaded large-value refs as opaque: masking their internal `key`/`id`
   // strings would corrupt the reference. Their content is handled separately
@@ -95,12 +181,12 @@ function transformStrings(value: unknown, handle: (s: string) => string): unknow
     return value
   }
   if (Array.isArray(value)) {
-    return value.map((item) => transformStrings(item, handle))
+    return value.map((item) => transformStrings(item, visit))
   }
   if (value !== null && typeof value === 'object') {
     const out: Record<string, unknown> = {}
     for (const [key, v] of Object.entries(value)) {
-      out[key] = transformStrings(v, handle)
+      out[key] = transformStrings(v, visit)
     }
     return out
   }
@@ -112,17 +198,17 @@ function transformStrings(value: unknown, handle: (s: string) => string): unknow
  * nested `children` are walked, leaving structural metadata (blockId, name,
  * status, timing) untouched so log correlation/display is preserved.
  */
-function transformSpan(span: unknown, handle: (s: string) => string): unknown {
+function transformSpan(span: unknown, visit: StringVisitor): unknown {
   if (span === null || typeof span !== 'object' || Array.isArray(span)) {
-    return transformStrings(span, handle)
+    return transformStrings(span, visit)
   }
   const source = span as Record<string, unknown>
   const out: Record<string, unknown> = { ...source }
   for (const field of SPAN_CONTENT_FIELDS) {
-    if (field in out) out[field] = transformStrings(out[field], handle)
+    if (field in out) out[field] = transformStrings(out[field], visit)
   }
   if (Array.isArray(source.children)) {
-    out.children = source.children.map((child) => transformSpan(child, handle))
+    out.children = source.children.map((child) => transformSpan(child, visit))
   }
   return out
 }
@@ -130,12 +216,56 @@ function transformSpan(span: unknown, handle: (s: string) => string): unknown {
 function transformUnit(
   key: keyof RedactablePayload,
   value: unknown,
-  handle: (s: string) => string
+  visit: StringVisitor
 ): unknown {
   if (key === 'traceSpans' && Array.isArray(value)) {
-    return value.map((span) => transformSpan(span, handle))
+    return value.map((span) => transformSpan(span, visit))
+  }
+  return transformStrings(value, visit)
+}
+
+/** What a collection pass found: the strings to mask and how many encoded file bodies. */
+interface CollectedStrings {
+  collected: string[]
+  encodedBinaryCount: number
+}
+
+function collectStrings(
+  walk: (visit: StringVisitor) => void,
+  replaceEncodedBinary: boolean
+): CollectedStrings {
+  const collected: string[] = []
+  let encodedBinaryCount = 0
+  walk({
+    text: (s) => {
+      collected.push(s)
+      return s
+    },
+    encodedBinary: replaceEncodedBinary
+      ? (s) => {
+          encodedBinaryCount++
+          return s
+        }
+      : undefined,
+  })
+  return { collected, encodedBinaryCount }
+}
+
+/** Mask the collected strings and return the substitution pass's visitor. */
+async function substitutionVisitor(
+  collected: string[],
+  options: PiiRedactionOptions,
+  replaceEncodedBinary: boolean
+): Promise<{ visit: StringVisitor; scrubbedCount: number }> {
+  const { masked, scrubbedCount } = await maskCollected(collected, options)
+  let index = 0
+  return {
+    visit: {
+      text: () => masked[index++],
+      encodedBinary: replaceEncodedBinary ? () => BINARY_CONTENT_REDACTED_MARKER : undefined,
+    },
+    scrubbedCount,
   }
-  return transformStrings(value, handle)
 }
 
 /**
@@ -194,17 +324,12 @@ async function maskCollected(
  * block-output stages, so it defaults callers toward `onFailure: 'throw'`.
  */
 export async function redactObjectStrings<T>(value: T, options: PiiRedactionOptions): Promise<T> {
-  const collected: string[] = []
-  transformStrings(value, (s) => {
-    collected.push(s)
-    return s
-  })
+  const replaceEncodedBinary = options.replaceEncodedBinary ?? false
+  const found = collectStrings((visit) => transformStrings(value, visit), replaceEncodedBinary)
+  if (found.collected.length === 0 && found.encodedBinaryCount === 0) return value
 
-  if (collected.length === 0) return value
-
-  const { masked } = await maskCollected(collected, options)
-  let index = 0
-  return transformStrings(value, () => masked[index++]) as T
+  const { visit } = await substitutionVisitor(found.collected, options, replaceEncodedBinary)
+  return transformStrings(value, visit) as T
 }
 
 /**
@@ -217,7 +342,9 @@ export async function redactObjectStrings<T>(value: T, options: PiiRedactionOpti
  * envelope). On a hard masking failure, the eligible strings of every request
  * chunk that failed are replaced with {@link REDACTION_FAILED_MARKER} (the default
  * `onFailure: 'scrub'`) rather than left unredacted — PII is never persisted on
- * the failure path.
+ * the failure path. This payload is the execution log, so encoded file bodies are
+ * always replaced with {@link BINARY_CONTENT_REDACTED_MARKER} (see
+ * {@link PiiRedactionOptions.replaceEncodedBinary}).
  */
 export async function redactPIIFromExecution(
   payload: RedactablePayload,
@@ -230,29 +357,21 @@ export async function redactPIIFromExecution(
     value: payload[key],
   }))
 
-  const collected: string[] = []
-  let totalBytes = 0
-  for (const unit of units) {
-    transformUnit(unit.key, unit.value, (s) => {
-      collected.push(s)
-      totalBytes += Buffer.byteLength(s, 'utf8')
-      return s
-    })
-  }
-
-  if (collected.length === 0) return payload
+  const found = collectStrings((visit) => {
+    for (const unit of units) transformUnit(unit.key, unit.value, visit)
+  }, true)
+  if (found.collected.length === 0 && found.encodedBinaryCount === 0) return payload
 
-  const { masked, scrubbedCount } = await maskCollected(collected, options)
-
-  let index = 0
+  const { visit, scrubbedCount } = await substitutionVisitor(found.collected, options, true)
   const result: RedactablePayload = { ...payload }
   for (const unit of units) {
-    result[unit.key] = transformUnit(unit.key, unit.value, () => masked[index++])
+    result[unit.key] = transformUnit(unit.key, unit.value, visit)
   }
 
   logger.info('PII redaction completed', {
-    stringCount: collected.length,
-    totalBytes,
+    stringCount: found.collected.length,
+    totalBytes: found.collected.reduce((sum, s) => sum + Buffer.byteLength(s, 'utf8'), 0),
+    encodedBinaryCount: found.encodedBinaryCount,
     durationMs: Math.round(performance.now() - startedAt),
     scrubbedCount,
   })
diff --git a/apps/sim/package.json b/apps/sim/package.json
index 3f69e82872e..09cebb10844 100644
--- a/apps/sim/package.json
+++ b/apps/sim/package.json
@@ -25,6 +25,7 @@
     "test:scim:e2e": "bun run scripts/test-scim-e2e.ts",
     "test:sim-links:e2e": "bun run scripts/test-sim-links-e2e.ts",
     "test:people:e2e": "bun run scripts/test-people-e2e.ts",
+    "test:pii-redaction:e2e": "bun --no-env-file scripts/test-pii-redaction-e2e.ts",
     "test:browser-extension:e2e": "bun --no-env-file scripts/test-browser-extension-e2e.ts",
     "test:isolated-vm-remote:e2e": "bun run scripts/test-isolated-vm-remote-e2e.ts",
     "test:mobile:e2e": "bun --no-env-file scripts/test-mobile-e2e.ts",
diff --git a/apps/sim/scripts/test-pii-redaction-e2e.ts b/apps/sim/scripts/test-pii-redaction-e2e.ts
new file mode 100644
index 00000000000..dd7cf8c6efa
--- /dev/null
+++ b/apps/sim/scripts/test-pii-redaction-e2e.ts
@@ -0,0 +1,252 @@
+import assert from 'node:assert/strict'
+import { mkdir, writeFile } from 'node:fs/promises'
+import { dirname } from 'node:path'
+import { createLogger } from '@sim/logger'
+import { getErrorMessage } from '@sim/utils/errors'
+
+/**
+ * Proves PII redaction of oversized and encoded content against a running local
+ * app and a real Presidio container (`docker/pii.Dockerfile`, the app's `PII_URL`).
+ * The redaction library runs in this process the way the log persist path runs in
+ * a worker: every mask request crosses the app's `/api/guardrails/mask-batch` route
+ * over HTTP, and the route calls Presidio, whose spaCy NER refuses any single text
+ * over 1,000,000 characters.
+ *
+ * Checks, each recorded in the report:
+ * - Presidio masks a text over spaCy's limit: a name straddling an NER window edge,
+ *   a name at the very end, and a custom pattern that spans whitespace.
+ * - A persisted log holding a 5M-character base64 PDF and a 3M-character text that
+ *   names a person at the end: the PDF becomes the binary marker without reaching
+ *   Presidio, the name is masked by real NER, and no request fails.
+ * - The same content as live block output: nothing is exempt from masking, the
+ *   run is not aborted, and the name is masked.
+ *
+ * Exits non-zero when any check fails, after writing the report.
+ */
+const logger = createLogger('PiiRedactionE2E')
+
+const reportPath = process.env.PII_E2E_REPORT_PATH
+if (!reportPath) throw new Error('PII_E2E_REPORT_PATH must be explicitly provided')
+
+/** spaCy's default `max_length`, the per-text ceiling of one NER pass. */
+const SPACY_MAX_LENGTH = 1_000_000
+/** Where Presidio's first NER window ends (`NER_WINDOW_CHARS` in apps/pii/server.py). */
+const FIRST_NER_WINDOW_END = 100_000
+
+const FILLER = 'The quarterly report covers revenue, costs, and the hiring plan.\n'
+const NAME_SENTENCE = 'Please send questions to Ada Lovelace before Friday.'
+
+interface MaskRequest {
+  status: number
+  texts: number
+  totalChars: number
+}
+
+/** Every mask-batch request this process sends, observed at `fetch`, unmodified. */
+const maskRequests: MaskRequest[] = []
+
+const checks: {
+  name: string
+  status: 'passed' | 'failed'
+  durationMs: number
+  details?: Record<string, unknown>
+  error?: string
+}[] = []
+
+async function check(
+  name: string,
+  run: () => Promise<Record<string, unknown> | undefined>
+): Promise<void> {
+  const startedAt = performance.now()
+  const requestsBefore = maskRequests.length
+  try {
+    const details = await run()
+    checks.push({
+      name,
+      status: 'passed',
+      durationMs: Math.round(performance.now() - startedAt),
+      details: { ...details, maskRequests: maskRequests.slice(requestsBefore) },
+    })
+  } catch (error) {
+    checks.push({
+      name,
+      status: 'failed',
+      durationMs: Math.round(performance.now() - startedAt),
+      details: { maskRequests: maskRequests.slice(requestsBefore) },
+      error: getErrorMessage(error),
+    })
+  }
+}
+
+function assertMaskRequestsSucceeded(since: number): number {
+  const requests = maskRequests.slice(since)
+  assert(requests.length > 0, 'Expected mask-batch requests')
+  for (const request of requests) {
+    assert.equal(request.status, 200, `A mask-batch request answered ${request.status}`)
+  }
+  return requests.reduce((sum, request) => sum + request.totalChars, 0)
+}
+
+function assertNameMasked(masked: unknown, failedMarker: string): void {
+  assert.equal(typeof masked, 'string', 'Expected the long text back as a string')
+  const text = masked as string
+  assert(!text.includes(failedMarker), 'The long text was scrubbed instead of masked')
+  assert(
+    text.endsWith('Please send questions to <PERSON> before Friday.'),
+    'The name at the end of the long text was not masked as one <PERSON> span'
+  )
+}
+
+async function main(): Promise<void> {
+  const loopbackHosts = new Set(['localhost', '127.0.0.1', '[::1]'])
+  const baseUrl = new URL(process.env.PII_E2E_BASE_URL ?? '')
+  const presidioUrl = new URL(process.env.PII_E2E_PRESIDIO_URL ?? '')
+  assert(loopbackHosts.has(baseUrl.hostname), 'The app must use a loopback host')
+  assert(loopbackHosts.has(presidioUrl.hostname), 'Presidio must use a loopback host')
+  assert(
+    (process.env.INTERNAL_API_SECRET ?? '').length >= 32,
+    'INTERNAL_API_SECRET must match the running app'
+  )
+  process.env.INTERNAL_API_BASE_URL = baseUrl.origin
+
+  const realFetch = globalThis.fetch
+  globalThis.fetch = (async (input: RequestInfo | URL, init?: RequestInit) => {
+    const response = await realFetch(input, init)
+    const url = typeof input === 'string' ? input : input instanceof URL ? input.href : input.url
+    if (url.endsWith('/api/guardrails/mask-batch') && typeof init?.body === 'string') {
+      const { texts } = JSON.parse(init.body) as { texts: string[] }
+      maskRequests.push({
+        status: response.status,
+        texts: texts.length,
+        totalChars: texts.reduce((sum, text) => sum + text.length, 0),
+      })
+    }
+    return response
+  }) as typeof fetch
+
+  const {
+    BINARY_CONTENT_REDACTED_MARKER,
+    REDACTION_FAILED_MARKER,
+    redactObjectStrings,
+    redactPIIFromExecution,
+  } = await import('@/lib/logs/execution/pii-redaction')
+
+  /** 3.75MB of fixed non-text bytes behind a PDF header: 5,000,000 characters of base64. */
+  const pdfHeader = Buffer.from('%PDF-1.7\n')
+  const pdfBytes = Buffer.alloc(3_750_000)
+  pdfHeader.copy(pdfBytes)
+  for (let i = pdfHeader.length; i < pdfBytes.length; i++) pdfBytes[i] = (i * 7919 + 13) % 256
+  const pdfBase64 = pdfBytes.toString('base64')
+  const longText = `${FILLER.repeat(Math.ceil(3_000_000 / FILLER.length))}${NAME_SENTENCE}`
+
+  await check('Presidio masks a text longer than one NER pass allows', async () => {
+    const prose = (length: number) =>
+      FILLER.repeat(Math.ceil(length / FILLER.length)).slice(0, length)
+    const lead = 'Please send questions to '
+    const text =
+      prose(FIRST_NER_WINDOW_END - lead.length - 6) +
+      `${lead}Grace Hopper before Friday.\n` +
+      prose(400_000) +
+      'badge EMP 123456 issued\n' +
+      prose(SPACY_MAX_LENGTH) +
+      NAME_SENTENCE
+    const response = await realFetch(new URL('/redact_batch', presidioUrl), {
+      method: 'POST',
+      headers: { 'content-type': 'application/json' },
+      body: JSON.stringify({
+        texts: [text],
+        language: 'en',
+        entities: ['PERSON'],
+        patterns: [{ regex: 'EMP\\s+\\d{6}', replacement: 'EMPLOYEE_ID', name: 'Employee ID' }],
+      }),
+    })
+    const body = await response.text()
+    assert.equal(
+      response.status,
+      200,
+      `Presidio answered ${response.status}: ${body.slice(0, 200)}`
+    )
+    const [masked] = (JSON.parse(body) as { texts: string[] }).texts
+    assert(
+      masked.includes(`${lead}<PERSON> before Friday.`),
+      'The name across the first NER window edge was not masked as one <PERSON> span'
+    )
+    assert(masked.endsWith('to <PERSON> before Friday.'), 'The name at the end survived')
+    assert(!masked.includes('EMP 123456'), 'The custom pattern spanning whitespace survived')
+    assert(masked.includes('<EMPLOYEE_ID>'), 'Expected the custom replacement')
+    return { textChars: text.length }
+  })
+
+  await check('A persisted log replaces the PDF and masks the long text', async () => {
+    const since = maskRequests.length
+    const result = await redactPIIFromExecution(
+      {
+        traceSpans: [
+          {
+            blockId: 'write-file',
+            input: { content: pdfBase64, name: 'report.pdf' },
+            output: { text: longText },
+          },
+        ],
+        workflowInput: { file: `data:application/pdf;base64,${pdfBase64}` },
+      },
+      { entityTypes: ['PERSON'] }
+    )
+    const [span] = result.traceSpans as {
+      input: { content: string }
+      output: { text: string }
+    }[]
+    assert.equal(span.input.content, BINARY_CONTENT_REDACTED_MARKER)
+    assert.equal((result.workflowInput as { file: string }).file, BINARY_CONTENT_REDACTED_MARKER)
+    assertNameMasked(span.output.text, REDACTION_FAILED_MARKER)
+    const sentChars = assertMaskRequestsSucceeded(since)
+    assert(
+      sentChars < pdfBase64.length,
+      `Requests carried ${sentChars} characters: the PDF was sent`
+    )
+    return { sentChars, maskedTail: span.output.text.slice(-60) }
+  })
+
+  await check('Live block output is masked in full without aborting the run', async () => {
+    const since = maskRequests.length
+    const result = await redactObjectStrings(
+      { content: pdfBase64, text: longText },
+      { entityTypes: ['PERSON'], onFailure: 'throw' }
+    )
+    assertNameMasked(result.text, REDACTION_FAILED_MARKER)
+    const sentChars = assertMaskRequestsSucceeded(since)
+    assert(sentChars >= pdfBase64.length + longText.length, 'Live data skipped part of its masking')
+    return {
+      sentChars,
+      pdfUnchanged: result.content === pdfBase64,
+      maskedTail: result.text.slice(-60),
+    }
+  })
+}
+
+const startedAt = new Date().toISOString()
+try {
+  await main()
+} catch (error) {
+  checks.push({ name: 'setup', status: 'failed', durationMs: 0, error: getErrorMessage(error) })
+}
+const failed = checks.filter((entry) => entry.status === 'failed')
+const report = {
+  suite: 'pii-redaction-e2e',
+  startedAt,
+  finishedAt: new Date().toISOString(),
+  passed: checks.length - failed.length,
+  failed: failed.length,
+  checks,
+}
+await mkdir(dirname(reportPath), { recursive: true })
+await writeFile(reportPath, `${JSON.stringify(report, null, 2)}\n`)
+for (const entry of checks) {
+  logger.info('PII redaction check', {
+    name: entry.name,
+    status: entry.status,
+    durationMs: entry.durationMs,
+    error: entry.error,
+  })
+}
+process.exit(failed.length === 0 ? 0 : 1)