From 51ec09d98f559eb28a2376444d67bbbce28395fc Mon Sep 17 00:00:00 2001 From: huangsen Date: Wed, 13 May 2026 14:32:00 +0800 Subject: [PATCH] feat(parser): implement AST-based markdown chunking with full document TOC - Replace legacy line-based chunking with AST tree approach that builds a complete document skeleton with content inlined under relevant sections - Add new chunking parameters: chunk_chars (default 2000) and embed_toc (default True) to control content size and TOC inclusion - Implement recursive chunking algorithm that respects structural boundaries (code lines, table rows, list items) and prevents splits inside blocks - Introduce part markers [Part X/N] for oversized leaf blocks that require splitting - Add CLI tool for inspecting parsed chunks and edges with options for preview and configuration - Refactor edge extraction to use FileEdge.from_text instead of parse_wikilinks for consistency BREAKING CHANGE: Chunk format changes significantly with full TOC skeleton wrapping content, affecting embedding models expecting breadcrumb prefixes. --- .../file_parser/linked_file_parser.py | 883 +++++++++++++++--- reme2/memory/maintainer.py | 2 +- reme2/memory/memory_io.py | 6 +- reme2/schema/__init__.py | 3 +- reme2/schema/file_edge.py | 278 +++--- tests/test_file_edge.py | 348 +++++++ tests/test_md_chunker.py | 726 ++++++++++++++ tests/test_wikilink.py | 219 ----- 8 files changed, 1979 insertions(+), 486 deletions(-) create mode 100644 tests/test_file_edge.py create mode 100644 tests/test_md_chunker.py delete mode 100644 tests/test_wikilink.py diff --git a/reme2/component/file_parser/linked_file_parser.py b/reme2/component/file_parser/linked_file_parser.py index 97683eab..3b5add8b 100644 --- a/reme2/component/file_parser/linked_file_parser.py +++ b/reme2/component/file_parser/linked_file_parser.py @@ -1,204 +1,795 @@ -"""Markdown file parser — frontmatter + wikilink graph + AST-aware chunking. +"""Markdown file parser — frontmatter + wikilink graph + AST tree chunks. -The chunker splits markdown into semantic blocks using: - - ATX headings (#, ##, ..., ######) as section anchors - - Blank lines (paragraph boundaries) as soft splits within a section - - Code fences (``` or ~~~) preserved as a single block +Chunking algorithm (full-doc skeleton + inlined content) +======================================================== -Each block carries a `heading_path` breadcrumb prepended to its text — gives -the embedding model section context AND lets retrieval results show callers -where the hit lives. The hash is computed over the final text (with -breadcrumb), so renaming a heading correctly invalidates child block -embeddings. +Each chunk renders the **complete heading skeleton of the entire +document**, with this chunk's content inlined under the section that +owns it. Sections that don't own this chunk's content appear as bare +headings — every chunk gives the reader a complete map of the document +and shows exactly where its slice belongs. -Hash-diff cache compatibility: blocks with identical (heading_path + body) -across edits produce the same hash, so the file_store can reuse old -embeddings and only call the embedding API for dirty blocks. +Two phases: -Edge extraction inlines `parse_wikilinks` directly: edges live in body -text only (bare wikilinks + Dataview line-level + Dataview inline-bracketed) -and the predicate vocabulary is closed at the `FileEdge` schema layer. -The slow path (maintainer's `enrich_links` / `discover_links` ops) handles -upgrading bare links and discovering new ones. +1. **Build phase** — fold mistletoe's flat `Document.children` into a + layered `MdNode` tree. A `section` node owns its heading + body + blocks + child subsections, established by a heading-level stack. + +2. **Chunk phase** — recursive `chunk(node, parent_section)`: + + if len(content) <= chunk_chars: + emit one chunk (TOC wrapped on top, additive to size) and return + if node is section/root: + walk children — body siblings pack as a run inside the + current section's TOC slot, subsections recurse + else (leaf body): + split by internal structure (List items / Table rows / + code lines / paragraph lines) + +Example — doc with sections A, B (containing B1), C — chunking content +of B1 produces a chunk like: + + # Doc + ## A + ## B + ### B1 + + + + ## C + +The current section (here `### B1`) is the **owner**: its slot holds +the chunk's body. All other sections appear as bare headings. + +**Budget rule**: ``chunk_chars`` constrains the **content** only — +the TOC skeleton is added as a free prefix on top of every chunk and +does NOT count toward the budget. This keeps content sizing predictable +even when a doc has a large heading outline. + +**Toggle**: ``embed_toc=False`` disables the TOC wrap entirely; chunks +become plain content with no document-level navigation prefix. The +chunking decisions themselves are unchanged — only the final emitted +text differs. + +Section integrity: a node is split only when it cannot fit as a whole. +Block-internal structure (code lines, table rows, list items) is +respected — splits land on those boundaries, never inside. + +**Part markers**: when a single leaf block (table / code fence / list / +paragraph) is too large to fit and gets split into N > 1 pieces, each +piece is annotated with a ``[Part X/N]\\n\\n`` prefix so readers know +they're seeing a fragment. Single-piece outputs are unmarked. + +Chunk identity: `hash_text(path::start::end::text)` — content-deterministic +so the file_store's hash-diff cache hits across re-parses of unchanged +sections. """ -import re +from __future__ import annotations + +from dataclasses import dataclass, field from pathlib import Path +from typing import Any import frontmatter +from mistletoe.block_token import Document +from mistletoe.markdown_renderer import MarkdownRenderer from .base_file_parser import BaseFileParser from ..component_registry import R from ...enumeration import FileSuffixEnum -from ...schema import FileChunk, FileEdge, FileNode, parse_wikilinks +from ...schema import FileChunk, FileEdge, FileNode from ...utils import hash_text +# -- Helpers -------------------------------------------------------------- + + +def _kind(node) -> str: + return type(node).__name__ + + +def _is_heading(node) -> bool: + return _kind(node) in ("Heading", "SetextHeading") + + +def _line_count(text: str) -> int: + return len(text.split("\n")) if text else 0 + + +def _heading_text(node, renderer: MarkdownRenderer) -> str: + """Heading text without `#` markers (for outline).""" + rendered = renderer.render(node).rstrip("\n") + if rendered.startswith("#"): + return rendered.lstrip("#").strip() + return rendered.split("\n", 1)[0].strip() + + +# Reserved overhead for the worst-case "[Part NNN/NNN]\n\n" prefix added +# to leaf-block split pieces. Reserved upfront in budgets so the prefix +# fits even for the chunk that just barely passed the size check. +_PART_MARKER_RESERVE = 18 + + +# -- AST tree ------------------------------------------------------------- + + +@dataclass +class MdNode: + """Composed markdown AST node. + + Three synthesised kinds wrap mistletoe blocks into a layered tree: + + * ``root`` — sole top-level node; ``children`` are bodies and/or + sections; carries no heading. + * ``section`` — synthesised from a heading + everything beneath it + until the next equal-or-shallower heading; children + are bodies and child sections. + * ``body`` — wraps one mistletoe block (paragraph / list / table / + code / quote / html / etc.); ``block`` is the + original mistletoe node, ``text`` is its rendered + markdown. + + Ranges (`start_line`, `end_line`) span the full subtree so callers can + record provenance on emitted chunks. + """ + + kind: str # "root" | "section" | "body" + heading: str | None = None + level: int = 0 + children: list["MdNode"] = field(default_factory=list) + block: Any = None + text: str = "" + start_line: int = 0 + end_line: int = 0 + + @R.register("md") class LinkedFileParser(BaseFileParser): - """Parser for Markdown files with YAML frontmatter and wikilink support.""" + """Markdown parser: frontmatter + wikilink edges + full-skeleton chunks.""" suffixes = [FileSuffixEnum.MD, FileSuffixEnum.MARKDOWN] - _HEADING_RE = re.compile(r"^(#{1,6})\s+(.*)$") - _FENCE_RE = re.compile(r"^(```|~~~)") - - def __init__(self, encoding: str = "utf-8", **kwargs): + def __init__( + self, + encoding: str = "utf-8", + chunk_chars: int = 2000, + embed_toc: bool = True, + **kwargs, + ): super().__init__(**kwargs) self.encoding = encoding + self.chunk_chars = max(100, chunk_chars) + self.embed_toc = embed_toc - async def parse(self, path: str) -> tuple[FileNode, list[FileChunk]]: + async def parse(self, path: str | Path) -> tuple[FileNode, list[FileChunk]]: file_path = Path(path) raw = file_path.read_text(encoding=self.encoding) post = frontmatter.loads(raw) stat = file_path.stat() - metadata = dict(post.metadata) - content = post.content absolute_path = str(file_path.absolute()) - edges = self._extract_edges(content) - chunks = self.chunk_markdown(content, absolute_path) + edges = self._dedup_edges(FileEdge.from_text(post.content)) + chunks = self._chunk(post.content, absolute_path) node = FileNode( path=absolute_path, st_mtime=stat.st_mtime, edges=edges, - **metadata, + **dict(post.metadata), ) return node, chunks - # -- Edge extraction -------------------------------------------------- - @staticmethod - def _extract_edges(text: str) -> list[FileEdge]: - """Body-only wikilink extraction with structural dedup.""" + def _dedup_edges(edges: list[FileEdge]) -> list[FileEdge]: seen: set[tuple] = set() out: list[FileEdge] = [] - for edge in parse_wikilinks(text or ""): - key = (edge.target, edge.predicate, edge.anchor, edge.alias, edge.embed) + for e in edges: + key = (e.link, e.predicate) if key in seen: continue seen.add(key) - out.append(edge) + out.append(e) return out - # -- Chunker ---------------------------------------------------------- + # -- Chunker entry --------------------------------------------------- - @staticmethod - def _breadcrumb(heading_path: list[str]) -> str: - return " > ".join(heading_path) if heading_path else "" - - @classmethod - def _make_block_text(cls, heading_path: list[str], body: str) -> str: - """Compose final block text: breadcrumb line (if any) + blank + body.""" - body = body.rstrip("\n") - crumb = cls._breadcrumb(heading_path) - if crumb: - return f"{crumb}\n\n{body}" if body else crumb - return body - - @classmethod - def chunk_markdown(cls, text: str, path: str) -> list[FileChunk]: - """Split markdown into AST-aware blocks (headings / paragraphs / fences).""" + def _chunk(self, text: str, path: str) -> list[FileChunk]: if not text or not text.strip(): return [] + out: list[FileChunk] = [] + with MarkdownRenderer() as renderer: + doc = Document(text) + tree = self._build_tree(doc, renderer) + self._chunk_tree(tree, tree, tree, renderer, path, out) + return out - lines = text.split("\n") - chunks: list[FileChunk] = [] + # -- Build phase: mistletoe doc → MdNode tree ------------------------- - heading_stack: list[tuple[int, str]] = [] # [(level, title)] - body_lines: list[str] = [] - body_start = 1 - in_fence = False - fence_marker = "" + def _build_tree(self, doc, renderer: MarkdownRenderer) -> MdNode: + """Fold mistletoe's flat children into a section tree. - def current_path() -> list[str]: - return [t for _, t in heading_stack] + Algorithm: walk children with a stack of open sections. Each + heading pops sections of equal-or-deeper level and pushes a new + section. Non-heading blocks attach as ``body`` children to the + current section (or to root before the first heading). + """ + root = MdNode(kind="root", level=0, start_line=1, end_line=1) + stack: list[MdNode] = [root] - def emit(block_text: str, start_line: int, end_line: int) -> None: - h = hash_text(block_text) - chunks.append( - FileChunk( - id=hash_text(f"{path}::{start_line}::{end_line}::{h}::{len(chunks)}"), - path=path, - start_line=start_line, - end_line=end_line, - text=block_text, - hash=h, - ), + for child in doc.children or []: + kind = _kind(child) + if kind == "BlankLine": + continue + if _is_heading(child): + level = max(1, getattr(child, "level", 1)) + # Close any sections of equal-or-greater level. + while len(stack) > 1 and stack[-1].level >= level: + stack.pop() + sec = MdNode( + kind="section", + heading=_heading_text(child, renderer), + level=level, + start_line=child.line_number or stack[-1].start_line, + ) + stack[-1].children.append(sec) + stack.append(sec) + else: + rendered = renderer.render(child).rstrip("\n") + if not rendered: + continue + start = child.line_number or stack[-1].start_line + body = MdNode( + kind="body", + block=child, + text=rendered, + start_line=start, + end_line=start + _line_count(rendered) - 1, + ) + stack[-1].children.append(body) + + # Propagate end_line bottom-up. + def _close(n: MdNode) -> None: + if not n.children: + if n.end_line < n.start_line: + n.end_line = n.start_line + return + for c in n.children: + _close(c) + n.end_line = max(c.end_line for c in n.children) + n.start_line = min(n.start_line or n.children[0].start_line, n.children[0].start_line) + + _close(root) + return root + + # -- Chunk phase: recursive ------------------------------------------ + + def _chunk_tree( + self, + tree: MdNode, + node: MdNode, + parent_section: MdNode, + renderer: MarkdownRenderer, + file_path: str, + out: list[FileChunk], + ) -> None: + """Try the whole subtree first; on overflow descend into children. + + ``tree`` is the whole-document root used to render the full TOC + skeleton (when ``embed_toc`` is on). ``parent_section`` is the + nearest enclosing section/root — for body nodes it's the slot + owner; for sections it becomes the slot owner when the section + itself fits whole. The ``chunk_chars`` budget only constrains + ``content`` — TOC overhead is excluded. + """ + if node.kind == "body": + owner = parent_section + content = node.text + owns_subtree = False + else: # root or section + owner = node + content = self._render_node_content(node) + owns_subtree = True + + if not content.strip(): + return + + if len(content) <= self.chunk_chars: + full = self._finalize(tree, owner, content, owns_subtree) + self._emit(full, node.start_line, node.end_line, file_path, out) + return + + if node.kind in ("root", "section"): + self._chunk_children(tree, node, renderer, file_path, out) + else: + self._split_leaf(tree, node, parent_section, renderer, file_path, out) + + def _chunk_children( + self, + tree: MdNode, + parent: MdNode, + renderer: MarkdownRenderer, + file_path: str, + out: list[FileChunk], + ) -> None: + """Walk a section's children: body run packs together, each + subsection recurses. Body runs use ``parent`` as TOC owner.""" + run: list[MdNode] = [] + + def flush_run() -> None: + nonlocal run + if run: + self._chunk_body_run(tree, run, parent, renderer, file_path, out) + run = [] + + for c in parent.children: + if c.kind == "section": + flush_run() + self._chunk_tree(tree, c, parent, renderer, file_path, out) + else: + run.append(c) + flush_run() + + def _chunk_body_run( + self, + tree: MdNode, + run: list[MdNode], + owner: MdNode, + renderer: MarkdownRenderer, + file_path: str, + out: list[FileChunk], + ) -> None: + """A run of consecutive body siblings sharing ``owner``'s slot. + + Try whole run first; on overflow greedy-pack into ``chunk_chars`` + (content-only budget); an oversized single body recurses through + ``_split_leaf``. + """ + composite = "\n\n".join(b.text for b in run) + if len(composite) <= self.chunk_chars: + full = self._finalize(tree, owner, composite, owns_subtree=False) + self._emit(full, run[0].start_line, run[-1].end_line, file_path, out) + return + + budget = self.chunk_chars + bucket: list[MdNode] = [] + bucket_chars = 0 + + def flush() -> None: + nonlocal bucket, bucket_chars + if not bucket: + return + text = "\n\n".join(b.text for b in bucket) + piece = self._finalize(tree, owner, text, owns_subtree=False) + self._emit(piece, bucket[0].start_line, bucket[-1].end_line, file_path, out) + bucket = [] + bucket_chars = 0 + + for body in run: + if len(body.text) > budget: + flush() + self._split_leaf(tree, body, owner, renderer, file_path, out) + continue + sep = 2 if bucket else 0 # "\n\n" + if bucket and bucket_chars + sep + len(body.text) > budget: + flush() + sep = 0 + bucket.append(body) + bucket_chars += sep + len(body.text) + flush() + + # -- Leaf-internal splitters ----------------------------------------- + + def _split_leaf( + self, + tree: MdNode, + body: MdNode, + owner: MdNode, + renderer: MarkdownRenderer, + file_path: str, + out: list[FileChunk], + ) -> None: + kind = _kind(body.block) if body.block is not None else "Paragraph" + if kind == "Table": + self._split_table(tree, body, owner, file_path, out) + elif kind == "CodeFence": + self._split_code(tree, body, owner, file_path, out) + elif kind == "List": + self._split_list(tree, body, owner, renderer, file_path, out) + else: + self._split_lines(tree, body, owner, file_path, out) + + def _split_table( + self, + tree: MdNode, + body: MdNode, + owner: MdNode, + file_path: str, + out: list[FileChunk], + ) -> None: + """Split a table by data rows; repeat header + separator per piece.""" + table = body.block + rendered = body.text + all_lines = rendered.split("\n") + header = "\n".join(all_lines[:2]) + data_lines = all_lines[2:] + rows = [r for r in (table.children or []) if _kind(r) == "TableRow"] + start = body.start_line + if len(rows) == len(data_lines): + units = [ + (data_lines[i], + rows[i].line_number or (start + 2 + i), + rows[i].line_number or (start + 2 + i)) + for i in range(len(data_lines)) + ] + else: + units = [ + (data_lines[i], start + 2 + i, start + 2 + i) + for i in range(len(data_lines)) + ] + self._emit_packed( + tree, owner, units, joiner="\n", + wrap=f"{header}\n{{inner}}", file_path=file_path, out=out, + ) + + def _split_code( + self, + tree: MdNode, + body: MdNode, + owner: MdNode, + file_path: str, + out: list[FileChunk], + ) -> None: + """Split a code fence by body lines; repeat opener / closer per piece.""" + code = body.block + indent = " " * (code.indentation or 0) + info = code.info_string or "" + opener = f"{indent}{code.delimiter}{info}" + closer = f"{indent}{code.delimiter}" + wrap = f"{opener}\n{{inner}}\n{closer}" + + raw = (code.children[0].content if code.children else "").rstrip("\n") + if not raw: + return + start = body.start_line + 1 + units = [ + (indent + ln, start + i, start + i) + for i, ln in enumerate(raw.split("\n")) + ] + self._emit_packed( + tree, owner, units, joiner="\n", wrap=wrap, + file_path=file_path, out=out, allow_empty=True, + ) + + def _split_list( + self, + tree: MdNode, + body: MdNode, + owner: MdNode, + renderer: MarkdownRenderer, + file_path: str, + out: list[FileChunk], + ) -> None: + """Split a list by items; greedy-pack items. + + Multi-part splits annotate each piece with ``[Part X/N]`` + (see ``_emit_parts``). An item that exceeds the budget on its + own is emitted as one part (continuity wins over hard cap). + """ + items = [c for c in (body.block.children or []) if _kind(c) == "ListItem"] + if not items: + self._split_lines(tree, body, owner, file_path, out) + return + + budget = max(64, self.chunk_chars - _PART_MARKER_RESERVE) + rendered_items = [ + (renderer.render(it).rstrip("\n"), it.line_number or body.start_line) + for it in items + ] + + parts: list[tuple[str, int, int]] = [] + bucket: list[tuple[str, int, int]] = [] + bucket_chars = 0 + + def flush() -> None: + nonlocal bucket, bucket_chars + if not bucket: + return + text = "\n".join(t for t, _, _ in bucket) + parts.append((text, bucket[0][1], bucket[-1][2])) + bucket = [] + bucket_chars = 0 + + for text, line in rendered_items: + if not text: + continue + end = line + _line_count(text) - 1 + if len(text) > budget: + # Oversized item: emit alone (overflow accepted). + flush() + parts.append((text, line, end)) + continue + sep = 1 if bucket else 0 # "\n" + if bucket and bucket_chars + sep + len(text) > budget: + flush() + sep = 0 + bucket.append((text, line, end)) + bucket_chars += sep + len(text) + flush() + + self._emit_parts(tree, owner, parts, wrap="{inner}", file_path=file_path, out=out) + + def _split_lines( + self, + tree: MdNode, + body: MdNode, + owner: MdNode, + file_path: str, + out: list[FileChunk], + ) -> None: + """Last-resort split: line-greedy. Used for paragraphs / quotes / + html / oversized list items.""" + start = body.start_line + units = [ + (line, start + i, start + i) + for i, line in enumerate(body.text.split("\n")) + ] + self._emit_packed( + tree, owner, units, joiner="\n", wrap="{inner}", + file_path=file_path, out=out, + ) + + def _emit_packed( + self, + tree: MdNode, + owner: MdNode, + units: list[tuple[str, int, int]], + joiner: str, + wrap: str, + file_path: str, + out: list[FileChunk], + allow_empty: bool = False, + ) -> None: + """Greedy-pack units into the wrap envelope; emit each bucket + as a chunk via ``_emit_parts`` (which adds ``[Part X/N]`` when + more than one piece results). + + The envelope (e.g. table header, code fence) IS counted against + ``chunk_chars`` because it's part of the chunk's content. The + TOC skeleton, when ``embed_toc`` is on, is added as a free + prefix downstream. + + A unit larger than the inner budget is emitted alone (continuity + wins over hard cap — readers can still see the overflowing line). + """ + envelope = len(wrap.replace("{inner}", "")) + inner_budget = max(64, self.chunk_chars - envelope - _PART_MARKER_RESERVE) + sep_len = len(joiner) + + parts: list[tuple[str, int, int]] = [] + bucket: list[tuple[str, int, int]] = [] + bucket_chars = 0 + + def flush() -> None: + nonlocal bucket, bucket_chars + if not bucket: + return + inner = joiner.join(t for t, _, _ in bucket) + parts.append((inner, bucket[0][1], bucket[-1][2])) + bucket = [] + bucket_chars = 0 + + for text, s, e in units: + if not text and not allow_empty: + continue + sep = sep_len if bucket else 0 + if bucket and bucket_chars + sep + len(text) > inner_budget: + flush() + sep = 0 + bucket.append((text, s, e)) + bucket_chars += sep + len(text) + flush() + + self._emit_parts(tree, owner, parts, wrap=wrap, file_path=file_path, out=out) + + def _emit_parts( + self, + tree: MdNode, + owner: MdNode, + parts: list[tuple[str, int, int]], + wrap: str, + file_path: str, + out: list[FileChunk], + ) -> None: + """Emit a list of leaf-block split parts. + + Each part is a ``(inner_text, start_line, end_line)`` triple. + ``wrap`` is a format string with ``{inner}`` substituted per + piece (e.g. ``"| header |\\n{inner}"`` for tables; ``"{inner}"`` + for plain line splits). + + When ``len(parts) > 1`` each piece is prefixed with + ``[Part X/N]\\n\\n`` so readers know it's a fragment of a + larger leaf block. A single part emits with no marker. + """ + total = len(parts) + for idx, (inner, s, e) in enumerate(parts, 1): + piece = wrap.replace("{inner}", inner) + if total > 1: + piece = f"[Part {idx}/{total}]\n\n{piece}" + full = self._finalize(tree, owner, piece, owns_subtree=False) + self._emit(full, s, e, file_path, out) + + # -- Render helpers --------------------------------------------------- + + def _finalize( + self, + tree: MdNode, + owner: MdNode, + content: str, + owns_subtree: bool, + ) -> str: + """Produce the final chunk text from raw ``content``. + + When ``embed_toc`` is on (default), wrap content with the full + document heading skeleton (see ``_render_with_full_toc``) so the + chunk shows where its slice belongs in the doc. When off, return + ``content`` unchanged — the chunk is just its own text. + + ``chunk_chars`` is checked against ``content`` upstream of this + call; the TOC skeleton is *additive* to the chunk text and does + not consume the budget. Callers wanting to know the final chunk + length must call ``len(self._finalize(...))`` themselves. + """ + if not self.embed_toc: + return content + return self._render_with_full_toc(tree, owner, content, owns_subtree) + + @classmethod + def _render_with_full_toc( + cls, + tree: MdNode, + owner: MdNode, + content: str, + owns_subtree: bool, + ) -> str: + """Render the full document heading skeleton with ``content`` + inlined under ``owner``'s heading. + + Walks the entire tree. Every section emits its heading; only the + ``owner`` slot also emits ``content``. If ``owns_subtree`` is + True, recursion stops at ``owner`` (its subsection headings are + assumed already present in ``content``); otherwise traversal + continues so descendant headings still appear in the TOC. + + Bodies are never emitted by this walk — they're brought in only + via ``content``. ``owner`` may be the root node, in which case + the content sits before the first heading. + """ + lines: list[str] = [] + + def walk(node: MdNode) -> None: + if node.kind == "section" and node.heading is not None: + lines.append(f"{'#' * max(1, node.level)} {node.heading}") + if node is owner: + if content: + lines.append(content) + if owns_subtree: + return + for c in node.children: + if c.kind == "section": + walk(c) + + walk(tree) + return "\n\n".join(p for p in lines if p) + + @classmethod + def _render_node_content(cls, node: MdNode) -> str: + """Render a node's content BENEATH its own heading. + + The node's own heading is NOT included — when `_render_with_full_toc` + emits the section's TOC entry, the slot it appends ``content`` to + already sits below that heading. Subsection headings ARE included + because they're deeper than the focused node and would otherwise + be swallowed when ``owns_subtree=True``. + """ + if node.kind == "body": + return node.text + parts: list[str] = [] + for c in node.children: + if c.kind == "section": + sub_heading = f"{'#' * max(1, c.level)} {c.heading or ''}" + inner = cls._render_node_content(c) + parts.append(sub_heading + ("\n\n" + inner if inner else "")) + else: + if c.text: + parts.append(c.text) + return "\n\n".join(parts) + + # -- Emit ------------------------------------------------------------- + + @staticmethod + def _emit( + text: str, start_line: int, end_line: int, path: str, out: list[FileChunk], + ) -> None: + chunk_id = hash_text(f"{path}::{start_line}::{end_line}::{text}") + out.append(FileChunk( + id=chunk_id, + path=path, + start_line=start_line, + end_line=end_line, + text=text, + )) + + +# -- CLI: parse a markdown file and print chunks + edges ------------------ + + +def _main() -> None: + """Parse a markdown file and print its edges + chunks for inspection. + + Usage: + python -m reme2.component.file_parser.linked_file_parser [--chunk-chars N] + """ + import argparse + import asyncio + + ap = argparse.ArgumentParser( + description="Parse a markdown file with LinkedFileParser and dump chunks + edges.", + ) + ap.add_argument("path", help="Path to a markdown file.") + ap.add_argument( + "--chunk-chars", type=int, default=2000, + help="Max characters per chunk content (default: 2000). " + "Excludes TOC skeleton when embed_toc is on.", + ) + ap.add_argument( + "--no-toc", action="store_true", + help="Disable the full-doc TOC skeleton wrap; chunks become plain content.", + ) + ap.add_argument( + "--show-edges", action="store_true", + help="Print extracted FileEdges before chunks.", + ) + ap.add_argument( + "--preview", type=int, default=0, + help="Truncate each chunk to N chars in output (0 = full text).", + ) + args = ap.parse_args() + + parser = LinkedFileParser( + chunk_chars=args.chunk_chars, + embed_toc=not args.no_toc, + ) + node, chunks = asyncio.run(parser.parse(args.path)) + + print(f"file: {node.path}") + print(f"chunk_chars: {args.chunk_chars}") + print(f"embed_toc: {parser.embed_toc}") + print(f"chunks: {len(chunks)}") + print(f"chars total: {sum(len(c.text) for c in chunks)}") + if chunks: + sizes = [len(c.text) for c in chunks] + print(f"chars min/avg/max: {min(sizes)} / {sum(sizes)//len(sizes)} / {max(sizes)}") + if args.show_edges: + print(f"\nedges ({len(node.edges)}):") + for e in node.edges: + print( + f" → {e.link}" + + (f" predicate={e.predicate}" if e.predicate else "") + + (f" anchor={e.anchor}" if e.anchor else "") ) - def flush_body(end_line: int) -> None: - nonlocal body_lines, body_start - if not body_lines: - return - # Strip leading/trailing blank lines from the block (paragraph - # boundaries eat their own newline, but whitespace can sneak in - # via the fence path). - while body_lines and not body_lines[0].strip(): - body_lines.pop(0) - body_start += 1 - while body_lines and not body_lines[-1].strip(): - body_lines.pop() - end_line -= 1 - if not body_lines: - body_lines = [] - return - raw_body = "\n".join(body_lines) - block_text = cls._make_block_text(current_path(), raw_body) - emit(block_text, body_start, end_line) - body_lines = [] + for i, c in enumerate(chunks): + print(f"\n{'=' * 72}") + print(f"chunk {i} lines {c.start_line}-{c.end_line} {len(c.text)} chars") + print("-" * 72) + text = c.text if args.preview <= 0 else c.text[: args.preview] + print(text) + if args.preview > 0 and len(c.text) > args.preview: + print(f"... ({len(c.text) - args.preview} more chars truncated)") - for i, line in enumerate(lines, 1): - stripped = line.strip() - # Code fences: keep contents intact, no inner splits. - if not in_fence and cls._FENCE_RE.match(stripped): - if body_lines: - flush_body(end_line=i - 1) - in_fence = True - fence_marker = stripped[:3] - body_lines = [line] - body_start = i - continue - if in_fence: - body_lines.append(line) - if stripped.startswith(fence_marker): - in_fence = False - fence_marker = "" - flush_body(end_line=i) - continue +if __name__ == "__main__": + _main() - # ATX heading: closes prior block, opens a new section. - m = cls._HEADING_RE.match(line) - if m: - if body_lines: - flush_body(end_line=i - 1) - level = len(m.group(1)) - title = m.group(2).strip() - heading_stack = [(lv, t) for lv, t in heading_stack if lv < level] - heading_stack.append((level, title)) - - # Heading line itself becomes a block (so the heading text is - # searchable as its own unit). - block_text = cls._make_block_text(current_path(), "") - emit(block_text, i, i) - body_start = i + 1 - continue - - # Blank line: paragraph boundary. - if not stripped: - if body_lines: - flush_body(end_line=i - 1) - body_start = i + 1 - continue - - # Regular content line. - if not body_lines: - body_start = i - body_lines.append(line) - - if body_lines: - flush_body(end_line=len(lines)) - - return chunks diff --git a/reme2/memory/maintainer.py b/reme2/memory/maintainer.py index 3fc1d566..1a4eb721 100644 --- a/reme2/memory/maintainer.py +++ b/reme2/memory/maintainer.py @@ -418,7 +418,7 @@ class Maintainer(BaseStep): 2. Read body, ask the LLM (one call per file) to assign one of `ALLOWED_PREDICATES` to each bare target — or 'skip'. 3. For each accepted (target, predicate), locate each bare - occurrence span via `parse_wikilinks` and build a unique + occurrence span via `FileEdge.from_text` and build a unique context window snippet. 4. Emit EnrichOp(path, target, predicate, old_string, new_string, confidence, reason). The apply step calls memory_update. diff --git a/reme2/memory/memory_io.py b/reme2/memory/memory_io.py index 44781393..6be0c6af 100644 --- a/reme2/memory/memory_io.py +++ b/reme2/memory/memory_io.py @@ -39,7 +39,7 @@ from pathlib import Path import frontmatter from ..schema import ChunkFilter, FileChunk, FileNode, extract_wikilinks -from ..schema.file_edge import WIKILINK_RE +from ..schema.file_edge import _WIKILINK_RE from ..utils.wikilink_resolver import ( resolve_wikilink as _resolve_wikilink, wikilink_candidates, @@ -119,8 +119,6 @@ def _edge_to_dict(node, edge) -> dict: "metadata": node.metadata, "predicate": edge.predicate, "anchor": edge.anchor, - "alias": edge.alias, - "embed": edge.get_embeddings, } @@ -232,7 +230,7 @@ def _replace_wikilink_targets(text: str, mapping: dict[str, str]) -> str: return m.group(0).replace(target_raw, mapping[target], 1) return m.group(0) - return WIKILINK_RE.sub(sub, text) + return _WIKILINK_RE.sub(sub, text) def create_file( diff --git a/reme2/schema/__init__.py b/reme2/schema/__init__.py index d474ed8c..5a6a5617 100644 --- a/reme2/schema/__init__.py +++ b/reme2/schema/__init__.py @@ -5,7 +5,7 @@ from .as_msg_stat import AsBlockStat, AsMsgStat from .emb_node import EmbNode from .chunk_filter import ChunkFilter from .file_chunk import FileChunk -from .file_edge import FileEdge, extract_wikilinks, parse_wikilinks +from .file_edge import FileEdge, extract_wikilinks from .file_node import FileNode from .request import Request from .response import Response @@ -26,5 +26,4 @@ __all__ = [ "Response", "StreamChunk", "extract_wikilinks", - "parse_wikilinks", ] diff --git a/reme2/schema/file_edge.py b/reme2/schema/file_edge.py index f8866572..c2f61b80 100644 --- a/reme2/schema/file_edge.py +++ b/reme2/schema/file_edge.py @@ -1,16 +1,30 @@ """FileEdge — typed wikilink edge between vault files. -This module is the single source of truth for both the edge **schema** -and the **inline parser** that recovers edges from body text. Edges -live exclusively in body text (frontmatter is not walked); the -predicate vocabulary is **open** — any identifier-shaped token -(`[A-Za-z][A-Za-z0-9_]*`) that the parser sees is preserved verbatim -on `FileEdge.predicate`. Vocabulary curation, if any, is the -maintainer's job, not the schema's. +Single source of truth for the edge **schema** and the **inline parser** +(`FileEdge.from_text`) that recovers edges from body text. Edges live +exclusively in body text (frontmatter is not walked); the predicate +vocabulary is **open** — any identifier-shaped token +(`[A-Za-z][A-Za-z0-9_]*`) the parser sees is preserved verbatim on +`FileEdge.predicate`. Vocabulary curation, if any, is the maintainer's +job, not the schema's. -## Inline forms recognised by `parse_wikilinks` +## Inline forms recognised by `FileEdge.from_text` + [[X]] bare wikilink → predicate=None + extends:: [[X]] line-level Dataview → predicate="extends" + [extends:: [[X]]] inline-bracketed → predicate="extends" +Multi-target — every wikilink under one typed context inherits its +predicate (any separator works, not just commas): + + extends:: [[A]], [[B]] line-level multi → 2 edges, both "extends" + extends:: [[A]] and [[B]] prose-style multi → 2 edges, both "extends" + [concerns:: [[A]], [[B]]] inline multi → 2 edges, both "concerns" + extends:: [[A#s1]], [[B#s2]] multi w/ anchors → anchors preserved per link + +Context precedence is **inline-bracketed > line-level > bare** — +a wikilink inside a `[predicate:: …]` envelope is typed by that +envelope even if the line happens to start `predicate:: …`. """ from __future__ import annotations @@ -20,35 +34,19 @@ import re from pydantic import BaseModel, ConfigDict, Field -# -- Schema --------------------------------------------------------------- +# -- Regexes (module-private) --------------------------------------------- - -class FileEdge(BaseModel): - """5-field minimal edge model. No provenance, no confidence.""" - - model_config = ConfigDict(extra="forbid") - - target: str = Field(..., description="Raw wikilink target as written in source.") - predicate: str | None = Field( - default=None, - description="Typed-edge predicate (Dataview-style). None for bare [[X]].", - ) - anchor: str | None = Field(default=None, description="Heading or block anchor (after #).") - alias: str | None = Field(default=None, description="Display alias (after |).") - embed: bool = Field(default=False, description="True for `![[X]]` embed prefix.") - - -# -- Regexes -------------------------------------------------------------- - -# Bare wikilink. `(?:!)?` is non-capturing so `m.group(1)` stays the -# target — read the embed prefix off `m.group(0).startswith("!")`. -WIKILINK_RE = re.compile( +# Wikilink. The optional `!` embed marker and `|alias` are matched but +# not captured — both are presentational and dropped from the edge. +# `target` is the file part; `anchor` is the optional `#…` suffix; we +# rejoin them into a single `link` string at edge-construction time. +_WIKILINK_RE = re.compile( r""" (?:!)? \[\[ (?P[^\]\|\#\n]+?) (?:\#(?P[^\]\|\n]+))? - (?:\|(?P[^\]\n]+))? + (?:\|[^\]\n]+)? \]\] """, re.VERBOSE, @@ -56,50 +54,28 @@ WIKILINK_RE = re.compile( # Line-level Dataview field. Anchored MULTILINE; allows leading bullet # (`-`/`*`/`+`) so `- extends:: [[X]]` works inside Markdown lists. -# Predicate identifier follows Dataview convention: letter, then -# letters / digits / underscore. -DATAVIEW_LINE_RE = re.compile( +_DATAVIEW_LINE_RE = re.compile( r"^[ \t]*(?:[-*+][ \t]+)?(?P[A-Za-z][A-Za-z0-9_]*)\s*::\s*(?P.+?)\s*$", re.MULTILINE, ) -# Inline-bracketed field: opens with `[predicate::`. The closing `]` is -# located by `_iter_inline_fields` via depth-counted bracket scan because -# the value may contain `[[wikilink]]` whose inner `[[…]]` brackets are -# part of the value, not field delimiters. +# Inline-bracketed field opener: `[predicate::`. The matching `]` is +# located by depth-counted scan because the value may contain +# `[[wikilink]]` whose inner brackets are part of the value. _INLINE_FIELD_OPEN_RE = re.compile(r"\[(?P[A-Za-z][A-Za-z0-9_]*)\s*::\s*") -# -- Internal helpers ----------------------------------------------------- +def _iter_inline_fields(text: str) -> list[tuple[int, int, str]]: + """Find inline-bracketed `[predicate:: …]` field spans by depth scan. - -def _edge_from_wm(wm: re.Match, *, predicate: str | None) -> FileEdge: - anchor = wm.group("anchor") - alias = wm.group("alias") - return FileEdge( - target=wm.group("target").strip(), - anchor=anchor.strip() if anchor else None, - alias=alias.strip() if alias else None, - embed=wm.group(0).startswith("!"), - predicate=predicate, - ) - - -def _iter_inline_fields(text: str) -> list[tuple[int, int, str, int]]: - """Find inline-bracketed `[predicate:: …]` fields by depth scan. - - Returns ``(start, end, predicate, value_start)`` tuples — `value_start` - is the absolute offset where the value begins inside `text`, used to - project wikilink spans back to absolute positions for dedup. - - Newlines terminate the scan: an inline field that spans a line break - is treated as malformed and skipped (matches Dataview's parser). + Returns a list of ``(start, end, predicate)`` triples. Newlines + terminate the scan: an inline field that spans a line break is + treated as malformed (matches Dataview semantics) and skipped. """ - out: list[tuple[int, int, str, int]] = [] + out: list[tuple[int, int, str]] = [] for m in _INLINE_FIELD_OPEN_RE.finditer(text): - value_start = m.end() depth = 1 # the outer '[' was the regex's first character - i = value_start + i = m.end() n = len(text) while i < n: c = text[i] @@ -110,65 +86,139 @@ def _iter_inline_fields(text: str) -> list[tuple[int, int, str, int]]: elif c == "]": depth -= 1 if depth == 0: - out.append((m.start(), i + 1, m.group("predicate"), value_start)) + out.append((m.start(), i + 1, m.group("predicate"))) break i += 1 return out -# -- Public parsing API --------------------------------------------------- +# -- Schema --------------------------------------------------------------- + + +class FileEdge(BaseModel): + """2-field minimal edge model: ``link`` + ``predicate``. + + ``link`` preserves the wikilink as written, including any ``#anchor`` + suffix (e.g. ``"X"`` or ``"X#sec"``). The presentational ``|alias`` + and ``!`` embed prefix are discarded by the parser. ``path`` and + ``anchor`` are derived ``@property``s that split ``link`` on the + first ``#`` for callers that want either part without re-splitting + the string themselves. + """ + + model_config = ConfigDict(extra="forbid") + + link: str = Field( + ..., + description=( + "Wikilink as written, preserving any '#anchor' suffix " + "(e.g. 'X' or 'X#sec'). Display alias and embed prefix discarded." + ), + ) + predicate: str | None = Field( + default=None, + description="Typed-edge predicate (Dataview-style). None for bare [[X]].", + ) + + @property + def path(self) -> str: + """File-part of ``link`` (before any ``#anchor``). + + For ``"X"`` returns ``"X"``; for ``"X#sec"`` returns ``"X"``; + for ``"topics/Foo#bar"`` returns ``"topics/Foo"``. Always + returns a non-empty string because ``link`` is required and the + regex never matches an empty target. + """ + return self.link.split("#", 1)[0].strip() + + @property + def anchor(self) -> str | None: + """Heading or block anchor parsed from ``link`` (text after first ``#``). + + Returns ``None`` if the link has no anchor or the anchor is empty. + """ + if "#" not in self.link: + return None + tail = self.link.split("#", 1)[1].strip() + return tail or None + + @classmethod + def _from_match(cls, wm: re.Match, *, predicate: str | None) -> FileEdge: + target = wm.group("target").strip() + anchor = wm.group("anchor") + link = f"{target}#{anchor.strip()}" if anchor else target + return cls(link=link, predicate=predicate) + + @classmethod + def from_text(cls, text: str) -> list[FileEdge]: + """Extract all edges from body text in source order. + + Single pass: every wikilink in the text becomes one edge, and + its ``predicate`` is decided by the surrounding context with + precedence **inline-bracketed > line-level > bare**: + + * ``[predicate:: [[X]]]`` → ``predicate="predicate"`` + * ``predicate:: [[X]]`` → ``predicate="predicate"`` (line-level Dataview) + * ``[[X]]`` → ``predicate=None`` (bare) + + No consumed-span bookkeeping, no second sort: ``finditer`` + already yields wikilinks in source order, and per-position + classification is unambiguous. + """ + if not text: + return [] + + # Inline-bracketed `[predicate:: ...]` envelopes need a + # depth-counted scan (regex can't match balanced `[[…]]` inside). + inline_spans = _iter_inline_fields(text) + + return [ + cls._from_match(wm, predicate=_predicate_for(text, wm.start(), inline_spans)) + for wm in _WIKILINK_RE.finditer(text) + ] + + +def _predicate_for( + text: str, + pos: int, + inline_spans: list[tuple[int, int, str]], +) -> str | None: + """Resolve the predicate governing a wikilink at offset ``pos``. + + Checks the two typed-edge contexts in precedence order; falls + through to ``None`` (bare) when neither applies. + """ + # 1. Inline-bracketed envelope `[predicate:: …]` containing pos. + for field_start, field_end, predicate in inline_spans: + if field_start <= pos < field_end: + return predicate + + # 2. Line-level `predicate:: value` whose value range covers pos. + line_start = text.rfind("\n", 0, pos) + 1 + line_end = text.find("\n", pos) + if line_end == -1: + line_end = len(text) + m = _DATAVIEW_LINE_RE.match(text[line_start:line_end]) + if m and line_start + m.start("value") <= pos: + return m.group("predicate") + + # 3. Bare wikilink — no predicate. + return None + + +# -- Public utility (target-only fast path) ------------------------------- def extract_wikilinks(text: str) -> list[str]: - """Targets-only list of wikilinks in body text (no dedup). + """Flat list of wikilink **file targets** in body text (no dedup). - Used by callers that need to follow links structurally without caring - about predicates (e.g. the ingestor's auto-discovery hint). + Returns just the file part of each wikilink (before any ``#anchor``) + because callers feed the result to `resolve_wikilink`, which matches + against vault file stems / paths and would not recognise an anchor + suffix. Single regex pass — cheaper than `FileEdge.from_text` when + callers don't need predicates (e.g. ingestor's auto-discovery hint, + memory_io anchor resolution). """ if not text: return [] - return [m.group("target").strip() for m in WIKILINK_RE.finditer(text)] - - -def parse_wikilinks(text: str) -> list[FileEdge]: - """Structured parse of all three edge forms. - - Order is by source position. Wikilinks attributed to an inline- - bracketed or line-level field are not also reported as bare; the - same wikilink span is consumed exactly once. - """ - if not text: - return [] - - consumed: list[tuple[int, int]] = [] - items: list[tuple[int, FileEdge]] = [] - - # 1. Inline-bracketed: most specific, handled first. - for field_start, field_end, predicate, value_start in _iter_inline_fields(text): - value = text[value_start : field_end - 1] - for wm in WIKILINK_RE.finditer(value): - items.append((field_start, _edge_from_wm(wm, predicate=predicate))) - consumed.append((field_start, field_end)) - - # 2. Line-level: skip if entirely inside an inline-bracketed span. - for m in DATAVIEW_LINE_RE.finditer(text): - if any(cs <= m.start() and m.end() <= ce for cs, ce in consumed): - continue - predicate = m.group("predicate") - value = m.group("value") - value_start = m.start("value") - for wm in WIKILINK_RE.finditer(value): - wl_start = value_start + wm.start() - wl_end = value_start + wm.end() - items.append((wl_start, _edge_from_wm(wm, predicate=predicate))) - consumed.append((wl_start, wl_end)) - - # 3. Bare: anything left over. - for wm in WIKILINK_RE.finditer(text): - s, e = wm.span() - if any(cs <= s and e <= ce for cs, ce in consumed): - continue - items.append((s, _edge_from_wm(wm, predicate=None))) - - items.sort(key=lambda pair: pair[0]) - return [edge for _, edge in items] + return [m.group("target").strip() for m in _WIKILINK_RE.finditer(text)] diff --git a/tests/test_file_edge.py b/tests/test_file_edge.py new file mode 100644 index 00000000..6c347351 --- /dev/null +++ b/tests/test_file_edge.py @@ -0,0 +1,348 @@ +"""FileEdge unit tests — body-only edge extraction. + +Covers the three legal inline forms (bare / line-level Dataview / +inline-bracketed Dataview), multi-target expansion, dedup against +typed wrappers, open-vocabulary predicate pass-through, the explicit +decision that frontmatter is no longer walked for links, and the +2-field schema (`link` + `predicate`) with `anchor` as a derived +property and `alias` / `embed` discarded at parse time. +""" + +from reme2.schema.file_edge import FileEdge, extract_wikilinks + + +# -------------------------------------------------------------------------- +# Bare wikilinks +# -------------------------------------------------------------------------- + + +def test_bare_wikilink(): + edges = FileEdge.from_text("see [[X]]") + assert len(edges) == 1 + assert edges[0].link == "X" + assert edges[0].predicate is None + assert edges[0].anchor is None + + +def test_anchor_alias_embed_collapse_into_link(): + """`![[X#sec|Alias]]` → link='X#sec' (alias and embed dropped).""" + edges = FileEdge.from_text("![[X#sec|Alias]]") + assert len(edges) == 1 + assert edges[0].link == "X#sec" + # anchor is a derived property parsed from link. + assert edges[0].anchor == "sec" + + +def test_alias_only_drops_to_link(): + edges = FileEdge.from_text("[[X|Alias]]") + assert len(edges) == 1 + assert edges[0].link == "X" + assert edges[0].anchor is None + + +def test_embed_only_drops_to_link(): + edges = FileEdge.from_text("![[X]]") + assert len(edges) == 1 + assert edges[0].link == "X" + + +# -------------------------------------------------------------------------- +# Line-level Dataview +# -------------------------------------------------------------------------- + + +def test_line_level_field(): + edges = FileEdge.from_text("extends:: [[Source Topic]]") + assert len(edges) == 1 + assert edges[0].link == "Source Topic" + assert edges[0].predicate == "extends" + + +def test_line_level_multi_target(): + edges = FileEdge.from_text("concerns:: [[A]], [[B]], [[C]]") + assert [(e.link, e.predicate) for e in edges] == [ + ("A", "concerns"), + ("B", "concerns"), + ("C", "concerns"), + ] + + +def test_line_level_with_bullet(): + edges = FileEdge.from_text("- extends:: [[X]]\n * concerns:: [[Y]]") + assert [(e.link, e.predicate) for e in edges] == [ + ("X", "extends"), + ("Y", "concerns"), + ] + + +# -------------------------------------------------------------------------- +# Multi-edge cases — many wikilinks under one or several typed contexts +# -------------------------------------------------------------------------- + + +def test_multi_target_with_anchors_preserves_each(): + """Each comma-separated target keeps its own anchor in `link`.""" + edges = FileEdge.from_text("extends:: [[A#sec1]], [[B#sec2]], [[C]]") + assert [(e.link, e.anchor, e.predicate) for e in edges] == [ + ("A#sec1", "sec1", "extends"), + ("B#sec2", "sec2", "extends"), + ("C", None, "extends"), + ] + + +def test_multi_target_non_comma_separator_still_typed(): + """Wikilinks anywhere in the value range (not just comma-separated) + inherit the line's predicate. Useful for prose-style fields.""" + edges = FileEdge.from_text("extends:: [[A]] and also [[B]]") + assert [(e.link, e.predicate) for e in edges] == [ + ("A", "extends"), + ("B", "extends"), + ] + + +def test_multi_dataview_lines_each_multi_target(): + """Multiple Dataview lines each with multi-target → all edges typed + by their respective line's predicate.""" + edges = FileEdge.from_text( + "extends:: [[A]], [[B]]\nrelates:: [[C]], [[D]]" + ) + assert [(e.link, e.predicate) for e in edges] == [ + ("A", "extends"), + ("B", "extends"), + ("C", "relates"), + ("D", "relates"), + ] + + +def test_inline_bracketed_then_bare_on_same_line(): + """Inline-bracketed governs only the wikilinks inside its brackets; + a trailing bare wikilink on the same line stays bare.""" + edges = FileEdge.from_text("[ext:: [[A]]] then [[B]]") + assert [(e.link, e.predicate) for e in edges] == [ + ("A", "ext"), + ("B", None), + ] + + +def test_mid_line_dataview_like_not_typed(): + """``predicate::`` only counts at line start (modulo bullet) — + a `predicate::` mid-line is just prose, so its wikilinks are bare.""" + edges = FileEdge.from_text("[ext:: [[A]]] and concerns:: [[B]]") + assert [(e.link, e.predicate) for e in edges] == [ + ("A", "ext"), + ("B", None), # `concerns::` mid-line is not Dataview + ] + + +# -------------------------------------------------------------------------- +# Inline-bracketed Dataview +# -------------------------------------------------------------------------- + + +def test_inline_bracketed(): + edges = FileEdge.from_text("This [extends:: [[Y]]] something else.") + assert len(edges) == 1 + assert edges[0].link == "Y" + assert edges[0].predicate == "extends" + + +def test_inline_bracketed_multi_target(): + edges = FileEdge.from_text("[concerns:: [[A]], [[B]]]") + assert [(e.link, e.predicate) for e in edges] == [ + ("A", "concerns"), + ("B", "concerns"), + ] + + +def test_inline_bracketed_skips_cross_line(): + # A `[predicate:: ...]` that spans a newline is malformed → the inner + # wikilink falls back to bare; the unmatched `[` does not eat tail text. + edges = FileEdge.from_text("[extends:: [[X]]\nbad]") + assert len(edges) == 1 + assert edges[0].link == "X" + assert edges[0].predicate is None + + +# -------------------------------------------------------------------------- +# Dedup: a wikilink inside a typed wrapper should not double-emit +# -------------------------------------------------------------------------- + + +def test_inline_bracketed_does_not_double_emit(): + edges = FileEdge.from_text("see [extends:: [[X]]] again.") + assert len(edges) == 1 + assert edges[0].predicate == "extends" + + +def test_line_level_value_does_not_double_emit(): + edges = FileEdge.from_text("extends:: [[X]]") + assert len(edges) == 1 + + +def test_typed_and_bare_coexist_for_same_target(): + edges = FileEdge.from_text("extends:: [[X]]\nFree text mentioning [[X]] again.") + links_preds = sorted(((e.link, e.predicate or "") for e in edges)) + assert links_preds == [("X", ""), ("X", "extends")] + + +# -------------------------------------------------------------------------- +# Open-vocabulary predicates — any identifier-shaped token passes through +# -------------------------------------------------------------------------- + + +def test_arbitrary_predicate_preserved(): + edges = FileEdge.from_text("anything_goes:: [[X]]") + assert len(edges) == 1 + assert edges[0].link == "X" + assert edges[0].predicate == "anything_goes" + + +def test_inline_arbitrary_predicate_preserved(): + edges = FileEdge.from_text("[wat:: [[X]]]") + assert len(edges) == 1 + assert edges[0].predicate == "wat" + + +def test_predicate_must_be_identifier_shaped(): + # A leading digit fails the regex `[A-Za-z][A-Za-z0-9_]*` so the line is + # not recognised as a Dataview field — the wikilink falls back to bare. + edges = FileEdge.from_text("123bad:: [[X]]") + assert len(edges) == 1 + assert edges[0].link == "X" + assert edges[0].predicate is None + + +def test_file_edge_accepts_any_predicate_string(): + e = FileEdge(link="X", predicate="totally_made_up") + assert e.predicate == "totally_made_up" + + +# -------------------------------------------------------------------------- +# Frontmatter is NOT walked for links — explicit regression +# -------------------------------------------------------------------------- + + +def test_frontmatter_links_block_is_ignored(): + # Even if a YAML-shaped string sits at the top of body, FileEdge.from_text + # only operates on body. We pass body text directly here, so this test + # asserts the API surface no longer accepts a metadata dict. + import inspect + + sig = inspect.signature(FileEdge.from_text) + assert list(sig.parameters.keys()) == ["text"], ( + "FileEdge.from_text should accept body text only — frontmatter walk removed" + ) + + +def test_no_frontmatter_walker_exported(): + from reme2.schema import file_edge as fe + + for removed in ( + "parse_wikilinks_from_metadata", + "extract_wikilinks_from_metadata", + "extract_inline_fields", + "extract_typed_edges", + "InlineField", + ): + assert not hasattr(fe, removed), ( + f"{removed} should have been removed when YAML edges were dropped" + ) + + +# -------------------------------------------------------------------------- +# FileEdge schema +# -------------------------------------------------------------------------- + + +def test_file_edge_extra_forbid(): + import pytest + from pydantic import ValidationError + + with pytest.raises(ValidationError): + FileEdge(link="X", target="X") # type: ignore[call-arg] + + +def test_file_edge_minimal_field_set(): + """Stored fields are just `link` and `predicate`. `path` and `anchor` + are `@property`s (not stored fields) so they shouldn't appear in + `model_dump()`. `target` / `alias` / `embed` were dropped entirely.""" + e = FileEdge(link="X") + dumped = e.model_dump() + assert set(dumped.keys()) == {"link", "predicate"} + for removed in ("target", "alias", "embed", "anchor", "path"): + assert removed not in dumped + + +def test_anchor_property_parses_from_link(): + """`anchor` is derived from `link`, not stored separately.""" + assert FileEdge(link="X").anchor is None + assert FileEdge(link="X#sec").anchor == "sec" + assert FileEdge(link="X#sec#more").anchor == "sec#more" + # Empty anchor is treated as no anchor. + assert FileEdge(link="X#").anchor is None + assert FileEdge(link="X# ").anchor is None + + +def test_path_property_parses_from_link(): + """`path` is derived from `link` — file part before any `#anchor`.""" + assert FileEdge(link="X").path == "X" + assert FileEdge(link="X#sec").path == "X" + assert FileEdge(link="topics/Foo").path == "topics/Foo" + assert FileEdge(link="topics/Foo#bar").path == "topics/Foo" + # Multiple '#' — only first splits; the rest live in anchor. + assert FileEdge(link="X#a#b").path == "X" + assert FileEdge(link="X#a#b").anchor == "a#b" + # Empty anchor → path is still the full prefix. + assert FileEdge(link="X#").path == "X" + + +def test_path_anchor_roundtrip_via_link(): + """Reconstructing `link` from `path` + `anchor` yields the original.""" + for link in ("X", "X#sec", "topics/Foo", "topics/Foo#bar", "X#a#b"): + e = FileEdge(link=link) + rebuilt = e.path if not e.anchor else f"{e.path}#{e.anchor}" + assert rebuilt == link, f"{link!r} → {rebuilt!r}" + + +def test_anchor_is_not_constructor_arg(): + """Since `anchor` and `path` are properties, passing them to the + constructor should fail (extra='forbid').""" + import pytest + from pydantic import ValidationError + + with pytest.raises(ValidationError): + FileEdge(link="X", anchor="sec") # type: ignore[call-arg] + with pytest.raises(ValidationError): + FileEdge(link="X", path="X") # type: ignore[call-arg] + + +# -------------------------------------------------------------------------- +# Back-compat: extract_wikilinks returns flat target list (used by ingestor) +# -------------------------------------------------------------------------- + + +def test_extract_wikilinks_flat_targets(): + targets = extract_wikilinks("see [[X]] and extends:: [[Y]] and [extends:: [[Z]]]") + assert targets == ["X", "Y", "Z"] + + +def test_extract_wikilinks_strips_anchor(): + """`extract_wikilinks` returns just the file part — anchor stripped + so callers can feed it to `resolve_wikilink`.""" + targets = extract_wikilinks("see [[X#sec]] and ![[Y#a|alias]]") + assert targets == ["X", "Y"] + + +# -------------------------------------------------------------------------- +# Source ordering stability +# -------------------------------------------------------------------------- + + +def test_edges_sorted_by_source_position(): + body = ( + "intro [[First]] then\n" + "extends:: [[Second]]\n" + "tail [[Third]]\n" + ) + edges = FileEdge.from_text(body) + assert [e.link for e in edges] == ["First", "Second", "Third"] diff --git a/tests/test_md_chunker.py b/tests/test_md_chunker.py new file mode 100644 index 00000000..5c12cd88 --- /dev/null +++ b/tests/test_md_chunker.py @@ -0,0 +1,726 @@ +"""Markdown AST chunker tests — full-skeleton TOC + inlined content. + +Each chunk renders the **complete heading skeleton of the document** +(every heading, top-to-bottom) with the chunk's content inlined under +the section that owns it. Sections that don't own this chunk's content +appear as bare headings — every chunk gives the reader a full document +map. + +Covers: +* Tree build: heading-stack folding, section ranges, body wrap-up. +* Whole-fit: small docs / sections emit as a single chunk. +* Skeleton completeness: every chunk lists every doc heading. +* Owner positioning: content sits under the right section's heading. +* Body run packing: bodies under one section share the same owner slot. +* Subsection recursion: each subsection chunks under the same skeleton + with its own owner slot. +* Leaf split: lists / tables / code fences / paragraphs split internally + with their structural header (table separator, code fence, list bullet) + preserved per piece — and the full doc skeleton wraps each piece. +""" + +from reme2.component.file_parser.linked_file_parser import ( + LinkedFileParser, + MdNode, +) + + +def _parser(chunk_chars: int, embed_toc: bool = True) -> LinkedFileParser: + """Construct a parser without invoking BaseComponent.__init__ (no app context).""" + p = LinkedFileParser.__new__(LinkedFileParser) + p.encoding = "utf-8" + p.chunk_chars = chunk_chars + p.embed_toc = embed_toc + return p + + +def _all_headings(text: str) -> list[str]: + """All markdown heading lines in `text`, in order.""" + return [ln.strip() for ln in text.split("\n") if ln.lstrip().startswith("#")] + + +# -------------------------------------------------------------------------- +# Tree build +# -------------------------------------------------------------------------- + + +def test_tree_groups_under_headings(): + txt = ( + "# Top\n" + "para1\n" + "\n" + "## Sub A\n" + "para2\n" + "\n" + "### Deeper\n" + "para3\n" + "\n" + "## Sub B\n" + "para4\n" + ) + from mistletoe.block_token import Document + from mistletoe.markdown_renderer import MarkdownRenderer + + p = _parser(2000) + with MarkdownRenderer() as r: + tree = p._build_tree(Document(txt), r) + + assert tree.kind == "root" + assert len(tree.children) == 1 + h1 = tree.children[0] + assert h1.kind == "section" and h1.heading == "Top" and h1.level == 1 + kinds = [c.kind for c in h1.children] + assert kinds == ["body", "section", "section"] + sub_a, sub_b = h1.children[1], h1.children[2] + assert sub_a.heading == "Sub A" and sub_a.level == 2 + assert sub_b.heading == "Sub B" and sub_b.level == 2 + assert [c.kind for c in sub_a.children] == ["body", "section"] + deeper = sub_a.children[1] + assert deeper.heading == "Deeper" and deeper.level == 3 + + +def test_tree_handles_body_before_first_heading(): + txt = "intro paragraph\n\n# H1\nbody\n" + from mistletoe.block_token import Document + from mistletoe.markdown_renderer import MarkdownRenderer + + p = _parser(2000) + with MarkdownRenderer() as r: + tree = p._build_tree(Document(txt), r) + assert [c.kind for c in tree.children] == ["body", "section"] + + +def test_tree_section_pop_on_equal_level(): + txt = "# Top\n## A\nx\n## B\ny\n" + from mistletoe.block_token import Document + from mistletoe.markdown_renderer import MarkdownRenderer + + p = _parser(2000) + with MarkdownRenderer() as r: + tree = p._build_tree(Document(txt), r) + h1 = tree.children[0] + assert [c.heading for c in h1.children if c.kind == "section"] == ["A", "B"] + + +# -------------------------------------------------------------------------- +# Whole-fit emit +# -------------------------------------------------------------------------- + + +def test_small_doc_emits_one_chunk(): + txt = "# Top\nhello world\n" + chunks = _parser(500)._chunk(txt, "/x.md") + assert len(chunks) == 1 + text = chunks[0].text + assert "# Top" in text and "hello world" in text + + +# -------------------------------------------------------------------------- +# Full-skeleton TOC: every chunk shows every heading +# -------------------------------------------------------------------------- + + +def test_every_chunk_lists_every_doc_heading(): + """No matter which slice is being chunked, the chunk text must + contain every heading in the document — that's what 'complete TOC + structure' means.""" + txt = ( + "# Doc\n" + "intro paragraph here\n" + "\n" + "## Section A\n" + "para A content here long\n" + "\n" + "### Subsection\n" + "deeper content here long\n" + "\n" + "## Section B\n" + "para B content long\n" + ) + chunks = _parser(80)._chunk(txt, "/x.md") + assert len(chunks) >= 3 + expected_headings = {"# Doc", "## Section A", "### Subsection", "## Section B"} + for c in chunks: + present = set(_all_headings(c.text)) + assert expected_headings <= present, ( + f"chunk missing headings {expected_headings - present}: {c.text!r}" + ) + + +def test_owner_section_holds_chunk_content(): + """The chunk's content sits directly under its owner heading — not + under any other section's heading.""" + txt = ( + "# Doc\n" + "\n" + "## A\n" + "alpha alpha alpha alpha here\n" + "\n" + "## B\n" + "bravo bravo bravo bravo here\n" + ) + chunks = _parser(60)._chunk(txt, "/x.md") + a_chunk = next(c for c in chunks if "alpha" in c.text) + b_chunk = next(c for c in chunks if "bravo" in c.text) + # In A's chunk, "alpha" must appear AFTER "## A" and BEFORE "## B" + a_idx = a_chunk.text.find("alpha") + a_a_idx = a_chunk.text.find("## A") + a_b_idx = a_chunk.text.find("## B") + assert a_a_idx < a_idx < a_b_idx + # In B's chunk, "bravo" must appear AFTER "## B" + b_idx = b_chunk.text.find("bravo") + b_b_idx = b_chunk.text.find("## B") + assert b_b_idx < b_idx + # And the OTHER section's body must NOT appear in this chunk. + assert "alpha" not in b_chunk.text + assert "bravo" not in a_chunk.text + + +def test_skeleton_unchanged_across_chunks(): + """Strip out body content — every chunk should yield the same + sequence of heading lines (the doc's skeleton).""" + txt = ( + "# Top\n" + "\n" + "## A\n" + "aaa aaa aaa long\n" + "\n" + "## B\n" + "bbb bbb bbb long\n" + "\n" + "## C\n" + "ccc ccc ccc long\n" + ) + chunks = _parser(60)._chunk(txt, "/x.md") + skeletons = [_all_headings(c.text) for c in chunks] + expected = ["# Top", "## A", "## B", "## C"] + for sk in skeletons: + assert sk == expected + + +def test_subsection_skeleton_preserved(): + """Subsection headings still appear in EVERY chunk's skeleton, not + just the chunk that owns the subsection's content.""" + txt = ( + "# Doc\n" + "\n" + "## A\n" + "para A long content here\n" + "\n" + "## B\n" + "\n" + "### B1\n" + "deep content here long\n" + "\n" + "## C\n" + "para C long content here\n" + ) + chunks = _parser(60)._chunk(txt, "/x.md") + a_chunk = next(c for c in chunks if "para A" in c.text) + # B1 heading must appear in A's chunk too — full skeleton preserved. + assert "### B1" in a_chunk.text + assert "## C" in a_chunk.text + # And in B1's chunk, A's heading must appear before B's. + b1_chunk = next(c for c in chunks if "deep content" in c.text) + assert "## A" in b1_chunk.text + assert "## B" in b1_chunk.text + assert "### B1" in b1_chunk.text + assert "## C" in b1_chunk.text + + +# -------------------------------------------------------------------------- +# Body run packing +# -------------------------------------------------------------------------- + + +def test_body_run_greedy_packs_under_one_owner(): + """Multiple bodies under one section pack into one chunk; each + chunk still carries the full skeleton.""" + txt = ( + "# Top\n" + "\n" + "## Single\n" + "para1 content here\n" + "\n" + "para2 different content\n" + "\n" + "para3 last paragraph\n" + ) + chunks = _parser(80)._chunk(txt, "/x.md") + for c in chunks: + # Skeleton has both top and single. + assert "# Top" in c.text + assert "## Single" in c.text + joined = "\n".join(c.text for c in chunks) + for tag in ("para1", "para2", "para3"): + assert tag in joined + + +# -------------------------------------------------------------------------- +# Leaf splits: full skeleton wraps each piece +# -------------------------------------------------------------------------- + + +def test_table_split_keeps_skeleton(): + txt = ( + "# Doc\n" + "\n" + "## Tables\n" + "\n" + "| name | value |\n" + "|------|-------|\n" + "| r1 | a |\n" + "| r2 | b |\n" + "| r3 | c |\n" + "| r4 | d |\n" + "\n" + "## Other\n" + "other content\n" + ) + chunks = _parser(80)._chunk(txt, "/x.md") + table_chunks = [c for c in chunks if "| name | value |" in c.text] + assert len(table_chunks) >= 2 + for c in table_chunks: + # Header repeats per piece. + assert "| name | value |" in c.text + assert "----" in c.text + # Skeleton complete: # Doc, ## Tables, ## Other all present. + assert "# Doc" in c.text + assert "## Tables" in c.text + assert "## Other" in c.text + + +def test_code_fence_split_keeps_skeleton(): + txt = ( + "# Doc\n" + "\n" + "## Code\n" + "\n" + "```python\n" + "def line1():\n" + " pass\n" + "\n" + "def line2():\n" + " pass\n" + "\n" + "def line3():\n" + " pass\n" + "\n" + "def line4():\n" + " pass\n" + "```\n" + "\n" + "## After\n" + "after content\n" + ) + chunks = _parser(100)._chunk(txt, "/x.md") + code_chunks = [c for c in chunks if "```python" in c.text] + assert len(code_chunks) >= 2 + for c in code_chunks: + # Fence opener + closer repeat per piece. + assert "```python" in c.text + # Skeleton: # Doc, ## Code, ## After. + assert "# Doc" in c.text + assert "## Code" in c.text + assert "## After" in c.text + + +def test_list_split_keeps_skeleton(): + txt = ( + "# Doc\n" + "\n" + "## Items\n" + "\n" + "- item one with some text\n" + "- item two with text\n" + "- item three text\n" + "- item four text\n" + "- item five text\n" + "- item six text\n" + "\n" + "## After\n" + "after content\n" + ) + chunks = _parser(110)._chunk(txt, "/x.md") + list_chunks = [c for c in chunks if "- item" in c.text] + assert len(list_chunks) >= 2 + for c in list_chunks: + assert "# Doc" in c.text + assert "## Items" in c.text + assert "## After" in c.text + joined = "\n".join(c.text for c in list_chunks) + for tag in ("one", "two", "three", "four", "five", "six"): + assert tag in joined + + +def test_paragraph_line_split_keeps_skeleton(): + txt = ( + "# P\n" + "\n" + "## Section\n" + "alpha line one with extra padding text here\n" + "beta line two with extra padding text here\n" + "gamma line three with extra padding text here\n" + "delta line four with extra padding text here\n" + "\n" + "## After\n" + "after content\n" + ) + chunks = _parser(110)._chunk(txt, "/x.md") + para_chunks = [c for c in chunks if any(t in c.text for t in ("alpha", "beta", "gamma", "delta"))] + assert len(para_chunks) >= 2 + for c in para_chunks: + assert "# P" in c.text + assert "## Section" in c.text + assert "## After" in c.text + + +# -------------------------------------------------------------------------- +# embed_toc toggle + content-only budget +# -------------------------------------------------------------------------- + + +def test_embed_toc_off_strips_skeleton(): + """With embed_toc=False, chunks contain only their own content — + no full-doc heading skeleton wrapping them.""" + txt = ( + "# Doc\n" + "\n" + "## A\n" + "para A long content here\n" + "\n" + "## B\n" + "para B long content here\n" + ) + chunks = _parser(60, embed_toc=False)._chunk(txt, "/x.md") + a = next(c for c in chunks if "para A" in c.text) + b = next(c for c in chunks if "para B" in c.text) + # Neither chunk should carry the OTHER section's heading. + assert "## B" not in a.text + assert "## A" not in b.text + # And neither should carry the doc title (no full-doc TOC). + assert "# Doc" not in a.text + assert "# Doc" not in b.text + + +def test_embed_toc_off_content_only(): + """A whole-doc chunk under embed_toc=False is just the rendered + content — its own section headings remain (they're part of the + content), but no extra TOC wrap is added.""" + txt = "# Top\nhello world content here\n" + chunks = _parser(500, embed_toc=False)._chunk(txt, "/x.md") + assert len(chunks) == 1 + text = chunks[0].text.strip() + # The doc's own heading IS the content of the root chunk. + assert text.startswith("# Top") + assert "hello world content here" in text + # But chunk size matches just the rendered doc — no extra prefix + # would have been added that isn't in the source. + assert text == "# Top\n\nhello world content here" + + +def test_embed_toc_default_is_on(): + """Default behavior keeps TOC embedding on.""" + p = _parser(500) + assert p.embed_toc is True + chunks = p._chunk("# Top\nbody content\n", "/x.md") + assert "# Top" in chunks[0].text + + +def test_chunk_chars_constrains_content_only(): + """chunk_chars limits CONTENT size; the TOC skeleton is additive + and may push final chunk text well beyond chunk_chars.""" + # Doc with a deep heading skeleton (~70 chars) and short body. + txt = ( + "# H1 long heading title here\n" + "## H2 long subheading title\n" + "### H3 deep heading title\n" + "x\n" # body of H3 + ) + chunks = _parser(50)._chunk(txt, "/x.md") + assert len(chunks) == 1 + final = chunks[0].text + # Final chunk text > chunk_chars because TOC was added on top. + assert len(final) > 50 + # All headings present (full skeleton). + assert "# H1" in final and "## H2" in final and "### H3" in final + # Body present. + assert "x" in final + + +def test_body_run_budget_excludes_toc(): + """A body run greedy-pack uses chunk_chars purely for body content; + the TOC skeleton overhead doesn't squeeze the budget.""" + # Two bodies whose joined size = 50 chars; with TOC skeleton, + # old behavior (TOC counted) might split, but new should fit. + txt = ( + "# Top\n" + "## Sub long heading title here\n" + "abcdefghij abcdefghij abcdefghij\n" # 32 chars body + "\n" + "klmnopqrst klmnopqrst klmnopqrst\n" # 32 chars body + ) + # 80 chars budget covers the joined body (32+2+32=66) but is + # smaller than body+TOC under old (counting) semantics (~120). + chunks = _parser(80)._chunk(txt, "/x.md") + # Both bodies in one chunk because content-only budget allows it. + assert len(chunks) == 1 + assert "abcdefghij" in chunks[0].text + assert "klmnopqrst" in chunks[0].text + + +def test_finalize_off_returns_content_unchanged(): + """White-box: _finalize with embed_toc=False is the identity for content.""" + from mistletoe.block_token import Document + from mistletoe.markdown_renderer import MarkdownRenderer + + p = _parser(500, embed_toc=False) + txt = "# A\n## B\nbody\n" + with MarkdownRenderer() as r: + tree = p._build_tree(Document(txt), r) + h1 = tree.children[0] + b = h1.children[0] # ## B + out = p._finalize(tree, b, "RAW", owns_subtree=False) + assert out == "RAW" + + +def test_finalize_on_wraps_with_toc(): + """White-box: _finalize with embed_toc=True wraps content with skeleton.""" + from mistletoe.block_token import Document + from mistletoe.markdown_renderer import MarkdownRenderer + + p = _parser(500, embed_toc=True) + txt = "# A\n## B\nbody\n" + with MarkdownRenderer() as r: + tree = p._build_tree(Document(txt), r) + h1 = tree.children[0] + b = h1.children[0] + out = p._finalize(tree, b, "RAW", owns_subtree=False) + assert "# A" in out and "## B" in out and "RAW" in out + + +# -------------------------------------------------------------------------- +# Empty / whitespace-only +# -------------------------------------------------------------------------- + + +def test_empty_text_yields_no_chunks(): + assert _parser(500)._chunk("", "/x.md") == [] + assert _parser(500)._chunk(" \n\n ", "/x.md") == [] + + +# -------------------------------------------------------------------------- +# Leaf-block split parts: [Part X/N] markers +# -------------------------------------------------------------------------- + + +def test_table_split_marks_parts(): + """Table split into N pieces gets [Part X/N] prefix on each.""" + txt = ( + "# Doc\n" + "\n" + "## T\n" + "\n" + "| a | b |\n" + "|---|---|\n" + "| 1 | x |\n" + "| 2 | y |\n" + "| 3 | z |\n" + "| 4 | u |\n" + "| 5 | v |\n" + "| 6 | w |\n" + ) + chunks = _parser(80)._chunk(txt, "/x.md") + table_chunks = [c for c in chunks if "| a | b" in c.text] + assert len(table_chunks) >= 2 + n = len(table_chunks) + for i, c in enumerate(table_chunks, 1): + assert f"[Part {i}/{n}]" in c.text + # The marker sits BEFORE the table header. + for c in table_chunks: + marker_pos = c.text.find("[Part") + header_pos = c.text.find("| a") + assert 0 <= marker_pos < header_pos + + +def test_code_split_marks_parts(): + txt = ( + "# Doc\n" + "\n" + "## C\n" + "\n" + "```python\n" + "def f1(): pass\n" + "def f2(): pass\n" + "def f3(): pass\n" + "def f4(): pass\n" + "def f5(): pass\n" + "def f6(): pass\n" + "```\n" + ) + chunks = _parser(70)._chunk(txt, "/x.md") + code_chunks = [c for c in chunks if "```python" in c.text] + assert len(code_chunks) >= 2 + n = len(code_chunks) + for i, c in enumerate(code_chunks, 1): + assert f"[Part {i}/{n}]" in c.text + # Marker before the fence opener. + assert c.text.find("[Part") < c.text.find("```python") + + +def test_list_split_marks_parts(): + txt = ( + "# Doc\n" + "\n" + "## L\n" + "\n" + "- item one with extra padding text content here\n" + "- item two with extra padding text content here\n" + "- item three with extra padding text content here\n" + "- item four with extra padding text content here\n" + "- item five with extra padding text content here\n" + "- item six with extra padding text content here\n" + ) + chunks = _parser(120)._chunk(txt, "/x.md") + list_chunks = [c for c in chunks if "- item" in c.text] + assert len(list_chunks) >= 2 + n = len(list_chunks) + for i, c in enumerate(list_chunks, 1): + assert f"[Part {i}/{n}]" in c.text + + +def test_paragraph_split_marks_parts(): + txt = ( + "# Doc\n" + "\n" + "## P\n" + "alpha line one with extra padding text here\n" + "beta line two with extra padding text here\n" + "gamma line three with extra padding text here\n" + "delta line four with extra padding text here\n" + ) + chunks = _parser(100)._chunk(txt, "/x.md") + para_chunks = [ + c for c in chunks + if any(t in c.text for t in ("alpha", "beta", "gamma", "delta")) + ] + assert len(para_chunks) >= 2 + n = len(para_chunks) + for i, c in enumerate(para_chunks, 1): + assert f"[Part {i}/{n}]" in c.text + + +def test_single_piece_leaf_has_no_part_marker(): + """When a leaf block fits in one piece, no [Part] prefix is added.""" + txt = ( + "# Doc\n" + "\n" + "## T\n" + "\n" + "| a | b |\n" + "|---|---|\n" + "| 1 | 2 |\n" + ) + chunks = _parser(500)._chunk(txt, "/x.md") + assert len(chunks) == 1 + assert "[Part" not in chunks[0].text + + +def test_part_marker_absent_for_body_run_packing(): + """Body run packing (separate blocks under one section) doesn't + use [Part] markers — that's reserved for splitting ONE leaf block.""" + txt = ( + "# Doc\n" + "\n" + "## S\n" + "para1 first paragraph content\n" + "\n" + "para2 second paragraph content\n" + "\n" + "para3 third paragraph content\n" + ) + # Force greedy-pack across separate paragraphs. + chunks = _parser(60)._chunk(txt, "/x.md") + for c in chunks: + assert "[Part" not in c.text + + +# -------------------------------------------------------------------------- +# Provenance +# -------------------------------------------------------------------------- + + +def test_chunk_line_ranges_track_source(): + txt = ( + "# Top\n" # line 1 + "intro\n" # line 2 + "\n" + "## Sub\n" # line 4 + "body\n" # line 5 + ) + chunks = _parser(500)._chunk(txt, "/x.md") + assert len(chunks) == 1 + assert chunks[0].start_line == 1 + assert chunks[0].end_line == 5 + + +def test_chunk_id_changes_with_content(): + a = _parser(500)._chunk("# X\nhello\n", "/x.md")[0] + b = _parser(500)._chunk("# X\nhello\n", "/x.md")[0] + c = _parser(500)._chunk("# X\nhellp\n", "/x.md")[0] + assert a.id == b.id + assert a.id != c.id + + +# -------------------------------------------------------------------------- +# Render helper directly +# -------------------------------------------------------------------------- + + +def test_render_with_full_toc_inlines_at_owner(): + """White-box: build a tree, call _render_with_full_toc directly and + verify the slot-fill behavior.""" + from mistletoe.block_token import Document + from mistletoe.markdown_renderer import MarkdownRenderer + + p = _parser(2000) + txt = "# A\n\n## B\nbody B\n\n## C\nbody C\n" + with MarkdownRenderer() as r: + tree = p._build_tree(Document(txt), r) + h1 = tree.children[0] + b = h1.children[0] # ## B + rendered = p._render_with_full_toc(tree, b, "INSERTED", owns_subtree=False) + # Skeleton lists # A, ## B, ## C; INSERTED sits under ## B but BEFORE ## C + assert "# A" in rendered + assert "## B" in rendered + assert "## C" in rendered + b_idx = rendered.find("## B") + inserted_idx = rendered.find("INSERTED") + c_idx = rendered.find("## C") + assert b_idx < inserted_idx < c_idx + + +def test_render_with_full_toc_root_owner(): + """Root owner means content sits BEFORE the first heading.""" + from mistletoe.block_token import Document + from mistletoe.markdown_renderer import MarkdownRenderer + + p = _parser(2000) + txt = "# A\n\nbody A\n" + with MarkdownRenderer() as r: + tree = p._build_tree(Document(txt), r) + rendered = p._render_with_full_toc(tree, tree, "PRELUDE", owns_subtree=False) + assert rendered.startswith("PRELUDE") + assert "# A" in rendered + + +# -------------------------------------------------------------------------- +# MdNode dataclass +# -------------------------------------------------------------------------- + + +def test_mdnode_defaults(): + n = MdNode(kind="body") + assert n.heading is None and n.level == 0 + assert n.children == [] and n.block is None + assert n.text == "" and n.start_line == 0 diff --git a/tests/test_wikilink.py b/tests/test_wikilink.py deleted file mode 100644 index d290ac7d..00000000 --- a/tests/test_wikilink.py +++ /dev/null @@ -1,219 +0,0 @@ -"""Wikilink protocol unit tests — body-only edge extraction. - -Covers the three legal inline forms (bare / line-level Dataview / -inline-bracketed Dataview), multi-target expansion, dedup against -typed wrappers, open-vocabulary predicate pass-through, and the -explicit decision that frontmatter is no longer walked for links. -""" - -from reme2.schema.file_edge import FileEdge, extract_wikilinks, parse_wikilinks - - -# -------------------------------------------------------------------------- -# Bare wikilinks -# -------------------------------------------------------------------------- - - -def test_bare_wikilink(): - edges = parse_wikilinks("see [[X]]") - assert len(edges) == 1 - assert edges[0].target == "X" - assert edges[0].predicate is None - assert edges[0].embed is False - - -def test_bare_with_anchor_alias_embed(): - edges = parse_wikilinks("![[X#sec|Alias]]") - assert len(edges) == 1 - assert edges[0].target == "X" - assert edges[0].anchor == "sec" - assert edges[0].alias == "Alias" - assert edges[0].embed is True - - -# -------------------------------------------------------------------------- -# Line-level Dataview -# -------------------------------------------------------------------------- - - -def test_line_level_field(): - edges = parse_wikilinks("extends:: [[Source Topic]]") - assert len(edges) == 1 - assert edges[0].target == "Source Topic" - assert edges[0].predicate == "extends" - - -def test_line_level_multi_target(): - edges = parse_wikilinks("concerns:: [[A]], [[B]], [[C]]") - assert [(e.target, e.predicate) for e in edges] == [ - ("A", "concerns"), - ("B", "concerns"), - ("C", "concerns"), - ] - - -def test_line_level_with_bullet(): - edges = parse_wikilinks("- extends:: [[X]]\n * concerns:: [[Y]]") - assert [(e.target, e.predicate) for e in edges] == [ - ("X", "extends"), - ("Y", "concerns"), - ] - - -# -------------------------------------------------------------------------- -# Inline-bracketed Dataview -# -------------------------------------------------------------------------- - - -def test_inline_bracketed(): - edges = parse_wikilinks("This [extends:: [[Y]]] something else.") - assert len(edges) == 1 - assert edges[0].target == "Y" - assert edges[0].predicate == "extends" - - -def test_inline_bracketed_multi_target(): - edges = parse_wikilinks("[concerns:: [[A]], [[B]]]") - assert [(e.target, e.predicate) for e in edges] == [ - ("A", "concerns"), - ("B", "concerns"), - ] - - -def test_inline_bracketed_skips_cross_line(): - # A `[predicate:: ...]` that spans a newline is malformed → the inner - # wikilink falls back to bare; the unmatched `[` does not eat tail text. - edges = parse_wikilinks("[extends:: [[X]]\nbad]") - assert len(edges) == 1 - assert edges[0].target == "X" - assert edges[0].predicate is None - - -# -------------------------------------------------------------------------- -# Dedup: a wikilink inside a typed wrapper should not double-emit -# -------------------------------------------------------------------------- - - -def test_inline_bracketed_does_not_double_emit(): - edges = parse_wikilinks("see [extends:: [[X]]] again.") - assert len(edges) == 1 - assert edges[0].predicate == "extends" - - -def test_line_level_value_does_not_double_emit(): - edges = parse_wikilinks("extends:: [[X]]") - assert len(edges) == 1 - - -def test_typed_and_bare_coexist_for_same_target(): - edges = parse_wikilinks("extends:: [[X]]\nFree text mentioning [[X]] again.") - targets_preds = sorted(((e.target, e.predicate or "") for e in edges)) - assert targets_preds == [("X", ""), ("X", "extends")] - - -# -------------------------------------------------------------------------- -# Open-vocabulary predicates — any identifier-shaped token passes through -# -------------------------------------------------------------------------- - - -def test_arbitrary_predicate_preserved(): - edges = parse_wikilinks("anything_goes:: [[X]]") - assert len(edges) == 1 - assert edges[0].target == "X" - assert edges[0].predicate == "anything_goes" - - -def test_inline_arbitrary_predicate_preserved(): - edges = parse_wikilinks("[wat:: [[X]]]") - assert len(edges) == 1 - assert edges[0].predicate == "wat" - - -def test_predicate_must_be_identifier_shaped(): - # A leading digit fails the regex `[A-Za-z][A-Za-z0-9_]*` so the line is - # not recognised as a Dataview field — the wikilink falls back to bare. - edges = parse_wikilinks("123bad:: [[X]]") - assert len(edges) == 1 - assert edges[0].target == "X" - assert edges[0].predicate is None - - -def test_file_edge_accepts_any_predicate_string(): - e = FileEdge(target="X", predicate="totally_made_up") - assert e.predicate == "totally_made_up" - - -# -------------------------------------------------------------------------- -# Frontmatter is NOT walked for links — explicit regression -# -------------------------------------------------------------------------- - - -def test_frontmatter_links_block_is_ignored(): - # Even if a YAML-shaped string sits at the top of body, parse_wikilinks - # only operates on body. We pass body text directly here, so this test - # asserts the API surface no longer accepts a metadata dict. - import inspect - - sig = inspect.signature(parse_wikilinks) - assert list(sig.parameters.keys()) == ["text"], ( - "parse_wikilinks should accept body text only — frontmatter walk removed" - ) - - -def test_no_frontmatter_walker_exported(): - from reme2.schema import file_edge as wl - - for removed in ( - "parse_wikilinks_from_metadata", - "extract_wikilinks_from_metadata", - "extract_inline_fields", - "extract_typed_edges", - "InlineField", - ): - assert not hasattr(wl, removed), ( - f"{removed} should have been removed when YAML edges were dropped" - ) - - -# -------------------------------------------------------------------------- -# FileEdge schema -# -------------------------------------------------------------------------- - - -def test_file_edge_extra_forbid(): - import pytest - from pydantic import ValidationError - - with pytest.raises(ValidationError): - FileEdge(target="X", source="regex") # type: ignore[call-arg] - - -def test_file_edge_minimal_field_set(): - e = FileEdge(target="X") - dumped = e.model_dump() - assert set(dumped.keys()) == {"target", "predicate", "anchor", "alias", "embed"} - - -# -------------------------------------------------------------------------- -# Back-compat: extract_wikilinks returns flat target list (used by ingestor) -# -------------------------------------------------------------------------- - - -def test_extract_wikilinks_flat_targets(): - targets = extract_wikilinks("see [[X]] and extends:: [[Y]] and [extends:: [[Z]]]") - assert targets == ["X", "Y", "Z"] - - -# -------------------------------------------------------------------------- -# Source ordering stability -# -------------------------------------------------------------------------- - - -def test_edges_sorted_by_source_position(): - body = ( - "intro [[First]] then\n" - "extends:: [[Second]]\n" - "tail [[Third]]\n" - ) - edges = parse_wikilinks(body) - assert [e.target for e in edges] == ["First", "Second", "Third"]