fix(middleware): scan PII emails in linear time (#8006)

* fix(middleware): scan PII emails in linear time

The email detector's pattern restarts at every position inside a
whitespace-free run and consumes the rest of the run before failing to
find '@', so a scan is quadratic in the run length. PiiRedactionMiddleware
re-redacts every user message on every model call, on the event loop, so
one pasted 100 KB hex dump stalled the Gateway for ~6 s per model call.

Scan from run starts only (a local part must reach the end of its run),
and match anchored where the previous match ended so glued addresses
such as 'a@b.cc+c@d.org' keep both matches, exactly as before.

* docs(changelog): reference #8006 for the linear PII email scan

* fix(middleware): group the embedded email pattern and pin every detector's linearity

Wrap the pattern behind the run-start lookbehind in a group so a future
top-level alternative in _EMAIL_PATTERN cannot escape the lookbehind, and
add one whitespace-free payload per other detector to the linearity test
so the AGENTS.md rule is enforced for all of them.

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

CHANGELOG.md

@@ -28,6 +28,13 @@ This section accumulates work toward the **2.3.0** milestone
 
 ### Fixed
 
+- **middleware:** With PII redaction enabled, a long run of text without
+  spaces (such as a pasted hex dump) no longer stalls the Gateway. The email
+  detector was quadratic in the length of such a run, and redaction re-scans
+  every user message on every model call, so one 100 KB paste blocked every
+  thread for about 6 s per model call. The scan is now linear, and it redacts
+  exactly the same addresses as before. ([#8006])
+  
 - **subagents:** Managed subagent creation now rejects Windows reserved device
   names (`CON`, `PRN`, `AUX`, `NUL`, `COM1-9`, `LPT1-9`) at the shared
   `normalize_managed_subagent_name(..., for_creation=True)` validator and
@@ -16136,3 +16143,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
 [#7958]: https://github.com/bytedance/deer-flow/pull/7958
 [#7981]: https://github.com/bytedance/deer-flow/pull/7981
 [#7993]: https://github.com/bytedance/deer-flow/pull/7993
+[#8006]: https://github.com/bytedance/deer-flow/pull/8006

CHANGELOG_zh.md

@@ -26,6 +26,10 @@
 
 ### 修复
 
+- **中间件：** 启用 PII 脱敏时，不含空格的长文本（例如粘贴的十六进制转储）不再让 Gateway
+  卡住。此前邮箱检测器的耗时与这类文本长度呈平方关系，而脱敏会在每次模型调用时重新扫描全部
+  用户消息，因此一次 100 KB 的粘贴会让每次模型调用阻塞所有线程约 6 秒。现在扫描为线性耗时，
+  且脱敏的地址与此前完全相同。([#8006])
 - **调度器：** 一个已保存的计划无法再计算的定时任务，不再让其他所有任务停止运行。此前
   已保存的时区或 cron 表达式失效时（例如升级后时区数据库中缺少该时区，或新版 croniter
   不再接受该表达式），在调度器轮询中计算其下一次运行时间会抛出异常；其启动租约过期后，
@@ -12817,3 +12821,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子
 [#7958]: https://github.com/bytedance/deer-flow/pull/7958
 [#7981]: https://github.com/bytedance/deer-flow/pull/7981
 [#7993]: https://github.com/bytedance/deer-flow/pull/7993
+[#8006]: https://github.com/bytedance/deer-flow/pull/8006

backend/packages/harness/deerflow/agents/middlewares/AGENTS.md

@@ -101,7 +101,7 @@ No match: skip inspection, log ID. Tests: `test_read_mark_tool_call_correlation.
    ordered by application — the last entry produced the final visible bytes — so an
    observer classifies raw→visible transforms from facts rather than by sniffing
    output wording.
-4. **PiiRedactionMiddleware** - *(optional, `pii_redaction.enabled`, default off, #3190)* Rewrites PII in genuine user messages (`wrap_model_call`) and remote-content tool results (`wrap_tool_call`, the tool-result allowlist) to irreversible value-derived placeholders. Deterministic regex detectors only. Innermost Layer-1 wrapper; compaction input, summaries and title input use `redact_text`. Memory admission redacts content, parsed/invalid/legacy call arguments and supported provenance before backend filtering. Disabled PII skips conversion; tool arguments track changes without deep equality. JSON content blocks remain recursive. SDK tests: `tests/test_memory_admission_depth.py`.
+4. **PiiRedactionMiddleware** - *(optional, `pii_redaction.enabled`, default off, #3190)* Rewrites PII in genuine user messages (`wrap_model_call`) and remote-content tool results (`wrap_tool_call`, the tool-result allowlist) to irreversible value-derived placeholders. Deterministic regex detectors only; each must stay linear on whitespace-free text, and the email scan (`_iter_email_matches`) must yield exactly `_EMAIL_PATTERN.sub`'s matches, including glued addresses. Innermost Layer-1 wrapper; compaction input, summaries and title input use `redact_text`. Memory admission redacts content, parsed/invalid/legacy call arguments and supported provenance before backend filtering. Disabled PII skips conversion; tool arguments track changes without deep equality. JSON content blocks remain recursive. SDK tests: `tests/test_memory_admission_depth.py`.
 
 5. **ThreadDataMiddleware** - Creates per-thread directories under the user's isolation scope (`backend/.deer-flow/users/{user_id}/threads/{thread_id}/user-data/{workspace,uploads,outputs}`); resolves identity via `resolve_runtime_user_id(runtime)`, including Gateway runtime context and standalone LangGraph Server auth, then falls back to the request ContextVar / `"default"`
 6. **UploadsMiddleware** - Lead-only current uploads: exact basenames in runtime owner's bucket; skip malformed references/probe errors (debug-log errors). Invalid/display-overflowing sizes use `0`; render >=1 TB compactly without changing stored sizes. Copy messages with metadata, including `response_metadata`.

backend/packages/harness/deerflow/agents/middlewares/pii_redaction_middleware.py

@@ -59,7 +59,7 @@ import hashlib
 import hmac
 import logging
 import re
-from collections.abc import Awaitable, Callable, Sequence
+from collections.abc import Awaitable, Callable, Iterator, Sequence
 from copy import deepcopy
 from dataclasses import dataclass
 from dataclasses import replace as dc_replace
@@ -155,9 +155,51 @@ class _Detector:
     name: str
     pattern: re.Pattern[str]
     validator: Callable[[str], bool] | None = None
+    # Replaces ``pattern.finditer`` when the pattern alone is too slow to scan
+    # with; it must yield exactly the matches ``pattern.sub`` would replace.
+    finditer: Callable[[str], Iterator[re.Match[str]]] | None = None
+
+    def sub(self, repl: Callable[[re.Match[str]], str], text: str) -> str:
+        if self.finditer is None:
+            return self.pattern.sub(repl, text)
+        pieces: list[str] = []
+        last = 0
+        for match in self.finditer(text):
+            pieces.append(text[last : match.start()])
+            pieces.append(repl(match))
+            last = match.end()
+        if not pieces:
+            return text
+        pieces.append(text[last:])
+        return "".join(pieces)
 
 
-_EMAIL_PATTERN = re.compile(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9-]+(?:\.[A-Za-z0-9-]+)*\.[A-Za-z]{2,}")
+_EMAIL_LOCAL_CHARS = "A-Za-z0-9._%+-"
+_EMAIL_PATTERN = re.compile(rf"[{_EMAIL_LOCAL_CHARS}]+@[A-Za-z0-9-]+(?:\.[A-Za-z0-9-]+)*\.[A-Za-z]{{2,}}")
+# Scanning with _EMAIL_PATTERN is quadratic in the longest local-part run: each
+# start inside a whitespace-free run (a hex dump, a long token) consumes the
+# rest of the run before failing to find "@" (100 KB took ~6 s, on the event
+# loop, on every model call). The local part must reach the run's end, so a
+# match exists anywhere in a run only if it exists at the run's start, which is
+# all this spelling tries.
+# The group keeps the lookbehind on every alternative should the pattern grow one.
+_EMAIL_RUN_START_PATTERN = re.compile(rf"(?<![{_EMAIL_LOCAL_CHARS}])(?:{_EMAIL_PATTERN.pattern})")
+
+
+def _iter_email_matches(text: str) -> Iterator[re.Match[str]]:
+    """Yield the matches ``_EMAIL_PATTERN.sub`` replaces, in linear time."""
+    pos = 0
+    while match := _EMAIL_RUN_START_PATTERN.search(text, pos):
+        yield match
+        pos = match.end()
+        # ``sub`` also resumes where the previous match ended, which can be
+        # inside a run ("a@b.cc+c@d.org" holds two matches). The run-start
+        # lookbehind rejects that position, so match there anchored, after
+        # every match. A failure there fails for the rest of that run too.
+        while glued := _EMAIL_PATTERN.match(text, pos):
+            yield glued
+            pos = glued.end()
+
 
 _API_KEY_PATTERN = re.compile(
     r"\b(?:"
@@ -189,7 +231,7 @@ _NATIONAL_ID_PATTERN = re.compile(
 )
 
 _DETECTORS: tuple[_Detector, ...] = (
-    _Detector("email", _EMAIL_PATTERN),
+    _Detector("email", _EMAIL_PATTERN, finditer=_iter_email_matches),
     _Detector("api_key", _API_KEY_PATTERN),
     _Detector("national_id", _NATIONAL_ID_PATTERN, _national_id_valid),
     _Detector("credit_card", _CREDIT_CARD_PATTERN, _luhn_valid),
@@ -284,7 +326,7 @@ class _Redactor:
 
     def redact(self, text: str) -> str:
         for detector in self._detectors:
-            text = detector.pattern.sub(self._replacer(detector), text)
+            text = detector.sub(self._replacer(detector), text)
         return text
 
     def _replacer(self, detector: _Detector) -> Callable[[re.Match[str]], str]:

backend/tests/test_pii_redaction_middleware.py

@@ -6,7 +6,9 @@ that the rewrite is request-scoped without mutating the original request or
 messages, the tool-boundary allowlist, and the pinned detector registry.
 """
 
+import random
 import re
+import time
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock, Mock
 
@@ -20,7 +22,9 @@ from pydantic import Field, ValidationError
 
 from deerflow.agents.middlewares.pii_redaction_middleware import (
     _DETECTORS,
+    _EMAIL_PATTERN,
     PiiRedactionMiddleware,
+    _iter_email_matches,
     _redact_content,
     _Redactor,
     redact_text,
@@ -122,6 +126,58 @@ class TestDetectors:
         assert messages[0].content == f"{EMAIL_ALICE} here"
         assert messages[1].content == f"reply to {EMAIL_ALICE}"
 
+    def test_glued_emails_are_both_redacted(self):
+        # Each second address starts inside the run the first one ended in.
+        messages, _ = _run_model_call(
+            _make_middleware(),
+            [HumanMessage("cc alice@example.com+bob@example.org, alice@example.com_carol@example.com")],
+        )
+        bob_glued = redact_text("+bob@example.org", _PII_CFG)
+        carol_glued = redact_text("_carol@example.com", _PII_CFG)
+        assert messages[0].content == f"cc {EMAIL_ALICE}{bob_glued}, {EMAIL_ALICE}{carol_glued}"
+
+    def test_email_scan_matches_the_reference_pattern(self):
+        rng = random.Random(3190)
+        corpus = ["a@b.cc+c@d.org", "a@b.cc+c@d.org_e@f.gh-g@h.ij", "x.y@a.b.cc.d@e.ff", "a@b.c", "@@a@b.cc", "a" * 64 + "@b.cc"]
+        for alphabet in ("ab.%+-_@xy1. Z", "a@.b-+", "aA1@.-_%+ ,;\n"):
+            corpus += ["".join(rng.choice(alphabet) for _ in range(rng.randint(0, 40))) for _ in range(5000)]
+        for text in corpus:
+            expected = [match.span() for match in _EMAIL_PATTERN.finditer(text)]
+            assert [match.span() for match in _iter_email_matches(text)] == expected, text
+
+    @pytest.mark.parametrize(
+        "make_payload",
+        [
+            pytest.param(lambda n: "a" * n, id="letters"),
+            pytest.param(lambda n: ("0123456789abcdef" * n)[:n], id="hex"),
+            # One worst shape per other detector, so the AGENTS.md rule that
+            # every detector stays linear is pinned, not only stated.
+            pytest.param(lambda n: ("sk-ghp_xoxb-AIzaAKIA" * n)[:n], id="api-key-prefixes"),
+            pytest.param(lambda n: "1" * n, id="digits"),
+            pytest.param(lambda n: ("1 1-1." * n)[:n], id="digit-separators"),
+            pytest.param(lambda n: ("+1 (212) " * n)[:n], id="phone-prefixes"),
+            pytest.param(lambda n: ("ABCD123456" * n)[:n], id="rfc-letters-digits"),
+        ],
+    )
+    def test_redaction_stays_linear_on_whitespace_free_text(self, make_payload):
+        # The email scan was quadratic in the longest local-part run: 100 KB
+        # of hex took ~6 s on the event loop, on every model call. Assert the
+        # shape, not an absolute budget (see test_skill_review_core.py): the
+        # quadratic path takes ~4x per doubling, the linear one ~2x.
+        def timed(payload):
+            def once():
+                started = time.perf_counter()
+                redact_text(payload, _PII_CFG)
+                return time.perf_counter() - started
+
+            return min(once() for _ in range(5))
+
+        small_elapsed = timed(make_payload(32768))
+        large_elapsed = timed(make_payload(65536))
+
+        assert small_elapsed < 1.0, f"redaction took {small_elapsed:.2f}s"
+        assert large_elapsed / small_elapsed < 3, f"redaction looks superlinear: 32K took {small_elapsed:.4f}s, 64K took {large_elapsed:.4f}s"
+
     def test_openai_style_api_key_redacted(self):
         messages, _ = _run_model_call(
             _make_middleware(),