Derive chunk spans from byte offsets behind a chunker contract

tools/dsl_indexer/chunk.py

@@ -1,134 +1,64 @@
 from pathlib import Path
-from typing import List, Tuple
+from typing import Callable, Dict, List, Tuple
 import hashlib
-import re
 
-from .config import CHUNK_OVERLAP_CHARS, CHUNK_TARGET_CHARS, REPO_TYPE_MAP, REPOS_DIR
+from .chunkers import base, text
+from .config import CHUNKING, REPO_TYPE_MAP, REPOS_DIR
 from .text_utils import tokenize
 from .chunk_types import Chunk
 
+# markdown and code route to text until their strategies land.
+_STRATEGIES: Dict[str, Callable[[base.ChunkContext], List[base.ChunkDraft]]] = {
+    "text": text.chunk,
+    "markdown": text.chunk,
+    "code": text.chunk,
+}
+
 
 def chunk_file(path: Path) -> List[Chunk]:
-    try:
-        text = path.read_text(encoding="utf-8")
-    except UnicodeDecodeError:
-        text = path.read_text(encoding="utf-8", errors="ignore")
+    source = base.normalize_source(path.read_bytes())
+    if not source.strip():
+        return []
 
     repo, rel_path = _repo_and_rel_path(path)
     repo_type = REPO_TYPE_MAP.get(repo, "spec")
-    if path.suffix.lower() in {".md", ".mdx"}:
-        chunks = _chunk_markdown(text, repo, rel_path, repo_type)
-    else:
-        chunks = _chunk_plain(text, repo, rel_path, repo_type)
-    return chunks
-
-
-def _repo_and_rel_path(path: Path) -> Tuple[str, str]:
-    rel = path.resolve().relative_to(REPOS_DIR)
-    parts = rel.parts
-    repo = parts[0]
-    rel_path = str(Path(*parts))
-    return repo, rel_path
-
-
-def _chunk_markdown(text: str, repo: str, rel_path: str, repo_type: str) -> List[Chunk]:
-    lines = text.splitlines()
-    sections = []
-    current_heading = "Document Start"
-    start_idx = 0
-
-    for i, line in enumerate(lines):
-        if re.match(r"^\s{0,3}#{1,6}\s+.+", line):
-            if i > start_idx:
-                sections.append((current_heading, start_idx, i))
-            current_heading = line.lstrip("# ").strip()
-            start_idx = i
-
-    sections.append((current_heading, start_idx, len(lines)))
+    profile, language = CHUNKING.resolve(path.suffix.lower())
+    ctx = base.ChunkContext(
+        source=source,
+        line_starts=base.line_starts(source),
+        target_size=profile.target_size,
+        hard_max_size=profile.hard_max_size,
+        overlap=profile.overlap,
+        language=language,
+    )
 
     chunks: List[Chunk] = []
-    for heading, start, end in sections:
-        section_lines = lines[start:end]
-        section_text = "\n".join(section_lines).strip()
-        if not section_text:
+    for draft in _STRATEGIES[profile.strategy](ctx):
+        resolved = base.finalize(ctx, draft)
+        if resolved is None:
             continue
-        pieces = _slice_with_overlap(section_text, CHUNK_TARGET_CHARS, CHUNK_OVERLAP_CHARS)
-        offset = 0
-        for idx, piece in enumerate(pieces):
-            line_start, line_end = _line_window_for_piece(section_text, piece, start + 1, offset)
-            offset = max(0, section_text.find(piece, offset) + len(piece))
-            chunks.append(
-                _build_chunk(
-                    repo=repo,
-                    rel_path=rel_path,
-                    repo_type=repo_type,
-                    section=heading,
-                    line_start=line_start,
-                    line_end=line_end,
-                    content=piece,
-                    ordinal=idx,
-                )
-            )
-    return chunks
-
-
-def _chunk_plain(text: str, repo: str, rel_path: str, repo_type: str) -> List[Chunk]:
-    lines = text.splitlines()
-    content = "\n".join(lines).strip()
-    if not content:
-        return []
-
-    chunks: List[Chunk] = []
-    pieces = _slice_with_overlap(content, CHUNK_TARGET_CHARS, CHUNK_OVERLAP_CHARS)
-    offset = 0
-    for idx, piece in enumerate(pieces):
-        line_start, line_end = _line_window_for_piece(content, piece, 1, offset)
-        offset = max(0, content.find(piece, offset) + len(piece))
+        content, line_start, line_end = resolved
         chunks.append(
             _build_chunk(
                 repo=repo,
                 rel_path=rel_path,
                 repo_type=repo_type,
-                section="General",
+                section=draft.section,
                 line_start=line_start,
                 line_end=line_end,
-                content=piece,
-                ordinal=idx,
+                content=content,
+                ordinal=len(chunks),
             )
         )
     return chunks
 
 
-def _slice_with_overlap(text: str, target: int, overlap: int) -> List[str]:
-    if len(text) <= target:
-        return [text]
-
-    chunks: List[str] = []
-    step = max(1, target - overlap)
-    start = 0
-    while start < len(text):
-        end = min(len(text), start + target)
-        piece = text[start:end].strip()
-        if piece:
-            chunks.append(piece)
-        if end == len(text):
-            break
-        start += step
-    return chunks
-
-
-def _line_window_for_piece(full_text: str, piece: str, base_line: int, offset_hint: int) -> Tuple[int, int]:
-    idx = full_text.find(piece, offset_hint)
-    if idx < 0:
-        idx = full_text.find(piece)
-    if idx < 0:
-        return base_line, base_line
-
-    before = full_text[:idx]
-    within = full_text[idx : idx + len(piece)]
-    line_start = base_line + before.count("\n")
-    line_end = line_start + within.count("\n")
-    return line_start, max(line_start, line_end)
+def _repo_and_rel_path(path: Path) -> Tuple[str, str]:
+    rel = path.resolve().relative_to(REPOS_DIR)
+    parts = rel.parts
+    repo = parts[0]
+    rel_path = str(Path(*parts))
+    return repo, rel_path
 
 
 def _build_chunk(

tools/dsl_indexer/chunkers/base.py

@@ -0,0 +1,100 @@
+"""Chunker contract and shared span math.
+
+A chunker's only job is to choose byte cut points. It never produces content,
+line numbers, IDs, or token counts — `finalize` does that, once, for everyone.
+
+Modules in this package import only from here, never from `config`, so they
+stay unit-testable without a `workspace.yaml`.
+"""
+
+from __future__ import annotations
+
+from bisect import bisect_right
+from dataclasses import dataclass
+from typing import List, Optional, Tuple
+
+WHITESPACE = b" \t\n\r\x0b\x0c"
+
+
+@dataclass(frozen=True)
+class ChunkDraft:
+    start_byte: int
+    end_byte: int
+    section: str
+
+
+@dataclass(frozen=True)
+class ChunkContext:
+    source: bytes
+    line_starts: List[int]
+    target_size: int
+    hard_max_size: int
+    overlap: int = 0
+    language: Optional[str] = None
+
+
+def normalize_source(raw: bytes) -> bytes:
+    """CRLF collapsed once, re-encoded so every byte span slices cleanly as UTF-8.
+
+    Lone \\r and \\x0b are deliberately left alone: treating them as line breaks
+    is what makes today's line numbers diverge from byte-derived ones.
+    """
+    return raw.replace(b"\r\n", b"\n").decode("utf-8", errors="ignore").encode("utf-8")
+
+
+def line_starts(source: bytes) -> List[int]:
+    """Byte offset of each line start. A trailing newline opens a final empty line."""
+    starts = [0]
+    for i, byte in enumerate(source):
+        if byte == 0x0A:
+            starts.append(i + 1)
+    return starts
+
+
+def line_of(starts: List[int], offset: int) -> int:
+    """1-based line number containing `offset`."""
+    return bisect_right(starts, offset)
+
+
+def snap_forward(source: bytes, index: int) -> int:
+    """Advance past UTF-8 continuation bytes so `index` lands on a codepoint boundary."""
+    while index < len(source) and 0x80 <= source[index] < 0xC0:
+        index += 1
+    return index
+
+
+def finalize(ctx: ChunkContext, draft: ChunkDraft) -> Optional[Tuple[str, int, int]]:
+    """Resolve a draft to (content, line_start, line_end), or None if it is all whitespace.
+
+    Offsets snap to codepoint boundaries and surrounding whitespace is trimmed
+    before lines are read, so the reported range always covers the content.
+    """
+    start = snap_forward(ctx.source, max(0, draft.start_byte))
+    end = snap_forward(ctx.source, min(len(ctx.source), draft.end_byte))
+    while start < end and ctx.source[start] in WHITESPACE:
+        start += 1
+    while end > start and ctx.source[end - 1] in WHITESPACE:
+        end -= 1
+    if start >= end:
+        return None
+
+    content = ctx.source[start:end].decode("utf-8", errors="ignore")
+    # end is exclusive: a span ending at column 0 belongs to the previous line.
+    return content, line_of(ctx.line_starts, start), line_of(ctx.line_starts, end - 1)
+
+
+def split_oversized(start: int, end: int, section: str, ctx: ChunkContext) -> List[ChunkDraft]:
+    """Cut [start, end) down to `hard_max_size`, preferring line boundaries. Tiles exactly."""
+    if end - start <= ctx.hard_max_size:
+        return [ChunkDraft(start, end, section)]
+
+    drafts: List[ChunkDraft] = []
+    cut = start
+    while cut < end:
+        limit = min(end, cut + ctx.hard_max_size)
+        if limit < end:
+            newline = ctx.source.rfind(b"\n", cut, limit)
+            limit = newline + 1 if newline > cut else snap_forward(ctx.source, limit)
+        drafts.append(ChunkDraft(cut, limit, section))
+        cut = limit
+    return drafts

tools/dsl_indexer/chunkers/text.py

@@ -0,0 +1,36 @@
+"""Fixed-size overlapping chunker — the universal fallback.
+
+Cut positions match the pre-contract implementation exactly, so boundaries for
+`.json`, `.yaml`, `.xml`, `.csproj` and friends do not move.
+"""
+
+from typing import List
+
+from .base import ChunkContext, ChunkDraft, snap_forward
+
+SECTION = "General"
+
+
+def chunk(ctx: ChunkContext) -> List[ChunkDraft]:
+    size = len(ctx.source)
+    if size == 0:
+        return []
+    if size <= ctx.target_size:
+        return [ChunkDraft(0, size, SECTION)]
+
+    drafts: List[ChunkDraft] = []
+    step = max(1, ctx.target_size - ctx.overlap)
+    start = 0
+    while start < size:
+        end = min(size, start + ctx.target_size)
+        drafts.append(
+            ChunkDraft(
+                snap_forward(ctx.source, start),
+                snap_forward(ctx.source, end),
+                SECTION,
+            )
+        )
+        if end == size:
+            break
+        start += step
+    return drafts

tools/dsl_indexer/config.py

@@ -23,6 +23,7 @@
 SOURCE_REPO_NAMES: list[str] = [r["name"] for r in SOURCE_REPOS]
 REPO_TYPE_MAP: dict[str, str] = {r["name"]: r["type"] for r in SOURCE_REPOS}
 WORKSPACE_NAME: str = _CFG.name
+CHUNKING = _CFG.chunking
 
 TEXT_EXTENSIONS = {
     ".md",
@@ -90,7 +91,6 @@
 }
 
 MAX_FILE_BYTES = 512_000
-CHUNK_TARGET_CHARS = 1100
-CHUNK_OVERLAP_CHARS = 180
 SNIPPET_MAX_CHARS = 260
-INDEX_VERSION = "1.1.0"
+# 1.2.0: chunk spans derive from byte offsets, rotating every chunk_id.
+INDEX_VERSION = "1.2.0"