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>

jobs/mail-body-ingest/src/mail_body_ingest/ingest.py

@@ -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=
 '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
 the first two):
 
@@ -88,6 +93,7 @@ import asyncpg
 import structlog
 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 strip_nul as _strip_nul
 from kb_retrieval.embed import (
     DEFAULT_MODEL,
     DEFAULT_OLLAMA_URL,
@@ -149,6 +155,13 @@ _QUOTE_LINE_RE = re.compile(r"^\s*>")
 _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):
     """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
@@ -316,13 +329,13 @@ def extract_threading(msg: email.message.Message) -> dict:
 
     in_reply_to = None
     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:
             in_reply_to = m.group(0)[1:-1]
 
     references: list[str] = []
     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])
 
     return {"type": "threading", "in_reply_to": in_reply_to, "references": references}
@@ -418,6 +431,9 @@ def _new_stats() -> dict:
         "chunks_conflict_skipped": 0,
         "chunks_errors": 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
         # 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.
@@ -570,6 +586,15 @@ async def run(
                     body = extract_body(msg)
                     clean_body, stripped = strip_quotes(body)
                     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)
                     threading_patch = extract_threading(msg)
                 except Exception:

jobs/mail-body-ingest/tests/test_ingest.py

@@ -442,15 +442,19 @@ def _write_eml(archive_root, rel_path: str, raw: bytes) -> None:
     dest.write_bytes(raw)
 
 
+def _patch_run(monkeypatch, conn, ollama_session) -> None:
+    async def _fake_connect(dsn):
+        return conn
+    monkeypatch.setattr("mail_body_ingest.ingest.asyncpg.connect", _fake_connect)
+
+    def _fake_session_factory(*args, **kwargs):
+        return ollama_session
+    monkeypatch.setattr("mail_body_ingest.ingest.aiohttp.ClientSession", _fake_session_factory)
+
+
 class TestRun:
     def _patch(self, monkeypatch, conn, ollama_session):
-        async def _fake_connect(dsn):
+        _patch_run(monkeypatch, conn, ollama_session)
-            return conn
-        monkeypatch.setattr("mail_body_ingest.ingest.asyncpg.connect", _fake_connect)
-
-        def _fake_session_factory(*args, **kwargs):
-            return ollama_session
-        monkeypatch.setattr("mail_body_ingest.ingest.aiohttp.ClientSession", _fake_session_factory)
 
     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(
@@ -772,3 +776,83 @@ class TestRun:
         assert stats["threading_already_present"] == 1
         assert stats["threading_updated"] == 0
         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"]

packages/kb-mail/src/kb_mail/text.py

@@ -18,3 +18,18 @@ def sanitize_surrogates(value: Optional[str]) -> Optional[str]:
     if value is None:
         return None
     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

packages/kb-mail/tests/test_text.py

@@ -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
 
 import json
 
-from kb_mail.text import sanitize_surrogates
+from kb_mail.text import sanitize_surrogates, strip_nul
 
 
 def test_none_stays_none():
@@ -30,3 +30,27 @@ def test_lone_surrogate_degraded_and_json_safe():
 def test_replacement_char_preserved():
     # U+FFFD is already valid UTF-8 and must survive untouched.
     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"