fix(mail-body-ingest): usuwaj NUL (0x00) z tekstu przed chunkowaniem i embedem

Chunki z NUL wywalaly insert do postgresa (asyncpg CharacterNotInRepertoireError:
invalid byte sequence for encoding "UTF8": 0x00) - 3 przypadki na mailach z 2007
w plastrze offset 50000 Etapu B. sanitize_surrogates tego nie lapie, bo NUL to
poprawny code point, nie osierocony surogat.

Strip dzieje sie zaraz po strip_quotes, czyli PRZED chunk_text i przed wywolaniem
Ollamy - dzieki temu embedding liczy sie z dokladnie tego samego stringa, ktory
trafia do document_chunk.text. Sanityzacja dopiero przy insercie zostawialaby
wektor opisujacy tekst, ktorego DB nigdy nie zobaczyla.

Skala widoczna w progress/summary: nul_bytes_stripped (ile znakow) oraz
mails_nul_sanitized (ilu maili dotyczylo). Zadne z nich nie wchodzi do rownan
balansu i nie wplywa na exit code - to normalizacja, nie blad.

Ten sam strip w extract_threading (In-Reply-To / References): json.dumps zamienia
NUL na escape u0000, ktory jsonb odrzuca tym samym bledem.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
oskar 2026-08-06 11:14:09 +02:00
parent 67aa09276d
commit 4ec0b7876c
4 changed files with 159 additions and 11 deletions

View file

@ -42,6 +42,11 @@ Newsletter chunks (`List-Unsubscribe`/`List-Id`/`Precedence: bulk|list`) are ins
stays in the DB (reversible: `UPDATE ... SET excluded_reason=NULL WHERE excluded_reason= stays in the DB (reversible: `UPDATE ... SET excluded_reason=NULL WHERE excluded_reason=
'newsletter'` + a re-embed run un-flags them later, per plan Decyzja 4). 'newsletter'` + a re-embed run un-flags them later, per plan Decyzja 4).
Body text is NUL-stripped (`kb_mail.text.strip_nul`) right after quote-stripping, i.e. before
chunking and before the embed call postgres rejects 0x00 in `text` outright, and sanitizing at
INSERT time instead would leave the stored vector describing a string the DB never got. Scale is
visible in progress/summary as `nul_bytes_stripped` / `mails_nul_sanitized`.
Ollama-offline tolerance, in three layers (`kb_retrieval.embed.embed_batch_resilient` implements Ollama-offline tolerance, in three layers (`kb_retrieval.embed.embed_batch_resilient` implements
the first two): the first two):
@ -88,6 +93,7 @@ import asyncpg
import structlog import structlog
from kb_mail.chunking import OVERLAP_CHARS, TARGET_CHARS, chunk_text from kb_mail.chunking import OVERLAP_CHARS, TARGET_CHARS, chunk_text
from kb_mail.text import sanitize_surrogates as _sanitize from kb_mail.text import sanitize_surrogates as _sanitize
from kb_mail.text import strip_nul as _strip_nul
from kb_retrieval.embed import ( from kb_retrieval.embed import (
DEFAULT_MODEL, DEFAULT_MODEL,
DEFAULT_OLLAMA_URL, DEFAULT_OLLAMA_URL,
@ -149,6 +155,13 @@ _QUOTE_LINE_RE = re.compile(r"^\s*>")
_ID_TOKEN_RE = re.compile(r"<[^<>]+>") _ID_TOKEN_RE = re.compile(r"<[^<>]+>")
def _normalize(value: Optional[str]) -> Optional[str]:
"""Header-text normalization for anything headed for postgres: lone surrogates degraded to
'?', NUL dropped. Body text takes the same two steps, but counts the NULs it removes
(`nul_bytes_stripped`) instead of dropping them silently see `run`."""
return _strip_nul(_sanitize(value))
class _MailHTMLTextExtractor(HTMLParser): class _MailHTMLTextExtractor(HTMLParser):
"""HTML -> text: skips `style`/`script`/`head` content and `blockquote`/`div.gmail_quote` """HTML -> text: skips `style`/`script`/`head` content and `blockquote`/`div.gmail_quote`
subtrees (Decision 2c quoted reply chains in HTML mail), inserts newlines at block-level subtrees (Decision 2c quoted reply chains in HTML mail), inserts newlines at block-level
@ -316,13 +329,13 @@ def extract_threading(msg: email.message.Message) -> dict:
in_reply_to = None in_reply_to = None
if in_reply_to_raw: if in_reply_to_raw:
m = _ID_TOKEN_RE.search(_sanitize(str(in_reply_to_raw)) or "") m = _ID_TOKEN_RE.search(_normalize(str(in_reply_to_raw)) or "")
if m: if m:
in_reply_to = m.group(0)[1:-1] in_reply_to = m.group(0)[1:-1]
references: list[str] = [] references: list[str] = []
if references_raw: if references_raw:
for m in _ID_TOKEN_RE.finditer(_sanitize(str(references_raw)) or ""): for m in _ID_TOKEN_RE.finditer(_normalize(str(references_raw)) or ""):
references.append(m.group(0)[1:-1]) references.append(m.group(0)[1:-1])
return {"type": "threading", "in_reply_to": in_reply_to, "references": references} return {"type": "threading", "in_reply_to": in_reply_to, "references": references}
@ -418,6 +431,9 @@ def _new_stats() -> dict:
"chunks_conflict_skipped": 0, "chunks_conflict_skipped": 0,
"chunks_errors": 0, "chunks_errors": 0,
"quoted_chars_stripped_total": 0, "quoted_chars_stripped_total": 0,
# Normalization counters — how much text the run had to repair, not a failure signal.
"nul_bytes_stripped": 0,
"mails_nul_sanitized": 0,
# Diagnostics — none of these participate in the balance equations. `embed_calls` counts # Diagnostics — none of these participate in the balance equations. `embed_calls` counts
# flushed batches, `embed_requests_total` the HTTP requests they actually cost (retries # flushed batches, `embed_requests_total` the HTTP requests they actually cost (retries
# and bisection included), so the ratio is the health signal: 1.0 = a clean run. # and bisection included), so the ratio is the health signal: 1.0 = a clean run.
@ -570,6 +586,15 @@ async def run(
body = extract_body(msg) body = extract_body(msg)
clean_body, stripped = strip_quotes(body) clean_body, stripped = strip_quotes(body)
stats["quoted_chars_stripped_total"] += stripped stats["quoted_chars_stripped_total"] += stripped
# Before chunking, so the text that gets embedded is byte-for-byte the text
# that gets inserted. Sanitizing at INSERT time instead would leave the
# vector describing a string the DB never stored.
nul_count = clean_body.count("\x00")
if nul_count:
clean_body = _strip_nul(clean_body)
stats["nul_bytes_stripped"] += nul_count
stats["mails_nul_sanitized"] += 1
_log.info("nul_bytes_stripped", envelope_id=envelope_id, count=nul_count)
newsletter = is_newsletter(msg) newsletter = is_newsletter(msg)
threading_patch = extract_threading(msg) threading_patch = extract_threading(msg)
except Exception: except Exception:

View file

@ -442,8 +442,7 @@ def _write_eml(archive_root, rel_path: str, raw: bytes) -> None:
dest.write_bytes(raw) dest.write_bytes(raw)
class TestRun: def _patch_run(monkeypatch, conn, ollama_session) -> None:
def _patch(self, monkeypatch, conn, ollama_session):
async def _fake_connect(dsn): async def _fake_connect(dsn):
return conn return conn
monkeypatch.setattr("mail_body_ingest.ingest.asyncpg.connect", _fake_connect) monkeypatch.setattr("mail_body_ingest.ingest.asyncpg.connect", _fake_connect)
@ -452,6 +451,11 @@ class TestRun:
return ollama_session return ollama_session
monkeypatch.setattr("mail_body_ingest.ingest.aiohttp.ClientSession", _fake_session_factory) monkeypatch.setattr("mail_body_ingest.ingest.aiohttp.ClientSession", _fake_session_factory)
class TestRun:
def _patch(self, monkeypatch, conn, ollama_session):
_patch_run(monkeypatch, conn, ollama_session)
async def test_dry_run_counts_without_calling_ollama_or_db(self, tmp_path, monkeypatch): async def test_dry_run_counts_without_calling_ollama_or_db(self, tmp_path, monkeypatch):
_write_eml(tmp_path, "gmail/2025/08/m1@x.eml", _plain_eml( _write_eml(tmp_path, "gmail/2025/08/m1@x.eml", _plain_eml(
{"From": "a@b.com", "Subject": "hi", "Date": "Fri, 01 Aug 2025 10:00:00 +0000"}, "hello there" {"From": "a@b.com", "Subject": "hi", "Date": "Fri, 01 Aug 2025 10:00:00 +0000"}, "hello there"
@ -772,3 +776,83 @@ class TestRun:
assert stats["threading_already_present"] == 1 assert stats["threading_already_present"] == 1
assert stats["threading_updated"] == 0 assert stats["threading_updated"] == 0
assert conn.executemany_calls == [] assert conn.executemany_calls == []
class TestNulBytes:
"""2007-era mail in the Etap B run (slice offset 50000) carried stray NULs; postgres answered
with CharacterNotInRepertoireError ('invalid byte sequence for encoding UTF8: 0x00') and the
chunks were lost. The strip happens before chunking, so the embedded text and the stored text
are the same string."""
async def test_body_with_nul_embeds_and_inserts_clean_text(self, tmp_path, monkeypatch):
_write_eml(tmp_path, "gmail/2007/03/nul@x.eml", _plain_eml(
{"From": "a@b.com", "Subject": "stary mail", "Date": "Thu, 01 Mar 2007 10:00:00 +0000"},
"przed\x00 i\x00 po",
))
row = _env_row("nul@x", "stary mail", ts=datetime(2007, 3, 1, tzinfo=timezone.utc))
row["raw_ref"] = "gmail/2007/03/nul@x.eml"
conn = _FakeConn(envelopes=[row])
ollama = _FakeOllamaSession(dim=1024)
_patch_run(monkeypatch, conn, ollama)
stats = await run(dsn="postgresql://fake", archive_root=tmp_path, apply=True)
assert stats["chunks_inserted"] == 1
assert stats["chunks_errors"] == 0
assert stats["nul_bytes_stripped"] == 2
assert stats["mails_nul_sanitized"] == 1
embedded = ollama.requests[0]["json"]["input"]
inserted_text = conn.execute_calls[0][2]
assert "\x00" not in inserted_text
assert [t for t in embedded if "\x00" in t] == []
# Same string embedded as stored -- that is the whole point of stripping this early.
assert embedded == [inserted_text]
assert "przed i po" in inserted_text
async def test_ordinary_body_is_untouched_and_counters_stay_zero(self, tmp_path, monkeypatch):
_write_eml(tmp_path, "gmail/2025/08/m1@x.eml", _plain_eml(
{"From": "a@b.com", "Subject": "hi", "Date": "Fri, 01 Aug 2025 10:00:00 +0000"},
"Zwykły tekst\n\nz akapitem i tabulatorem\ti końcem.",
))
conn = _FakeConn(envelopes=[_env_row("m1@x", "hi")])
ollama = _FakeOllamaSession(dim=1024)
_patch_run(monkeypatch, conn, ollama)
stats = await run(dsn="postgresql://fake", archive_root=tmp_path, apply=True)
assert stats["nul_bytes_stripped"] == 0
assert stats["mails_nul_sanitized"] == 0
assert stats["chunks_inserted"] == 1
inserted_text = conn.execute_calls[0][2]
assert "Zwykły tekst\n\nz akapitem i tabulatorem\ti końcem." in inserted_text
async def test_body_of_only_nuls_counts_as_empty_not_as_a_chunk(self, tmp_path, monkeypatch):
_write_eml(tmp_path, "gmail/2007/03/nul@x.eml", _plain_eml(
{"From": "a@b.com", "Subject": "pusty", "Date": "Thu, 01 Mar 2007 10:00:00 +0000"},
"\x00\x00\x00",
))
row = _env_row("nul@x", "pusty", ts=datetime(2007, 3, 1, tzinfo=timezone.utc))
row["raw_ref"] = "gmail/2007/03/nul@x.eml"
conn = _FakeConn(envelopes=[row])
ollama = _FakeOllamaSession(dim=1024)
_patch_run(monkeypatch, conn, ollama)
stats = await run(dsn="postgresql://fake", archive_root=tmp_path, apply=True)
assert stats["nul_bytes_stripped"] == 3
assert stats["body_empty"] == 1
assert stats["mails_chunked"] == 0
assert stats["chunks_total"] == 0
assert ollama.requests == []
def test_threading_ids_are_nul_stripped_before_jsonb(self):
# Same crash class, different column: json.dumps turns a NUL into \\u0000, which postgres
# jsonb rejects too. extract_threading normalizes before the id tokens are cut out.
msg = parse_message(_plain_eml(
{"From": "a@b.com", "In-Reply-To": "<msg\x001@x.com>",
"References": "<msg\x000@x.com>"}, "hi",
))
threading = extract_threading(msg)
assert threading["in_reply_to"] == "msg1@x.com"
assert threading["references"] == ["msg0@x.com"]

View file

@ -18,3 +18,18 @@ def sanitize_surrogates(value: Optional[str]) -> Optional[str]:
if value is None: if value is None:
return None return None
return value.encode("utf-8", errors="replace").decode("utf-8") return value.encode("utf-8", errors="replace").decode("utf-8")
def strip_nul(value: Optional[str]) -> Optional[str]:
"""Drop NUL (U+0000) — the one character python happily encodes as UTF-8 but postgres
rejects outright: `invalid byte sequence for encoding "UTF8": 0x00` on a `text` insert,
`unsupported Unicode escape sequence` on jsonb. `sanitize_surrogates` does not catch it
(NUL is a valid code point, not a lone surrogate), so mail bodies carrying stray NULs
observed on 2007-era mail in the Etap B run kill the INSERT for the whole chunk.
Only NUL. Other C0 controls are legal in postgres `text` and stripping them would silently
rewrite body content for no gain.
"""
if value is None:
return None
return value.replace("\x00", "") if "\x00" in value else value

View file

@ -1,9 +1,9 @@
"""Unit tests for kb_mail.text.sanitize_surrogates.""" """Unit tests for kb_mail.text.sanitize_surrogates / strip_nul."""
from __future__ import annotations from __future__ import annotations
import json import json
from kb_mail.text import sanitize_surrogates from kb_mail.text import sanitize_surrogates, strip_nul
def test_none_stays_none(): def test_none_stays_none():
@ -30,3 +30,27 @@ def test_lone_surrogate_degraded_and_json_safe():
def test_replacement_char_preserved(): def test_replacement_char_preserved():
# U+FFFD is already valid UTF-8 and must survive untouched. # U+FFFD is already valid UTF-8 and must survive untouched.
assert sanitize_surrogates("bad<EFBFBD>id") == "bad<EFBFBD>id" assert sanitize_surrogates("bad<EFBFBD>id") == "bad<EFBFBD>id"
def test_sanitize_surrogates_does_not_remove_nul():
# Why strip_nul has to exist: NUL is a valid code point, so it survives the
# encode/decode round-trip untouched -- and then postgres rejects the insert.
assert sanitize_surrogates("a\x00b") == "a\x00b"
def test_strip_nul_none_stays_none():
assert strip_nul(None) is None
def test_strip_nul_removes_every_occurrence():
assert strip_nul("a\x00b\x00\x00c") == "abc"
def test_strip_nul_leaves_ordinary_text_untouched():
text = "Zwykły tekst\nz nową linią\ti tabulatorem\r\n"
assert strip_nul(text) is text # same object: no NUL, no rewrite
def test_strip_nul_keeps_other_c0_controls():
# Postgres accepts these in `text`; silently rewriting mail bodies is not the job here.
assert strip_nul("a\x01b\x0cc\x1fd") == "a\x01b\x0cc\x1fd"