refactor(phase-5): core domain logic for retrieval, chunking, prompts

FORWARD_PORT_LOG.md

@@ -0,0 +1,27 @@
+# Forward Port Log
+
+Tracks `dev` changes during MODE 2 isolation (Phases 5-9).
+Each entry: what changed on `dev`, whether it was forward-ported or deferred
+to the cutover re-implementation queue.
+
+> Created retroactively at the start of Phase 5 (2026-04-29). The Phase 0-4
+> Mode 1 work merged from `dev` cleanly so this log starts empty.
+
+---
+
+## Forward-ported (critical)
+
+_Security fixes, data-loss bugs, production outages re-implemented against
+the new architecture. Each entry pairs the dev commit with the refactor
+commit so a reviewer can audit equivalence._
+
+(none yet)
+
+## Deferred to cutover (features)
+
+_Non-critical changes that landed on `dev` during MODE 2. These will be
+re-implemented directly in the new architecture during MODE 3 (Phases
+10-12) or post-cutover. List the dev PR number / commit and the target
+location in the new layout._
+
+(none yet)

REFACTORING_DECISION_LOG.md

@@ -55,7 +55,53 @@ match reality, then record the reasoning here.
 
 ---
 
-## Template for future entries
+## Phase 5 — Core domain logic for retrieval, chunking, prompts (2026-04-29)
+
+**1. New `RetrievalSearcher` port in `core/retrieval/searcher.py`, separate from
+the narrow `VectorStore` ABC.**
+The retriever needs four operations (search by query string, multi-query
+search, related-chunk lookup, ancestor lookup) that the Phase-4
+`VectorStore` ABC does not cover — that ABC is intentionally narrow
+(`search(embedding, top_k)`). We added a transitional ABC the retriever
+depends on, implemented by `services/storage/milvus_ray_shim.py` over the
+legacy Ray actor.
+- Why: STRATEGY §5A says the retriever should "call `VectorStore.search()`
+  (port method), not `vectordb.async_search.remote()`". But the legacy Ray
+  actor's `async_search` takes a query *string* and embeds internally; the
+  narrow `VectorStore.search(embedding, ...)` ABC doesn't fit. Pre-embedding
+  in the shim before calling Ray is impractical because the actor also owns
+  BM25 and surrounding-chunks semantics. A retrieval-facing port keeps the
+  retriever clean of Ray today and survives Phase 7 — when the Vectordb
+  god object is decomposed, these methods either move onto a richer
+  `VectorStore` or split between `VectorStore` and `ChunkRepository`.
+- Alternative considered: extend `VectorStore` with the four legacy methods.
+  Rejected — bloats the ABC with operations that should not exist past
+  Phase 7. Also considered: skip the new core port and have the retriever
+  call the Ray actor through the shim with the legacy method names —
+  rejected because that leaks legacy method names into core/ and makes the
+  retriever harder to test.
+
+**2. Skipped: bringing up integration tests for the new code.**
+Phase 5 ships pure-domain unit tests only (50 new tests in `core/`, no Ray /
+Milvus / real LLM). The new pipeline is dormant until Phase 8 wires it.
+- Why: Mode 2 forbids touching the legacy wiring; the new pipeline has
+  nowhere to be plugged in yet. Integration coverage will land with Phase 8
+  orchestrators (or Phase 7 storage if it goes first).
+- Alternative considered: stand up a fake searcher in an integration
+  fixture and run a full retriever-pipeline-RRF round trip. Defers the
+  same coverage to Phase 8 with less code; not worth the extra fixtures.
+
+**3. `Query`, `SearchQueries`, `TemporalPredicate` lifted into
+`core/models/query.py`.**
+The legacy `components/pipeline.py` defined these inline. The new
+`RetrieverPipeline` consumes them — they're domain types, not pipeline
+internals.
+- Why: STRATEGY §2 calls these out as `pipeline.py SearchQueries → core/models/query.py`.
+- Alternative considered: keep them in `core/retrieval/`. Rejected — they
+  describe a query in the abstract; the orchestrator (Phase 8) and the API
+  layer will both use them, not just retrieval.
+
+---
 
 ## Phase 1 — Registry + Exceptions (2026-04-21)
 

openrag/core/chunking/__init__.py

@@ -1,6 +1,24 @@
-"""ChunkingStrategy ABC + registry."""
+"""ChunkingStrategy ABC + registry + concrete strategies."""
 
 from .chunking_strategy import ChunkingStrategy
+from .markdown_utils import (
+    MDElement,
+    chunk_table,
+    get_chunk_page_number,
+    parse_markdown_table,
+    split_md_elements,
+)
+from .recursive import BaseChunker, RecursiveSplitter
 from .registry import chunking_registry
 
-__all__ = ["ChunkingStrategy", "chunking_registry"]
+__all__ = [
+    "ChunkingStrategy",
+    "chunking_registry",
+    "BaseChunker",
+    "RecursiveSplitter",
+    "MDElement",
+    "chunk_table",
+    "get_chunk_page_number",
+    "parse_markdown_table",
+    "split_md_elements",
+]

openrag/core/chunking/markdown_utils.py

@@ -0,0 +1,224 @@
+"""Markdown parsing primitives used by chunking strategies.
+
+Pure functions extracted from ``components/indexer/chunker/utils.py``. They
+recognize page markers, image-description blocks, and tables; split a
+markdown document into typed elements; and split oversize tables along
+their semantic groups.
+
+This module has no IO and no config dependency.
+"""
+
+from __future__ import annotations
+
+import re
+from collections.abc import Callable
+from dataclasses import dataclass
+from typing import Literal
+
+from openrag.core.utils.text import clean_markdown_table_spacing
+
+# Header + delimiter + at least one row.
+TABLE_RE = re.compile(
+    r"((?:^|\n)\|.*?\|\r?\n\|\s*[:-]+(?:\s*\|[:-]+)*\|\r?\n(?:\|.*?\|\r?\n)+)",
+    re.DOTALL | re.MULTILINE,
+)
+
+# `<image_description>...</image_description>` block injected by the VLM step.
+IMAGE_RE = re.compile(r"(<image_description>(.*?)</image_description>)", re.DOTALL)
+
+# `[PAGE_N]` page-boundary markers — content BEFORE [PAGE_N] is on page N.
+PAGE_RE = re.compile(r"\[PAGE_(\d+)\]")
+
+
+ElementType = Literal["text", "table", "image"]
+
+
+@dataclass
+class MDElement:
+    """A typed segment of markdown content with optional source page number."""
+
+    type: ElementType
+    content: str
+    page_number: int | None = None
+
+    def __repr__(self) -> str:
+        return f"MDElement(type={self.type}, page_number={self.page_number}, content={self.content[:100]}...)"
+
+
+def span_inside(span: tuple[int, int], container: tuple[int, int]) -> bool:
+    """Return True if ``span`` is fully contained within ``container``."""
+    return container[0] <= span[0] and span[1] <= container[1]
+
+
+def get_page_number(position: int, page_markers: list[tuple[int, int]]) -> int:
+    """Look up the page number for a position in the source markdown.
+
+    ``page_markers`` is a sorted list of ``(offset, page_n)`` tuples taken
+    from ``[PAGE_N]`` matches. Content AFTER ``[PAGE_N]`` belongs to page
+    ``N + 1``; content before any marker is page 1.
+    """
+    current_page = 1
+    for marker_pos, page_num in page_markers:
+        if position >= marker_pos:
+            current_page = page_num + 1
+        else:
+            break
+    return current_page
+
+
+def split_md_elements(md_text: str) -> list[MDElement]:
+    """Split markdown into ``MDElement`` segments of text, table, and image.
+
+    Tables nested inside an ``<image_description>`` block are NOT extracted
+    as separate elements — they belong to the image.
+    """
+    page_markers: list[tuple[int, int]] = []
+    for match in PAGE_RE.finditer(md_text):
+        page_markers.append((match.start(), int(match.group(1))))
+    page_markers.sort()
+
+    all_matches: list[tuple[tuple[int, int], ElementType, str, int | None]] = []
+    image_spans: list[tuple[int, int]] = []
+
+    for match in IMAGE_RE.finditer(md_text):
+        span = match.span()
+        page_num = get_page_number(span[0], page_markers)
+        all_matches.append((span, "image", match.group(1).strip(), page_num))
+        image_spans.append(span)
+
+    for match in TABLE_RE.finditer(md_text):
+        span = match.span()
+        if not any(span_inside(span, image_span) for image_span in image_spans):
+            page_num = get_page_number(span[0], page_markers)
+            all_matches.append((span, "table", match.group(1).strip(), page_num))
+
+    all_matches.sort(key=lambda x: x[0][0])
+
+    parts: list[MDElement] = []
+    last = 0
+
+    for (start, end), match_type, content, page_num in all_matches:
+        if start > last:
+            text_segment = md_text[last:start]
+            if text_segment.strip():
+                parts.append(MDElement(type="text", content=text_segment.strip()))
+        parts.append(MDElement(type=match_type, content=content, page_number=page_num))
+        last = end
+
+    if last < len(md_text):
+        remaining = md_text[last:]
+        if remaining.strip():
+            parts.append(MDElement(type="text", content=remaining.strip()))
+
+    return parts
+
+
+def get_chunk_page_number(chunk_str: str, previous_chunk_ending_page: int = 1) -> dict[str, int]:
+    """Resolve start and end pages for a text chunk containing ``[PAGE_N]`` markers.
+
+    Returns ``{"start_page": int, "end_page": int}``.
+    """
+    matches = list(PAGE_RE.finditer(chunk_str))
+
+    if not matches:
+        return {
+            "start_page": previous_chunk_ending_page,
+            "end_page": previous_chunk_ending_page,
+        }
+
+    first_match = matches[0]
+    last_match = matches[-1]
+    last_char_idx = len(chunk_str) - 1
+
+    if first_match.start() == 0:
+        start_page = int(first_match.group(1)) + 1
+    else:
+        start_page = previous_chunk_ending_page
+
+    if last_match.end() - 1 == last_char_idx:
+        end_page = int(last_match.group(1))
+    else:
+        end_page = int(last_match.group(1)) + 1
+
+    return {"start_page": start_page, "end_page": end_page}
+
+
+def parse_markdown_table(markdown_table: str) -> tuple[list[str], list[list[str]]]:
+    """Parse a markdown table into header lines + groups of rows.
+
+    Rows are grouped by the first column ("Domain"): a non-empty Domain
+    starts a new group, an empty Domain continues the current group. This
+    preserves the document's logical structure when chunking large tables.
+    """
+    lines = markdown_table.strip().split("\n")
+    header_lines = lines[:2]
+    data_rows = lines[2:]
+
+    groups: list[list[str]] = []
+    current_group: list[str] = []
+
+    for row in data_rows:
+        cells = [cell.strip() for cell in row.split("|")[1:-1]]
+        if not cells:
+            continue
+        domain = cells[0]
+        if domain:
+            if current_group:
+                groups.append(current_group)
+            current_group = [row]
+        else:
+            current_group.append(row)
+
+    if current_group:
+        groups.append(current_group)
+
+    return header_lines, groups
+
+
+def chunk_table(
+    table_element: MDElement,
+    chunk_size: int,
+    length_function: Callable[[str], int],
+) -> list[MDElement]:
+    """Split an oversize markdown table into multiple ``MDElement`` chunks.
+
+    Each chunk repeats the table header. When a new chunk starts, the LAST
+    row of the previous chunk is replayed as overlap so context is preserved
+    across the boundary.
+    """
+    txt = clean_markdown_table_spacing(table_element.content)
+    header_lines, groups = parse_markdown_table(txt)
+
+    header_text = "\n".join(header_lines)
+    group_texts = ["\n".join(g) for g in groups]
+
+    header_ntoks = length_function(header_text)
+    groups_ntoks = [length_function(g) for g in group_texts]
+
+    subtables: list[str] = []
+    body_rows: list[str] = []  # rows under the current chunk, header excluded
+    body_size = 0
+    prev_last_row: str | None = None
+
+    for group_txt, g_ntoks in zip(group_texts, groups_ntoks, strict=True):
+        # Only flush when we actually have body content to flush — otherwise an
+        # oversized first group would emit a header-only chunk.
+        if body_rows and header_ntoks + body_size + g_ntoks > chunk_size:
+            subtables.append("\n".join([header_text, *body_rows]))
+            body_rows = []
+            body_size = 0
+            # Replay only the last row of the previous chunk as overlap
+            # (matches the docstring contract; prev_last_row is the trailing
+            # line of the last admitted group).
+            if prev_last_row:
+                body_rows.append(prev_last_row)
+                body_size += length_function(prev_last_row)
+        body_rows.append(group_txt)
+        body_size += g_ntoks
+        # The "last row" is the trailing line of this group, not the whole group.
+        prev_last_row = group_txt.rsplit("\n", 1)[-1]
+
+    if body_rows:
+        subtables.append("\n".join([header_text, *body_rows]))
+
+    return [MDElement(type="table", content=subtable, page_number=table_element.page_number) for subtable in subtables]