scriptoriumadmin/ReMe
1
0
Fork 0
mirror of https://github.com/agentscope-ai/ReMe.git synced 2026-09-22 00:32:49 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 6

feat(parser): implement AST-based markdown chunking with full document TOC

Browse source 
- 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.
This commit is contained in:
huangsen 2026-05-13 14:32:00 +08:00
parent 90da850200
commit 51ec09d98f
8 changed files with 1979 additions and 486 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

883
reme2/component/file_parser/linked_file_parser.py
View file

@ -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
<chunk content>
## 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 <path> [--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

2
reme2/memory/maintainer.py
View file

@ -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.

6
reme2/memory/memory_io.py
View file

@ -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(

3
reme2/schema/__init__.py
View file

@ -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",
]

278
reme2/schema/file_edge.py
View file

@ -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<target>[^\]\|\#\n]+?)
(?:\#(?P<anchor>[^\]\|\n]+))?
(?:\|(?P<alias>[^\]\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<predicate>[A-Za-z][A-Za-z0-9_]*)\s*::\s*(?P<value>.+?)\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<predicate>[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)]

348
tests/test_file_edge.py Normal file
View file

@ -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"]

726
tests/test_md_chunker.py Normal file
View file

@ -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

219
tests/test_wikilink.py
View file

@ -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"]