feat(kb): faza 3 krok 1 — porządki po pilocie retrieval (śmieć + dedup + klucz modelu)

Migracja 003 (UNIQUE+model, excluded_reason) zastosowana na żywej bazie kb-postgres@PIHA
(2683 chunki, bez DELETE). Kalibracja heurystyki ocr_junk (3 sygnały z planu §3.1) ujawniła
realną sprzeczność z planem: sygnał 1 (dowolny znak kontrolny) fałszywie łapał paperless:119
(wymagany aktywny) i legalny angielski tekst — próg doprecyzowany do >=5 wystąpień na
podstawie rozkładu na korpusie. 8 chunków oflagowanych ocr_junk po odjęciu potwierdzonych
fałszywych alarmów (kalendarz, mikro-fragmenty referencyjne).

Dedup: SQL z planu (exact content hash) znalazł 2 pary, ale nie wykrył znanego z pilota
duplikatu paperless:14≡74 (różne OCR, 99.2% chunków identycznych treściowo) — dodany fuzzy
check na potwierdzenie. Odrzucono 3 kandydatury o wysokim nakładaniu jako różne
wersje/typy dokumentów dzielące boilerplate PZU, nie duplikaty. 130 chunków oflagowanych
duplicate + entities[duplicate_of] na 3 kopertach.

chunk_embed.py: heurystyka is_ocr_junk() przed embedem (junk -> insert bez wywołania Ollamy,
embedding=NULL), ON CONFLICT rozszerzony o model, nowy licznik chunks_junk_flagged w bilansie,
testy (kody kreskowe, mojibake nie-junk, dot-leader, idempotencja, model w kluczu konfliktu).

Weryfikacja: 7 zapytań eval-setu z WHERE excluded_reason IS NULL — żadne trafienie nie
degraduje, kontrole negatywne bez zmian (>0.55), śmieć zniknął z top-5 zapytania 2,
paperless:119 pozostał aktywnym trafieniem. Bilans: 2545 aktywne / 130 duplicate / 8 ocr_junk.

Co najmniej 3 decyzje wymagały zatrzymania i potwierdzenia z Oskarem w sesji (kalibracja
progu sygnału 1, dołączenie 14≡74 mimo braku exact-hash matcha, odrzucenie 3 fałszywych
kandydatur dedup) — plan przewidywał, że heurystyka będzie się mylić; wszystkie decyzje
udokumentowane w transkrypcie sesji.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

jobs/documents-ingest/src/documents_ingest/chunk_embed.py

@@ -30,15 +30,16 @@ DSN can come from KB_DSN, Ollama URL from OLLAMA_URL (default http://localhost:1
 
 Idempotency: a pre-fetched set of existing (envelope_id, chunk_index) pairs for this
 `model` skips chunks already embedded — no re-embedding, no wasted Ollama calls on rerun.
-`insert_chunk`'s own `ON CONFLICT (envelope_id, chunk_index) DO NOTHING` is the second line
+`insert_chunk`'s own `ON CONFLICT (envelope_id, chunk_index, model) DO NOTHING` is the
-of defense; its command tag is checked so a silently-skipped row is counted as
+second line of defense; its command tag is checked so a silently-skipped row is counted as
-`chunks_conflict_skipped`, never miscounted as `chunks_inserted`. Note: that UNIQUE
+`chunks_conflict_skipped`, never miscounted as `chunks_inserted`.
-constraint is on (envelope_id, chunk_index) only, not including `model` — re-embedding the
+
-same pilot with a *different* model would hit this path and its embedding would be
+OCR-junk filter (module 5, phase 3, plan §3.1): `is_ocr_junk()` screens each chunk before
-discarded (a real chunk of work wasted, correctly reported, but not persisted). Out of
+embedding. A junk chunk is INSERTed with `excluded_reason='ocr_junk'` and `embedding=NULL`
-scope for this pilot (single model, bge-m3); flagged as a follow-up if/when a second model
+— no Ollama call, no HNSW entry — and counted as `chunks_junk_flagged`. The three signals
-is ever indexed (the real fix is `UNIQUE (envelope_id, chunk_index, model)` at the schema
+and their thresholds are calibrated in-code (see `is_ocr_junk` docstring) against the
-layer).
+2683-chunk pilot corpus; retrieval and all downstream consumers filter on
+`WHERE excluded_reason IS NULL`.
 """
 from __future__ import annotations
 
@@ -75,11 +76,49 @@ OVERLAP_CHARS = OVERLAP_TOKENS * CHARS_PER_TOKEN
 _PARA_SPLIT = re.compile(r"\n\s*\n")
 
 _INSERT_SQL = """
-INSERT INTO document_chunk (envelope_id, chunk_index, text, embedding, model)
+INSERT INTO document_chunk (envelope_id, chunk_index, text, embedding, model, excluded_reason)
-VALUES ($1, $2, $3, $4::vector, $5)
+VALUES ($1, $2, $3, $4::vector, $5, $6)
-ON CONFLICT (envelope_id, chunk_index) DO NOTHING
+ON CONFLICT (envelope_id, chunk_index, model) DO NOTHING
 """
 
+# Plan §3.1, signal 1: refined from "any C0 control char" to a raw-count threshold during
+# calibration on the 2683-chunk pilot corpus (2026-07-16) — legit OCR text carries a handful
+# of stray control bytes (paperless:119 mojibake, a confirmed retrieval hit; an English paper
+# with 1-2 stray bytes per chunk), while true binary/barcode noise carries 10-63 per chunk.
+# A threshold of 5 cleanly separates the two on this corpus.
+CONTROL_CHAR_JUNK_THRESHOLD = 5
+# Plan §3.1, signal 2: share of characters outside the "wordy" class (alnum + PL diacritics +
+# common punctuation + whitespace). On the pilot corpus no legitimate chunk crosses 20%, so
+# this never fires alone here — kept for corpora where junk isn't diluted by legible headers.
+NON_WORDY_RATIO_THRESHOLD = 0.30
+# Plan §3.1, signal 3: share of whitespace-split tokens that look like words. Legit text
+# clusters well above 40%; junk (scrambled fonts, dot-leader ToCs) sits below 25% on the
+# pilot corpus.
+WORDLIKE_RATIO_THRESHOLD = 0.25
+
+_CONTROL_CHAR_RE = re.compile(r"[\x00-\x08\x0b-\x1f]")
+_NON_WORDY_CHAR_RE = re.compile(
+    r"[^a-zA-Z0-9ąćęłńóśźżĄĆĘŁŃÓŚŹŻ\s.,;:!?\"'`()\[\]{}<>/\\|@#$%^&*_+=~-]"
+)
+_WORDLIKE_TOKEN_RE = re.compile(r"^[a-zA-Ząćęłńóśźż-]{2,30}$")
+
+
+def is_ocr_junk(text: str) -> bool:
+    """Three-signal OCR-junk heuristic (plan §3.1, decision 1): binary noise (ground barcodes,
+    scrambled fonts), not merely ugly text — partial mojibake alone is not junk (paperless:119
+    stays a confirmed retrieval hit despite it). Chunk-level, never document-level."""
+    if not text:
+        return False
+    if len(_CONTROL_CHAR_RE.findall(text)) >= CONTROL_CHAR_JUNK_THRESHOLD:
+        return True
+    if len(_NON_WORDY_CHAR_RE.findall(text)) / len(text) > NON_WORDY_RATIO_THRESHOLD:
+        return True
+    tokens = text.split()
+    if not tokens:
+        return False
+    wordlike = sum(1 for t in tokens if _WORDLIKE_TOKEN_RE.match(t))
+    return (wordlike / len(tokens)) < WORDLIKE_RATIO_THRESHOLD
+
 
 class EmbeddingDimensionError(RuntimeError):
     """Ollama returned a vector of the wrong dimension for the target `document_chunk` schema."""
@@ -209,12 +248,15 @@ async def embed_chunk(
 
 async def insert_chunk(
     conn: asyncpg.Connection, envelope_id: str, chunk_index: int, text: str,
-    embedding: list[float], model: str,
+    embedding: Optional[list[float]], model: str, excluded_reason: Optional[str] = None,
 ) -> str:
     """Returns asyncpg's command tag (e.g. 'INSERT 0 1' or 'INSERT 0 0' if ON CONFLICT
-    DO NOTHING skipped the row) so the caller can tell a real insert from a no-op."""
+    DO NOTHING skipped the row) so the caller can tell a real insert from a no-op.
+    `embedding=None` (junk chunks, `excluded_reason='ocr_junk'`) inserts a NULL vector —
+    pgvector's HNSW index skips NULLs automatically."""
+    vector_literal = _vector_literal(embedding) if embedding is not None else None
     return await conn.execute(
-        _INSERT_SQL, envelope_id, chunk_index, text, _vector_literal(embedding), model
+        _INSERT_SQL, envelope_id, chunk_index, text, vector_literal, model, excluded_reason
     )
 
 
@@ -240,14 +282,17 @@ async def run(
     Returns stats that must always balance:
 
         documents_fetched  = empty_content + documents_chunked
-        chunks_total        = chunks_already_embedded + chunks_inserted
+        chunks_total        = chunks_already_embedded + chunks_inserted + chunks_junk_flagged
                                + chunks_conflict_skipped + chunks_errors
 
-    `chunks_conflict_skipped` counts inserts where `ON CONFLICT (envelope_id, chunk_index)
+    Each chunk is screened by `is_ocr_junk()` before embedding (plan §3.1): a junk chunk
-    DO NOTHING` silently discarded the row (the pre-fetched `existing` set is the first line
+    skips Ollama entirely and is INSERTed with `excluded_reason='ocr_junk'`,
-    of defense against this and should make it rare; see the module docstring's note on the
+    `embedding=NULL`, counted as `chunks_junk_flagged` — never as `chunks_inserted`.
-    UNIQUE constraint not including `model`) — tracked separately so a silent no-op is never
+
-    miscounted as a successful write.
+    `chunks_conflict_skipped` counts inserts where `ON CONFLICT (envelope_id, chunk_index,
+    model) DO NOTHING` silently discarded the row (the pre-fetched `existing` set is the
+    first line of defense against this and should make it rare) — tracked separately so a
+    silent no-op is never miscounted as a successful write.
 
     Every embedding response's dimension is checked against `EXPECTED_DIM` (raises
     EmbeddingDimensionError and aborts the whole run on mismatch) — never silently indexes
@@ -262,6 +307,7 @@ async def run(
         "chunks_total": 0,
         "chunks_already_embedded": 0,
         "chunks_inserted": 0,
+        "chunks_junk_flagged": 0,
         "chunks_conflict_skipped": 0,
         "chunks_errors": 0,
         "embed_calls": 0,
@@ -297,10 +343,37 @@ async def run(
                         stats["chunks_already_embedded"] += 1
                         continue
 
+                    junk = is_ocr_junk(chunk)
+
                     if not apply:
+                        if junk:
+                            stats["chunks_junk_flagged"] += 1
+                        else:
                             stats["chunks_inserted"] += 1
                         continue
 
+                    if junk:
+                        try:
+                            command_tag = await insert_chunk(
+                                conn, envelope_id, idx, chunk, None, model, excluded_reason="ocr_junk"
+                            )
+                        except Exception:
+                            _log.warning(
+                                "skip.insert_error", envelope_id=envelope_id, chunk_index=idx, exc_info=True
+                            )
+                            stats["chunks_errors"] += 1
+                            continue
+
+                        existing.add(key)
+                        if _rows_affected(command_tag) == 0:
+                            _log.warning(
+                                "chunk.conflict_skipped", envelope_id=envelope_id, chunk_index=idx, model=model
+                            )
+                            stats["chunks_conflict_skipped"] += 1
+                        else:
+                            stats["chunks_junk_flagged"] += 1
+                        continue
+
                     assert session is not None
                     try:
                         embedding, elapsed = await embed_chunk(session, ollama_url, model, chunk)
@@ -345,7 +418,7 @@ async def run(
 
     balance_docs = stats["empty_content"] + stats["documents_chunked"]
     balance_chunks = (
-        stats["chunks_already_embedded"] + stats["chunks_inserted"]
+        stats["chunks_already_embedded"] + stats["chunks_inserted"] + stats["chunks_junk_flagged"]
         + stats["chunks_conflict_skipped"] + stats["chunks_errors"]
     )
     if balance_docs != stats["documents_fetched"] or balance_chunks != stats["chunks_total"]:
@@ -414,7 +487,7 @@ def main() -> None:
     balanced = (
         stats["documents_fetched"] == stats["empty_content"] + stats["documents_chunked"]
         and stats["chunks_total"] == (
-            stats["chunks_already_embedded"] + stats["chunks_inserted"]
+            stats["chunks_already_embedded"] + stats["chunks_inserted"] + stats["chunks_junk_flagged"]
             + stats["chunks_conflict_skipped"] + stats["chunks_errors"]
         )
     )

jobs/documents-ingest/tests/test_chunk_embed.py

@@ -15,6 +15,8 @@ from documents_ingest.chunk_embed import (
     fetch_documents,
     fetch_existing_chunk_keys,
     hard_split,
+    insert_chunk,
+    is_ocr_junk,
     run,
     split_paragraphs,
     _vector_literal,
@@ -144,6 +146,46 @@ class TestVectorLiteral:
         assert _vector_literal([0.1, 0.2, -0.3]) == "[0.1,0.2,-0.3]"
 
 
+class TestIsOcrJunk:
+    """Pilot cases from docs/kb/modules/05-faza3-plan.md §3.1 and the 2026-07-16 calibration
+    session (docs/kb/eval/retrieval-pilot-2026-07-16.md)."""
+
+    def test_clean_text_is_not_junk(self):
+        text = (
+            "Ogólne Warunki Ubezpieczenia PZU Auto ustalają zasady odpowiedzialności "
+            "ubezpieczyciela za szkody powstałe w związku z ruchem pojazdów mechanicznych."
+        )
+        assert is_ocr_junk(text) is False
+
+    def test_ground_barcode_binary_noise_is_junk(self):
+        # paperless:39/96/128/186/192 pattern: dense run of C0 control chars (scrambled
+        # barcode), well above the calibrated raw-count threshold of 5.
+        text = "\x01\x02\x03\x05\x06\x07\x0f\x10\x11\x12\x14\x15\x16\x17\x18\x19qQQQ\x11q\x01\x11!!AA"
+        assert is_ocr_junk(text) is True
+
+    def test_partial_mojibake_is_not_junk(self):
+        # paperless:119: a confirmed retrieval hit (query 4, eval-set) despite partial
+        # mojibake and a couple of stray control bytes — mojibake alone must not exclude it.
+        text = (
+            "Wygenerowano z systemu wydarzeń FIRST Israel | Turniej Regionalny FIRST LEGO "
+            "League\n\x01integraln¹ czêœæ dokumentacji zawodów\nARTEFAKTY Z WĘDKI"
+        )
+        assert is_ocr_junk(text) is False
+
+    def test_stray_single_control_char_in_legible_text_is_not_junk(self):
+        # paperless:141 pattern: one stray control byte in an otherwise clean English
+        # paragraph (a PDF-extraction artifact, not OCR noise).
+        text = "separable\x0fconvolution is equal to the combination of a self-attention layer"
+        assert is_ocr_junk(text) is False
+
+    def test_dot_leader_toc_is_junk(self):
+        text = ". . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . ."
+        assert is_ocr_junk(text) is True
+
+    def test_empty_text_is_not_junk(self):
+        assert is_ocr_junk("") is False
+
+
 class _FakeEmbedResponse:
     def __init__(self, payload, status=200):
         self._payload = payload
@@ -203,6 +245,29 @@ class TestEmbedChunk:
             await embed_chunk(_Session(), "http://fake-ollama", "bge-m3", "text")
 
 
+class TestInsertSql:
+    def test_on_conflict_target_includes_model(self):
+        # Plan §3.3 fix: the old UNIQUE(envelope_id, chunk_index) silently no-opped a
+        # second embedding model via ON CONFLICT DO NOTHING (review finding, phase 2 step
+        # 6). The target must now include `model`.
+        from documents_ingest.chunk_embed import _INSERT_SQL
+        assert "ON CONFLICT (envelope_id, chunk_index, model)" in _INSERT_SQL
+
+
+class TestInsertChunk:
+    async def test_junk_chunk_inserts_null_embedding_with_reason(self):
+        conn = _FakeConn()
+        await insert_chunk(conn, "paperless:1", 0, "junk text", None, "bge-m3", excluded_reason="ocr_junk")
+
+        assert conn.execute_calls == [("paperless:1", 0, "junk text", None, "bge-m3", "ocr_junk")]
+
+    async def test_normal_chunk_inserts_vector_with_no_reason(self):
+        conn = _FakeConn()
+        await insert_chunk(conn, "paperless:1", 0, "real text", [0.1, 0.2], "bge-m3")
+
+        assert conn.execute_calls == [("paperless:1", 0, "real text", "[0.1,0.2]", "bge-m3", None)]
+
+
 class _FakeConn:
     def __init__(self, docs=None, existing_keys=None, existing_model="bge-m3", execute_results=None):
         self._docs = docs or []
@@ -340,10 +405,13 @@ class TestRun:
 
     async def test_embed_error_is_isolated_and_counted(self, monkeypatch):
         # Two chunks: force one to fail via a document long enough to produce 2 chunks
-        # (chunk_size/overlap kept small to make the test fast and explicit).
+        # (chunk_size/overlap kept small to make the test fast and explicit). Real
-        big = "a" * 100 + "\n\n" + "b" * 100
+        # word-like tokens (not a single 100-char run) so neither chunk trips is_ocr_junk.
+        para1 = " ".join(["word"] * 20)
+        para2 = " ".join(["term"] * 20)
+        big = para1 + "\n\n" + para2
         conn = _FakeConn(docs=[_row("paperless:1", big)])
-        ollama = _FakeOllamaSession(dim=1024, fail_for={"a" * 100})
+        ollama = _FakeOllamaSession(dim=1024, fail_for={para1})
         self._patch(monkeypatch, conn, ollama)
 
         stats = await run(dsn="postgresql://fake", apply=True, chunk_size=100, chunk_overlap=20)
@@ -358,7 +426,8 @@ class TestRun:
     async def test_insert_error_is_isolated_and_counted(self, monkeypatch):
         # Two chunks; the first chunk's INSERT raises (simulated DB blip) but the run
         # continues and the second chunk still gets embedded and inserted normally.
-        big = "a" * 100 + "\n\n" + "b" * 100
+        # Real word-like tokens so neither chunk trips is_ocr_junk.
+        big = " ".join(["word"] * 20) + "\n\n" + " ".join(["term"] * 20)
         conn = _FakeConn(docs=[_row("paperless:1", big)])
 
         real_execute = conn.execute
@@ -416,6 +485,66 @@ class TestRun:
         assert stats["chunks_already_embedded"] == 0
         assert stats["chunks_inserted"] == 1
 
+    async def test_junk_chunk_flagged_without_calling_ollama(self, monkeypatch):
+        junk = ". . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . ."
+        conn = _FakeConn(docs=[_row("paperless:1", junk)])
+        ollama = _FakeOllamaSession(dim=1024)
+        self._patch(monkeypatch, conn, ollama)
+
+        stats = await run(dsn="postgresql://fake", apply=True)
+
+        assert stats["chunks_junk_flagged"] == 1
+        assert stats["chunks_inserted"] == 0
+        assert ollama.requests == []
+        assert conn.execute_calls == [("paperless:1", 0, junk, None, "bge-m3", "ocr_junk")]
+        assert stats["chunks_total"] == (
+            stats["chunks_already_embedded"] + stats["chunks_inserted"] + stats["chunks_junk_flagged"]
+            + stats["chunks_conflict_skipped"] + stats["chunks_errors"]
+        )
+
+    async def test_junk_chunk_dry_run_counted_without_db_or_ollama(self, monkeypatch):
+        junk = ". . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . ."
+        conn = _FakeConn(docs=[_row("paperless:1", junk)])
+        ollama = _FakeOllamaSession(dim=1024)
+        self._patch(monkeypatch, conn, ollama)
+
+        stats = await run(dsn="postgresql://fake", apply=False)
+
+        assert stats["chunks_junk_flagged"] == 1
+        assert stats["chunks_inserted"] == 0
+        assert ollama.requests == []
+        assert conn.execute_calls == []
+
+    async def test_junk_chunk_rerun_is_idempotent(self, monkeypatch):
+        # Same (envelope_id, chunk_index) already flagged ocr_junk under this model — a
+        # rerun must skip it, same as a normal embedded chunk (plan §3.1: backfill is a
+        # one-shot, but chunk_embed.py itself must stay idempotent for new junk too).
+        junk = ". . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . ."
+        conn = _FakeConn(
+            docs=[_row("paperless:1", junk)],
+            existing_keys=[("paperless:1", 0)],
+        )
+        ollama = _FakeOllamaSession(dim=1024)
+        self._patch(monkeypatch, conn, ollama)
+
+        stats = await run(dsn="postgresql://fake", apply=True)
+
+        assert stats["chunks_already_embedded"] == 1
+        assert stats["chunks_junk_flagged"] == 0
+        assert conn.execute_calls == []
+
+    async def test_junk_conflict_skip_not_double_counted(self, monkeypatch):
+        junk = ". . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . ."
+        conn = _FakeConn(docs=[_row("paperless:1", junk)], execute_results=["INSERT 0 0"])
+        ollama = _FakeOllamaSession(dim=1024)
+        self._patch(monkeypatch, conn, ollama)
+
+        stats = await run(dsn="postgresql://fake", apply=True)
+
+        assert stats["chunks_conflict_skipped"] == 1
+        assert stats["chunks_junk_flagged"] == 0
+        assert ollama.requests == []
+
     async def test_stats_balance_documents_and_chunks(self, monkeypatch):
         conn = _FakeConn(docs=[
             _row("paperless:1", "content one"),

services/kb-postgres/init/003_chunk_model_key.sql

@@ -0,0 +1,17 @@
+-- KB spine: document_chunk — unique-key fix + exclusion flag
+-- Additive: does not modify 001_envelope.sql or 002_chunks.sql.
+-- Version: 003 — chunk model key + excluded_reason (module 5, phase 3, plan §3.3)
+--
+-- (a) UNIQUE rozszerzony o model — bez tego drugi model embeddingów cicho się no-opuje
+--     na ON CONFLICT DO NOTHING (finding z review kroku 6 fazy 2).
+-- (b) excluded_reason — flaga wykluczenia chunka z retrievalu i kompilacji
+--     (NULL = aktywny; 'ocr_junk' | 'duplicate').
+
+ALTER TABLE document_chunk
+    DROP CONSTRAINT IF EXISTS document_chunk_envelope_id_chunk_index_key;
+ALTER TABLE document_chunk
+    ADD CONSTRAINT document_chunk_envelope_chunk_model_key
+    UNIQUE (envelope_id, chunk_index, model);
+
+ALTER TABLE document_chunk
+    ADD COLUMN IF NOT EXISTS excluded_reason TEXT;  -- NULL = aktywny