fix(kb): chunker ceiling + supersession NULL-identifier collision

Two bugs surfaced during Phase 1 validation on real circulars:

1. The chunker produced one giant chunk for large statutes (ספר הליקויים,
   ספר המבחנים) that the section regex could not split. This exceeded
   Voyage's 120K-token batch limit and the ingest failed. Add a hard
   2000-token ceiling per chunk with a character-based fallback split that
   preserves heading_path and section_ref.

2. ingest_source matched "existing source" by kind + COALESCE(identifier,
   ''), so any two sources with NULL identifier compared equal. The second
   ingest would then supersede the first unrelated document. Match by
   identifier only when one is provided; otherwise fall back to
   (kind + title) and require the candidate to still be active.

Refs Task Master #2

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

api/services/kb/chunker.py

@@ -13,6 +13,10 @@ from typing import Literal
 # Approx tokens per char for Hebrew/mixed text (voyage-multilingual-2 tokenizer).
 _TOKENS_PER_CHAR = 0.35
 
+# Hard safety ceiling per chunk. Voyage accepts up to 32K tokens per item,
+# but a single batch is capped at 120K tokens. Keep chunks well below both.
+_MAX_TOKENS_PER_CHUNK = 2000
+
 # A "section" in Israeli statutes: "סעיף 12", "תקנה 5", "5א.", "12(ב)"
 _SECTION_RE = re.compile(
     r"(?m)^\s*(?:סעיף\s+|תקנה\s+|§\s*)?(\d+[א-ת]?\.?)\s*(?:\(([א-ת\d]+)\))?\s*[\.\:]",
@@ -50,6 +54,33 @@ def _as_dicts(chunks: list[Chunk]) -> list[dict]:
     ]
 
 
+def _split_oversized(chunks: list[Chunk]) -> list[Chunk]:
+    """Split any chunk whose token_count exceeds the ceiling into fixed-size pieces.
+
+    Preserves heading_path/section_ref on each piece and re-indexes.
+    """
+    max_chars = int(_MAX_TOKENS_PER_CHUNK / _TOKENS_PER_CHAR)
+    result: list[Chunk] = []
+    for c in chunks:
+        if c.token_count <= _MAX_TOKENS_PER_CHUNK:
+            c.chunk_index = len(result)
+            result.append(c)
+            continue
+        text = c.content
+        for start in range(0, len(text), max_chars):
+            piece = text[start : start + max_chars]
+            result.append(
+                Chunk(
+                    chunk_index=len(result),
+                    heading_path=c.heading_path,
+                    section_ref=c.section_ref,
+                    content=piece,
+                    token_count=_count_tokens(piece),
+                )
+            )
+    return result
+
+
 def chunk_statute(text: str) -> list[dict]:
     """Law / regulation: one chunk per section. Preserves section numbering."""
     text = text.strip()
@@ -160,5 +191,18 @@ def chunk_circular(
 
 def chunk(text: str, kind: Literal["law", "regulation", "circular"]) -> list[dict]:
     if kind in ("law", "regulation"):
-        return chunk_statute(text)
-    return chunk_circular(text)
+        raw = chunk_statute(text)
+    else:
+        raw = chunk_circular(text)
+    # Reconstruct Chunk objects to apply the safety ceiling.
+    chunks = [
+        Chunk(
+            chunk_index=r["chunk_index"],
+            heading_path=r["heading_path"] or "",
+            section_ref=r["section_ref"] or "",
+            content=r["content"],
+            token_count=r["token_count"],
+        )
+        for r in raw
+    ]
+    return _as_dicts(_split_oversized(chunks))

api/services/kb/ingest.py

@@ -86,14 +86,28 @@ async def ingest_source(
     pool = await get_pool()
     async with pool.acquire() as conn:
         async with conn.transaction():
+            # Match a prior source by (kind + identifier) when identifier is provided,
+            # otherwise fall back to (kind + title). Without this, two different
+            # sources with NULL identifier would both compare equal and the second
+            # ingest would incorrectly supersede the first.
+            if identifier:
                 existing = await conn.fetchrow(
                     """
                     SELECT id, checksum FROM kb_source
-                WHERE kind = $1 AND COALESCE(identifier,'') = COALESCE($2,'')
+                    WHERE kind = $1 AND identifier = $2 AND superseded_by IS NULL
                     ORDER BY created_at DESC LIMIT 1
                     """,
                     kind, identifier,
                 )
+            else:
+                existing = await conn.fetchrow(
+                    """
+                    SELECT id, checksum FROM kb_source
+                    WHERE kind = $1 AND title = $2 AND identifier IS NULL AND superseded_by IS NULL
+                    ORDER BY created_at DESC LIMIT 1
+                    """,
+                    kind, title,
+                )
             was_updated = False
 
             if existing and existing["checksum"] == checksum: