From 507199da903ffa1c4f9e5aadfb380889339e71b8 Mon Sep 17 00:00:00 2001 From: BeiMu-new <2690322907@qq.com> Date: Thu, 5 Nov 2026 17:36:26 +0800 Subject: [PATCH] fix(tag-index): keep legacy Windows spellings consistent across index consumers Persisted FileNode.path and FileChunk.path written by older Windows builds carry native separators, while the tag index, the search filter and freshly indexed files use workspace-relative POSIX. Only the rebuild path normalized the spelling, so a tag listing could report a file that node retrieval and tagged search then failed to match. - add the shared canonical identity in reme.utils.path_utils - canonicalize on load for both graph backends and for chunks - merge rows that describe the same file into the record that wins the merge, and hand the losing record's chunk handles to the store, which drops the chunks and keyword entries instead of leaving superseded text searchable under the surviving path and tags - deduplicate a write batch that names one file under both spellings, so the staged chunks and the published chunk handles stay in step - rebuild migrated edges from the winning record's links instead of copying the persisted edges of both spellings - canonicalize write, delete and link lookups as well as query arguments - canonicalize search filter paths/prefixes and date extraction - keep the containment validation unchanged; only index files are rewritten --- reme/components/file_graph/base_file_graph.py | 51 +++ .../components/file_graph/local_file_graph.py | 17 +- reme/components/file_graph/nx_file_graph.py | 47 +- .../components/file_store/local_file_store.py | 97 ++++- reme/components/tag_index/local_tag_index.py | 25 +- reme/steps/index/search.py | 4 +- reme/utils/__init__.py | 2 + reme/utils/path_utils.py | 17 + tests/unit/test_tag_index.py | 402 +++++++++++++++++- 9 files changed, 636 insertions(+), 26 deletions(-) create mode 100644 reme/utils/path_utils.py diff --git a/reme/components/file_graph/base_file_graph.py b/reme/components/file_graph/base_file_graph.py index 98a14e4d..c8eed489 100644 --- a/reme/components/file_graph/base_file_graph.py +++ b/reme/components/file_graph/base_file_graph.py @@ -5,6 +5,7 @@ from abc import abstractmethod from ..base_component import BaseComponent from ...enumeration import ComponentEnum, LinkScopeEnum from ...schema import FileLink, FileNode +from ...utils import canonical_index_path class BaseFileGraph(BaseComponent): @@ -19,6 +20,56 @@ class BaseFileGraph(BaseComponent): component_type = ComponentEnum.FILE_GRAPH + def __init__(self, **kwargs) -> None: + super().__init__(**kwargs) + self._dropped_chunk_ids: set[str] = set() + + # -- Path identity ----------------------------------------------------- + + @staticmethod + def canonical_path(path: str) -> str: + """Return the workspace-relative POSIX identity of ``path``.""" + return canonical_index_path(path) + + @classmethod + def canonical_node(cls, node: FileNode) -> FileNode: + """Return ``node`` with its own path and link paths canonicalized.""" + links = [ + link.model_copy( + update={ + "source_path": canonical_index_path(link.source_path), + "target_path": canonical_index_path(link.target_path), + }, + ) + for link in node.links + ] + path = canonical_index_path(node.path) + if path == node.path and links == node.links: + return node + return node.model_copy(update={"path": path, "links": links}) + + def merge_chunk_ids(self, keep: FileNode, extra: FileNode) -> FileNode: + """Return the winning record, discarding the other spelling's chunk handles. + + ``keep`` describes the version that owns the merged node: its path, + links, tags and chunk list all describe the content reachable through + that path today. Handles from ``extra`` describe the version that lost + the merge, so merging them into the node made superseded text searchable + under the surviving tags. They are reported through + :meth:`take_dropped_chunk_ids` instead, and the owning store removes the + matching chunks and keyword entries. + """ + dropped = [chunk_id for chunk_id in extra.chunk_ids if chunk_id not in keep.chunk_ids] + if dropped: + self._dropped_chunk_ids.update(dropped) + return keep + + def take_dropped_chunk_ids(self) -> list[str]: + """Return and clear chunk handles discarded while merging spellings.""" + dropped = sorted(self._dropped_chunk_ids) + self._dropped_chunk_ids.clear() + return dropped + # -- Lifecycle --------------------------------------------------------- async def _start(self) -> None: diff --git a/reme/components/file_graph/local_file_graph.py b/reme/components/file_graph/local_file_graph.py index f47c75e4..26ea28fb 100644 --- a/reme/components/file_graph/local_file_graph.py +++ b/reme/components/file_graph/local_file_graph.py @@ -33,8 +33,13 @@ class LocalFileGraph(BaseFileGraph): try: for line in read_jsonl_zst(self._graph_file): if line.strip(): - node = FileNode.model_validate_json(line) - self._nodes[node.path] = node + node = self.canonical_node(FileNode.model_validate_json(line)) + existing = self._nodes.get(node.path) + if existing is None: + self._nodes[node.path] = node + continue + keep, extra = (node, existing) if node.st_mtime >= existing.st_mtime else (existing, node) + self._nodes[node.path] = self.merge_chunk_ids(keep, extra) self.logger.debug(f"Loaded {len(self._nodes)} nodes from {self._graph_file}") except Exception as e: self.logger.exception(f"Failed to load {self._graph_file}: {e}") @@ -79,6 +84,7 @@ class LocalFileGraph(BaseFileGraph): async def upsert_nodes(self, nodes: list[FileNode]) -> None: for node in nodes: + node = self.canonical_node(node) path = node.path old = self._nodes.get(path) if old is not None: @@ -93,6 +99,7 @@ class LocalFileGraph(BaseFileGraph): async def delete_nodes(self, paths: list[str]) -> None: for path in paths: + path = self.canonical_path(path) node = self._nodes.pop(path, None) if node is None: continue @@ -105,7 +112,8 @@ class LocalFileGraph(BaseFileGraph): async def get_nodes(self, paths: list[str] | None = None) -> list[FileNode]: if paths is None: return list(self._nodes.values()) - return [self._nodes[p] for p in paths if p in self._nodes] + canonical = [self.canonical_path(p) for p in paths] + return [self._nodes[p] for p in canonical if p in self._nodes] async def rebuild_links(self) -> None: self._inverse.clear() @@ -124,13 +132,14 @@ class LocalFileGraph(BaseFileGraph): async def get_outlinks(self, path: str, scope: LinkScopeEnum | str = LinkScopeEnum.REAL) -> list[FileLink]: scope = self._normalize_scope(scope) - node = self._nodes.get(path) + node = self._nodes.get(self.canonical_path(path)) if node is None: return [] return [lnk for lnk in node.links if lnk.target_path and self._scope_match(lnk.target_path, scope)] async def get_inlinks(self, path: str, scope: LinkScopeEnum | str = LinkScopeEnum.REAL) -> list[FileLink]: scope = self._normalize_scope(scope) + path = self.canonical_path(path) sources: set[str] = set() if scope in (LinkScopeEnum.REAL, LinkScopeEnum.ALL): sources |= self._inverse.get(path, set()) diff --git a/reme/components/file_graph/nx_file_graph.py b/reme/components/file_graph/nx_file_graph.py index 27257925..baea72e6 100644 --- a/reme/components/file_graph/nx_file_graph.py +++ b/reme/components/file_graph/nx_file_graph.py @@ -34,6 +34,9 @@ class NxFileGraph(BaseFileGraph): try: with open(self._graph_file, "rb") as f: self._graph = pickle.load(f) + migrated = self._canonicalize_graph() + if migrated: + self.logger.info(f"Canonicalized {migrated} legacy path spelling(s) in {self._graph_file}") self.logger.info(f"Loaded {self._real_count()} nodes from {self._graph_file}") except Exception as e: self.logger.exception(f"Failed to load {self._graph_file}: {e}") @@ -53,6 +56,43 @@ class NxFileGraph(BaseFileGraph): def _real_count(self) -> int: return sum(1 for _, d in self._graph.nodes(data=True) if "node" in d) + def _has_legacy_keys(self) -> bool: + """Report whether any persisted node key is not canonical. + + Every edge endpoint is itself a node key, so canonical nodes imply + canonical edges. Iterating ``MultiDiGraph.edges`` here yields + ``(source, target, key)`` and raised instead of answering, which made + every load of a graph with edges fail before migration could run. + """ + return any(self.canonical_path(key) != key for key in self._graph.nodes) + + def _canonicalize_graph(self) -> int: + """Rewrite persisted node keys and rebuild edges from winning records.""" + if not self._has_legacy_keys(): + return 0 + graph = self._graph.__class__() + migrated = 0 + for key, data in self._graph.nodes(data=True): + node = data.get("node") + if node is None: + # Placeholders are recreated from the winning records below. + continue + path = self.canonical_path(key) + migrated += path != key + node = self.canonical_node(node) + existing = graph.nodes[path].get("node") if graph.has_node(path) else None + if existing is not None: + keep, extra = (node, existing) if node.st_mtime >= existing.st_mtime else (existing, node) + graph.add_node(path, node=self.merge_chunk_ids(keep, extra)) + continue + graph.add_node(path, node=node) + # Copying the persisted edges verbatim would keep the dropped spelling's + # links as outlinks of the merged node, in addition to the winner's. + for path, data in list(graph.nodes(data=True)): + graph.add_edges_from(self._edges_from(path, data["node"])) + self._graph = graph + return migrated + def _is_real(self, key: str) -> bool: return "node" in self._graph.nodes[key] @@ -70,6 +110,7 @@ class NxFileGraph(BaseFileGraph): async def upsert_nodes(self, nodes: list[FileNode]) -> None: for node in nodes: + node = self.canonical_node(node) path = node.path if self._graph.has_node(path): self._graph.remove_edges_from(list(self._graph.out_edges(path, keys=True))) @@ -78,6 +119,7 @@ class NxFileGraph(BaseFileGraph): async def delete_nodes(self, paths: list[str]) -> None: for path in paths: + path = self.canonical_path(path) if not self._graph.has_node(path): continue self._graph.remove_edges_from(list(self._graph.out_edges(path, keys=True))) @@ -89,7 +131,8 @@ class NxFileGraph(BaseFileGraph): view = self._graph.nodes if paths is None: return [d["node"] for _, d in view(data=True) if "node" in d] - return [view[p]["node"] for p in paths if p in view and "node" in view[p]] + canonical = [self.canonical_path(p) for p in paths] + return [view[p]["node"] for p in canonical if p in view and "node" in view[p]] async def rebuild_links(self) -> None: self._graph.remove_edges_from(list(self._graph.edges(keys=True))) @@ -105,6 +148,7 @@ class NxFileGraph(BaseFileGraph): # -- Link access ------------------------------------------------------- async def get_outlinks(self, path: str, scope: LinkScopeEnum = LinkScopeEnum.REAL) -> list[FileLink]: + path = self.canonical_path(path) view = self._graph.nodes if path not in view or "node" not in view[path]: return [] @@ -115,6 +159,7 @@ class NxFileGraph(BaseFileGraph): ] async def get_inlinks(self, path: str, scope: LinkScopeEnum = LinkScopeEnum.REAL) -> list[FileLink]: + path = self.canonical_path(path) view = self._graph.nodes if path not in view or not self._scope_match(path, scope): return [] diff --git a/reme/components/file_store/local_file_store.py b/reme/components/file_store/local_file_store.py index 7cee6d85..af0dc217 100644 --- a/reme/components/file_store/local_file_store.py +++ b/reme/components/file_store/local_file_store.py @@ -20,7 +20,7 @@ from ..keyword_index import BaseKeywordIndex from ..tag_index import BaseTagIndex from ...enumeration import LinkScopeEnum from ...schema import FileChunk, FileLink, FileNode -from ...utils import batch_cosine_similarity +from ...utils import batch_cosine_similarity, canonical_index_path from ...utils.async_utils import complete_in_thread from ...utils.jsonl_zst import read_jsonl_zst, write_jsonl_zst @@ -301,7 +301,7 @@ class LocalFileStore(BaseFileStore): for line in read_jsonl_zst(self.chunks_path, self.encoding): line = line.strip() if line: - chunk = self._deserialize_chunk(line) + chunk = self._canonical_chunk(self._deserialize_chunk(line)) self.file_chunks[chunk.id] = chunk self.logger.info(f"Loaded {len(self.file_chunks)} chunks from {self.chunks_path}") except Exception as e: @@ -312,6 +312,7 @@ class LocalFileStore(BaseFileStore): ) graph_repair_started_at = time.monotonic() + await self._drop_superseded_chunks() graph_repaired = await self._repair_graph_chunk_consistency() self.logger.info( f"{self.name}: graph consistency check complete: repaired={graph_repaired}, " @@ -338,6 +339,28 @@ class LocalFileStore(BaseFileStore): f"{self.name}: file store load phases complete: " f"elapsed={time.monotonic() - started_at:.3f}s", ) + async def _drop_superseded_chunks(self) -> list[str]: + """Remove chunk handles that lost a duplicate path-spelling merge. + + The graph reports the handles that belonged to the version which lost + the merge. Keeping them made the superseded text searchable under the + surviving path and tags, and the store then saw them as orphans, which + made :meth:`_repair_graph_chunk_consistency` clear the whole derived + index on the next start. + """ + assert self.file_graph is not None + dropped = self.file_graph.take_dropped_chunk_ids() + if not dropped: + return [] + removed = [chunk_id for chunk_id in dropped if self.file_chunks.pop(chunk_id, None) is not None] + if removed and self.keyword_index is not None: + with suppress(Exception): + await self.keyword_index.delete_docs(removed) + self.logger.warning( + f"{self.name}: dropped {len(removed)} superseded chunk(s) from a conflicting path spelling", + ) + return removed + async def _repair_graph_chunk_consistency(self) -> bool: """Clear torn graph/chunk state so the filesystem scan rebuilds it. @@ -366,6 +389,16 @@ class LocalFileStore(BaseFileStore): await self.clear() return True + @staticmethod + def _canonical_chunk(chunk: FileChunk) -> FileChunk: + """Return ``chunk`` carrying the canonical workspace-relative POSIX path. + + The chunk id is intentionally left untouched: it is an opaque handle that + ``FileNode.chunk_ids`` and every derived index already reference. + """ + path = canonical_index_path(chunk.path) + return chunk if path == chunk.path else chunk.model_copy(update={"path": path}) + @staticmethod def _deserialize_chunk(line: str) -> FileChunk: """Read compact vectors while retaining legacy JSON-list compatibility.""" @@ -836,6 +869,17 @@ class LocalFileStore(BaseFileStore): return assert self.file_graph is not None + # Persisted spellings and freshly chunked files must share one identity, + # otherwise the same file is stored and queried as two different paths. + files = [ + ( + self.file_graph.canonical_node(node), + [self._canonical_chunk(chunk) for chunk in chunks], + ) + for node, chunks in files + ] + files = self._dedupe_upsert_batch(files) + old_map = {n.path: n for n in await self.file_graph.get_nodes([node.path for node, _ in files])} old_chunk_ids = {cid for n in old_map.values() for cid in n.chunk_ids} new_nodes, needs_embed, keyword_docs = self._stage_upsert(files, old_map) @@ -849,6 +893,27 @@ class LocalFileStore(BaseFileStore): await self.keyword_index.add_docs(keyword_docs) self._mutation_generation += 1 + def _dedupe_upsert_batch( + self, + files: list[tuple[FileNode, list[FileChunk]]], + ) -> list[tuple[FileNode, list[FileChunk]]]: + """Keep only the last entry per canonical path within one write batch. + + Two spellings of the same file in a single batch would stage the first + entry's chunks and then publish only the second entry's handles, so the + earlier chunks stayed reachable from neither the graph nor the indexes. + A later start then treated them as orphans and cleared the index. + """ + latest: dict[str, tuple[FileNode, list[FileChunk]]] = {} + for node, chunks in files: + if node.path in latest: + self.logger.warning( + f"{self.name}: collapsing duplicate upsert of {node.path} in one batch; keeping the last entry", + ) + del latest[node.path] + latest[node.path] = (node, chunks) + return list(latest.values()) + def _stage_upsert( self, files: list[tuple[FileNode, list[FileChunk]]], @@ -1109,9 +1174,11 @@ class LocalFileStore(BaseFileStore): for key in ("path", "paths"): if key in search_filter: has_exact_paths = True - exact_paths.update(self._as_filter_values(search_filter[key])) + exact_paths.update( + canonical_index_path(str(value)) for value in self._as_filter_values(search_filter[key]) + ) prefixes = tuple( - str(value) + canonical_index_path(str(value)) for key in ("path_prefix", "path_prefixes", "prefix", "prefixes") if key in search_filter for value in self._as_filter_values(search_filter[key]) @@ -1132,12 +1199,13 @@ class LocalFileStore(BaseFileStore): return legacy def matches(chunk: FileChunk) -> bool: - if has_exact_paths and chunk.path not in exact_paths: + path = canonical_index_path(chunk.path) + if has_exact_paths and path not in exact_paths: return False - if prefixes and not chunk.path.startswith(prefixes): + if prefixes and not path.startswith(prefixes): return False if start_date or end_date: - path_date = self._extract_date_from_path(chunk.path) + path_date = self._extract_date_from_path(path) if not path_date: if strict_date: return False @@ -1173,7 +1241,7 @@ class LocalFileStore(BaseFileStore): Returns a validated YYYY-MM-DD string, or None if no date is found. """ - parts = path.split("/") + parts = canonical_index_path(path).split("/") if len(parts) < 2: return None # Accept only exact "YYYY-MM-DD" (dir) or "YYYY-MM-DD.md" (day index). @@ -1192,20 +1260,23 @@ class LocalFileStore(BaseFileStore): if not search_filter: return True + path = canonical_index_path(chunk.path) exact_paths = set() has_exact_path_filter = False for key in ("path", "paths"): if key in search_filter: has_exact_path_filter = True - exact_paths.update(cls._as_filter_values(search_filter[key])) - if has_exact_path_filter and chunk.path not in exact_paths: + exact_paths.update( + canonical_index_path(str(value)) for value in cls._as_filter_values(search_filter[key]) + ) + if has_exact_path_filter and path not in exact_paths: return False prefixes = [] for key in ("path_prefix", "path_prefixes", "prefix", "prefixes"): if key in search_filter: - prefixes.extend(str(v) for v in cls._as_filter_values(search_filter[key])) - if prefixes and not any(chunk.path.startswith(prefix) for prefix in prefixes): + prefixes.extend(canonical_index_path(str(v)) for v in cls._as_filter_values(search_filter[key])) + if prefixes and not any(path.startswith(prefix) for prefix in prefixes): return False # Date range filtering based on date embedded in chunk path. @@ -1215,7 +1286,7 @@ class LocalFileStore(BaseFileStore): end_date = search_filter.get("end_date") strict_date = bool(search_filter.get("strict_date_filter", False)) if start_date or end_date: - path_date = cls._extract_date_from_path(chunk.path) + path_date = cls._extract_date_from_path(path) if not path_date: if strict_date: return False diff --git a/reme/components/tag_index/local_tag_index.py b/reme/components/tag_index/local_tag_index.py index 7392d668..62dfcf6d 100644 --- a/reme/components/tag_index/local_tag_index.py +++ b/reme/components/tag_index/local_tag_index.py @@ -6,6 +6,7 @@ from pathlib import PurePosixPath from .base_tag_index import BaseTagIndex, TagListItem, TagListResult, TagOrder, TagOrderBy from ..component_registry import R from ...schema import FileFrontMatter, FileNode +from ...utils import canonical_index_path @R.register("local") @@ -61,14 +62,30 @@ class LocalTagIndex(BaseTagIndex): """Normalize lookup tags without truncating the query expression.""" return self._normalize_tags(value, limit=None) + @staticmethod + def _is_windows_absolute(path: str) -> bool: + """Return True for a drive-letter absolute path such as ``C:/a.md``.""" + return len(path) >= 3 and path[0].isalpha() and path[1:3] == ":/" + @staticmethod def _validate_path(path: str) -> str: - if not isinstance(path, str) or not path or "\\" in path: + if not isinstance(path, str) or not path: raise ValueError(f"Invalid workspace-relative tag-index path: {path!r}") - pure = PurePosixPath(path) - if pure.is_absolute() or path != pure.as_posix() or any(part in ("", ".", "..") for part in pure.parts): + # Nodes persisted by older Windows builds may still carry native + # separators. The index contract is workspace-relative POSIX, so + # canonicalize that legacy spelling instead of letting a single such + # node abort an otherwise valid rebuild. + canonical = canonical_index_path(path) + pure = PurePosixPath(canonical) + invalid_parts = any(part in ("", ".", "..") for part in pure.parts) + if ( + pure.is_absolute() + or canonical != pure.as_posix() + or invalid_parts + or LocalTagIndex._is_windows_absolute(canonical) + ): raise ValueError(f"Invalid workspace-relative tag-index path: {path!r}") - return path + return canonical def _prepare_nodes(self, nodes: list[FileNode]) -> list[tuple[str, tuple[str, ...]]]: prepared: list[tuple[str, tuple[str, ...]]] = [] diff --git a/reme/steps/index/search.py b/reme/steps/index/search.py index f4e20b3c..3df2d4d9 100644 --- a/reme/steps/index/search.py +++ b/reme/steps/index/search.py @@ -11,7 +11,7 @@ from ..base_step import BaseStep from ..file_io import extract_daily_date from ...components import R from ...schema import FileChunk -from ...utils import expand_links, render_expansion_lines +from ...utils import expand_links, render_expansion_lines, canonical_index_path _RRF_K: Final = 60 _MAX_CANDIDATES: Final = 200 @@ -182,7 +182,7 @@ class SearchStep(_ToolContextMixin, BaseStep): for key in ("path", "paths"): if key in search_filter: has_exact_path_filter = True - exact_paths.update(_filter_values(search_filter.pop(key))) + exact_paths.update(canonical_index_path(str(value)) for value in _filter_values(search_filter.pop(key))) if has_exact_path_filter: allowed_paths.intersection_update(exact_paths) search_filter["paths"] = sorted(allowed_paths) diff --git a/reme/utils/__init__.py b/reme/utils/__init__.py index 5cdbfafc..26ceecd2 100644 --- a/reme/utils/__init__.py +++ b/reme/utils/__init__.py @@ -21,6 +21,7 @@ from .service_utils import ( running_service_config, ) from .similarity_utils import cosine_similarity, batch_cosine_similarity +from .path_utils import canonical_index_path from .token_utils import estimate_token_count from .qwen_token_tools import QwenTokenTools from .web_static import REME_WEB_STATIC_DIR, resolve_web_static_dir @@ -55,6 +56,7 @@ __all__ = [ "running_service_config", "cosine_similarity", "batch_cosine_similarity", + "canonical_index_path", "estimate_token_count", "QwenTokenTools", "REME_WEB_STATIC_DIR", diff --git a/reme/utils/path_utils.py b/reme/utils/path_utils.py new file mode 100644 index 00000000..cc7dc71e --- /dev/null +++ b/reme/utils/path_utils.py @@ -0,0 +1,17 @@ +"""Workspace-relative path identity shared by every index consumer.""" + + +def canonical_index_path(path: str) -> str: + """Return the workspace-relative POSIX spelling used as index identity. + + Indexes persisted by older Windows builds store native separators in + ``FileNode.path`` and ``FileChunk.path``, while the tag index, the search + filter and freshly indexed files use POSIX separators. Every consumer + compares these identities, so the same canonical form is applied when + loading, when writing and to query arguments; otherwise the two spellings + address different files. + + Only separators are rewritten. Rejecting absolute or escaping paths stays + with the caller that owns the containment rule. + """ + return path.replace("\\", "/") diff --git a/tests/unit/test_tag_index.py b/tests/unit/test_tag_index.py index 19f1c9f0..dbc7f19b 100644 --- a/tests/unit/test_tag_index.py +++ b/tests/unit/test_tag_index.py @@ -3,6 +3,8 @@ # pylint: disable=protected-access import asyncio +import json +import pickle from pathlib import Path import pytest @@ -12,9 +14,10 @@ from reme.components.file_chunker import MarkdownFileChunker from reme.components.file_store import LocalFileStore from reme.components.tag_index import LocalTagIndex from reme.config import resolve_app_config -from reme.enumeration import ComponentEnum -from reme.schema import FileChunk, FileFrontMatter, FileNode +from reme.enumeration import ComponentEnum, LinkScopeEnum +from reme.schema import FileChunk, FileFrontMatter, FileLink, FileNode from reme.steps.index.list_tags import ListTagsStep +from reme.utils.jsonl_zst import read_jsonl_zst, write_jsonl_zst def _node(path: str, tags: object = None, *, key: str = "memory_tags") -> FileNode: @@ -84,6 +87,57 @@ async def test_rebuild_is_atomic_and_supports_all_or_any_queries() -> None: assert index.tag_to_paths == before_tags +@pytest.mark.asyncio +async def test_rebuild_canonicalizes_legacy_windows_separators() -> None: + """Accept native separators persisted by older builds instead of failing the rebuild.""" + + index = LocalTagIndex() + await index.rebuild( + [ + _node("daily/a.md", ["python"]), + _node("daily\\b.md", ["python", "legacy"]), + ], + ) + + assert index.is_healthy + assert index.n_files == 2 + assert await index.paths_for_tags(["legacy"]) == ["daily/b.md"] + assert await index.tags_for_path("daily\\b.md") == ["python", "legacy"] + assert await index.tags_for_path("daily/b.md") == ["python", "legacy"] + + await index.delete(["daily\\b.md"]) + assert await index.tags_for_path("daily/b.md") == [] + await index.close() + + +@pytest.mark.asyncio +async def test_mixed_separator_spellings_address_one_entry() -> None: + """Treat a Windows spelling and its POSIX form as the same indexed file.""" + + index = LocalTagIndex() + await index.rebuild( + [ + _node("daily\\a.md", ["old"]), + _node("daily/a.md", ["new"]), + ], + ) + + assert index.n_files == 1 + assert await index.tags_for_path("daily\\a.md") == ["new"] + await index.close() + + +@pytest.mark.asyncio +async def test_windows_absolute_and_escaping_paths_stay_rejected() -> None: + """Canonicalization must not admit absolute Windows paths or parent traversal.""" + + index = LocalTagIndex() + for invalid in ("C:\\Users\\someone\\a.md", "..\\escape.md"): + with pytest.raises(ValueError, match="Invalid workspace-relative"): + await index.rebuild([_node(invalid, ["bad"])]) + await index.close() + + @pytest.mark.asyncio async def test_queries_are_not_truncated_by_per_file_tag_limit() -> None: """Apply the count limit to indexed files without dropping lookup conditions.""" @@ -576,3 +630,347 @@ def test_default_config_enables_tag_index_with_explicit_key() -> None: {"backend": "auto_memory_cc_step"}, {"backend": "auto_tag_step"}, ] + + +def _rewrite_persisted_separators(root: Path, separator: str = "\\") -> int: + """Rewrite persisted index paths to emulate an older Windows build. + + Only local index files are touched: the user-owned Markdown sources must be + left byte-identical. + """ + rewritten = 0 + for path in sorted(root.rglob("*.jsonl.zst")): + lines = [line for line in read_jsonl_zst(path) if line.strip()] + if not lines: + continue + new_lines = [] + for line in lines: + payload = json.loads(line) + before = json.dumps(payload, sort_keys=True) + payload["path"] = payload["path"].replace("/", separator) + for link in payload.get("links") or []: + for key in ("source_path", "target_path"): + value = link.get(key) + if isinstance(value, str): + link[key] = value.replace("/", separator) + rewritten += before != json.dumps(payload, sort_keys=True) + new_lines.append(json.dumps(payload)) + write_jsonl_zst(path, new_lines) + return rewritten + + +@pytest.mark.asyncio +async def test_persisted_windows_spellings_feed_tag_listing_retrieval_and_search(monkeypatch, tmp_path: Path) -> None: + """Keep tag listing, node retrieval and tagged search on one identity.""" + + monkeypatch.chdir(tmp_path) + note = tmp_path / "daily" / "a.md" + note.parent.mkdir(parents=True) + note.write_text("# A\n\nalpha memory\n", encoding="utf-8") + original_note = note.read_bytes() + + store = _standalone_file_store_with_tag_index(name="test", embedding_store="") + await store.start() + await store.upsert( + [(_node("daily/a.md", ["ReMe"]), [_chunk("chunk-a", "daily/a.md", "alpha memory")])], + ) + await store.close() + + assert _rewrite_persisted_separators(tmp_path) == 2 + + store = _standalone_file_store_with_tag_index(name="test", embedding_store="") + await store.start() + try: + assert [node.path for node in await store.get_nodes(["daily/a.md"])] == ["daily/a.md"] + assert [node.path for node in await store.get_nodes(["daily\\a.md"])] == ["daily/a.md"] + assert [node.path for node in await store.get_nodes()] == ["daily/a.md"] + assert await store.tag_index.paths_for_tags(["reme"]) == ["daily/a.md"] + assert await store.tag_index.tags_for_path("daily\\a.md") == ["reme"] + + allowed = await store.tag_index.paths_for_tags(["reme"]) + hits = await store.keyword_search("alpha", 5, {"paths": allowed}) + assert [chunk.path for chunk in hits] == ["daily/a.md"] + assert note.read_bytes() == original_note + finally: + await store.close() + + +@pytest.mark.asyncio +async def test_both_spellings_address_one_file_across_update_and_delete(monkeypatch, tmp_path: Path) -> None: + """Canonicalize writes and deletes so neither spelling duplicates the file.""" + + monkeypatch.chdir(tmp_path) + store = _standalone_file_store_with_tag_index(name="test", embedding_store="") + await store.start() + try: + await store.upsert( + [(_node("daily\\a.md", ["Legacy"]), [_chunk("chunk-legacy", "daily\\a.md", "legacy body")])], + ) + assert [node.path for node in await store.get_nodes(["daily/a.md"])] == ["daily/a.md"] + assert await store.tag_index.paths_for_tags(["legacy"]) == ["daily/a.md"] + + await store.upsert( + [(_node("daily/a.md", ["ReMe"]), [_chunk("chunk-new", "daily/a.md", "new body")])], + ) + nodes = await store.get_nodes() + assert [node.path for node in nodes] == ["daily/a.md"] + assert list(nodes[0].chunk_ids) == ["chunk-new"] + assert await store.tag_index.paths_for_tags(["legacy"]) == [] + + await store.delete("daily\\a.md") + assert await store.get_nodes() == [] + assert await store.tag_index.paths_for_tags(["reme"]) == [] + assert await store.keyword_search("new", 5, {}) == [] + finally: + await store.close() + + +@pytest.mark.parametrize( + "path", + ["..\\escape.md", "daily\\..\\..\\escape.md", "C:\\absolute\\a.md", "/absolute/a.md"], +) +def test_legacy_normalization_keeps_containment_validation(path: str) -> None: + """Canonicalizing separators must not weaken the workspace containment rule.""" + + async def run() -> None: + index = LocalTagIndex() + await index.start() + with pytest.raises(ValueError): + await index.upsert_nodes([_node(path, ["ReMe"])]) + await index.close() + + asyncio.run(run()) + + +def test_nx_file_graph_migrates_persisted_separators(monkeypatch, tmp_path: Path) -> None: + """Merge legacy node and edge keys when the networkx backend loads.""" + + pytest.importorskip("networkx") + from reme.components.file_graph import NxFileGraph # pylint: disable=import-outside-toplevel + + monkeypatch.chdir(tmp_path) + graph = NxFileGraph(name="test") + legacy = graph._graph.__class__() + link = FileLink(source_path="daily\\a.md", target_path="daily\\b.md") + node = _node("daily\\a.md", ["ReMe"]).model_copy(update={"links": [link]}) + legacy.add_node("daily\\a.md", node=node) + legacy.add_edge( + "daily\\a.md", + "daily\\b.md", + link=link, + ) + with open(graph._graph_file, "wb") as file: + pickle.dump(legacy, file) + + asyncio.run(graph.load()) + + assert list(graph._graph.nodes) == ["daily/a.md", "daily/b.md"] + assert graph._graph.nodes["daily/a.md"]["node"].path == "daily/a.md" + assert [node.path for node in asyncio.run(graph.get_nodes(["daily\\a.md"]))] == ["daily/a.md"] + outlinks = asyncio.run(graph.get_outlinks("daily\\a.md", LinkScopeEnum.ALL)) + assert [link.target_path for link in outlinks] == ["daily/b.md"] + + +def _append_legacy_duplicates(root: Path) -> int: + """Add a second index row per file under the legacy Windows spelling.""" + added = 0 + for path in sorted(root.rglob("*.jsonl.zst")): + lines = [line for line in read_jsonl_zst(path) if line.strip()] + if not lines: + continue + new_lines = list(lines) + for line in lines: + payload = json.loads(line) + legacy = dict(payload) + legacy["path"] = payload["path"].replace("/", "\\") + if "id" in legacy: + legacy["id"] = "legacy-" + str(legacy["id"]) + if "chunk_ids" in legacy: + legacy["chunk_ids"] = ["legacy-" + str(chunk_id) for chunk_id in legacy["chunk_ids"]] + if "st_mtime" in legacy: + legacy["st_mtime"] = legacy["st_mtime"] - 10 + new_lines.append(json.dumps(legacy)) + added += 1 + write_jsonl_zst(path, new_lines) + return added + + +@pytest.mark.asyncio +async def test_mixed_spellings_merge_without_wiping_the_index(monkeypatch, tmp_path: Path) -> None: + """Coexisting legacy and canonical rows merge into one winning node.""" + + monkeypatch.chdir(tmp_path) + note = tmp_path / "daily" / "a.md" + note.parent.mkdir(parents=True) + note.write_text("# A\n\nalpha memory\n", encoding="utf-8") + + store = _standalone_file_store_with_tag_index(name="test", embedding_store="") + await store.start() + await store.upsert( + [(_node("daily/a.md", ["ReMe"]), [_chunk("chunk-a", "daily/a.md", "alpha memory")])], + ) + await store.close() + + assert _append_legacy_duplicates(tmp_path) == 2 + + store = _standalone_file_store_with_tag_index(name="test", embedding_store="") + await store.start() + try: + nodes = await store.get_nodes() + assert [node.path for node in nodes] == ["daily/a.md"] + assert list(nodes[0].chunk_ids) == ["chunk-a"] + assert list(store.file_chunks) == ["chunk-a"] + assert await store.tag_index.paths_for_tags(["reme"]) == ["daily/a.md"] + hits = await store.keyword_search("alpha", 10, {}) + assert {chunk.path for chunk in hits} == {"daily/a.md"} + finally: + await store.close() + + +@pytest.mark.asyncio +async def test_superseded_spelling_cannot_keep_its_text_searchable(monkeypatch, tmp_path: Path) -> None: + """Chunks of the losing spelling must leave the store and the indexes.""" + + monkeypatch.chdir(tmp_path) + note = tmp_path / "daily" / "a.md" + note.parent.mkdir(parents=True) + note.write_text("# A\n\ncurrent body\n", encoding="utf-8") + + store = _standalone_file_store_with_tag_index(name="test", embedding_store="") + await store.start() + await store.upsert( + [(_node("daily/a.md", ["current"]), [_chunk("chunk-current", "daily/a.md", "current body")])], + ) + chunks_path, graph_path = store.chunks_path, store.file_graph._graph_file + await store.close() + + superseded = _node("daily\\a.md", ["obsolete"]).model_copy( + update={"chunk_ids": ["chunk-old"], "st_mtime": 0.0}, + ) + graph_lines = [line for line in read_jsonl_zst(graph_path) if line.strip()] + write_jsonl_zst(graph_path, graph_lines + [superseded.model_dump_json()]) + chunk_lines = [line for line in read_jsonl_zst(chunks_path) if line.strip()] + write_jsonl_zst( + chunks_path, + chunk_lines + + [ + LocalFileStore._serialize_chunk( + _chunk("chunk-old", "daily\\a.md", "removed secret"), + ), + ], + ) + + store = _standalone_file_store_with_tag_index(name="test", embedding_store="") + await store.start() + try: + nodes = await store.get_nodes() + assert [node.path for node in nodes] == ["daily/a.md"] + assert list(nodes[0].chunk_ids) == ["chunk-current"] + assert list(store.file_chunks) == ["chunk-current"] + assert await store.keyword_search("removed", 10, {}) == [] + assert await store.tag_index.paths_for_tags(["obsolete"]) == [] + assert await store.tag_index.paths_for_tags(["current"]) == ["daily/a.md"] + allowed = await store.tag_index.paths_for_tags(["current"]) + assert await store.keyword_search("removed", 10, {"paths": allowed}) == [] + finally: + await store.close() + + +@pytest.mark.asyncio +async def test_one_batch_with_both_spellings_keeps_graph_and_chunks_in_step( + monkeypatch, + tmp_path: Path, +) -> None: + """A duplicate path in one batch must not leave unreachable chunks behind.""" + + monkeypatch.chdir(tmp_path) + store = _standalone_file_store_with_tag_index(name="test", embedding_store="") + await store.start() + try: + await store.upsert( + [ + (_node("daily\\a.md", ["Legacy"]), [_chunk("chunk-legacy", "daily\\a.md", "legacy body")]), + (_node("daily/a.md", ["ReMe"]), [_chunk("chunk-new", "daily/a.md", "new body")]), + ], + ) + nodes = await store.get_nodes() + assert [node.path for node in nodes] == ["daily/a.md"] + assert list(nodes[0].chunk_ids) == ["chunk-new"] + assert list(store.file_chunks) == ["chunk-new"] + assert await store.keyword_search("legacy", 10, {}) == [] + finally: + await store.close() + + store = _standalone_file_store_with_tag_index(name="test", embedding_store="") + await store.start() + try: + assert [node.path for node in await store.get_nodes()] == ["daily/a.md"] + assert list(store.file_chunks) == ["chunk-new"] + assert await store.keyword_search("legacy", 10, {}) == [] + finally: + await store.close() + + +def test_nx_file_graph_load_handles_persisted_edges(monkeypatch, tmp_path: Path) -> None: + """Legacy-key probing must not unpack MultiDiGraph edges as pairs.""" + + pytest.importorskip("networkx") + from reme.components.file_graph import NxFileGraph # pylint: disable=import-outside-toplevel + + monkeypatch.chdir(tmp_path) + graph = NxFileGraph(name="test") + persisted = graph._graph.__class__() + persisted.add_node("daily/a.md", node=_node("daily/a.md", ["ReMe"])) + persisted.add_edge( + "daily/a.md", + "daily/b.md", + link=FileLink(source_path="daily/a.md", target_path="daily/b.md"), + ) + with open(graph._graph_file, "wb") as file: + pickle.dump(persisted, file) + + asyncio.run(graph.load()) + assert graph._has_legacy_keys() is False + assert [node.path for node in asyncio.run(graph.get_nodes())] == ["daily/a.md"] + outlinks = asyncio.run(graph.get_outlinks("daily/a.md", LinkScopeEnum.ALL)) + assert [link.target_path for link in outlinks] == ["daily/b.md"] + + +def test_nx_file_graph_keeps_only_the_winning_spellings_links(monkeypatch, tmp_path: Path) -> None: + """Migration must rebuild edges from the record that won the merge.""" + + pytest.importorskip("networkx") + from reme.components.file_graph import NxFileGraph # pylint: disable=import-outside-toplevel + + monkeypatch.chdir(tmp_path) + graph = NxFileGraph(name="test") + legacy = _node("daily\\a.md", ["Legacy"]).model_copy( + update={ + "st_mtime": 0.0, + "links": [FileLink(source_path="daily\\a.md", target_path="daily\\obsolete.md")], + }, + ) + current = _node("daily/a.md", ["ReMe"]).model_copy( + update={"links": [FileLink(source_path="daily/a.md", target_path="daily/current.md")]}, + ) + persisted = graph._graph.__class__() + persisted.add_node("daily\\a.md", node=legacy) + persisted.add_node("daily/a.md", node=current) + persisted.add_edge( + "daily\\a.md", + "daily\\obsolete.md", + link=FileLink(source_path="daily\\a.md", target_path="daily\\obsolete.md"), + ) + persisted.add_edge( + "daily/a.md", + "daily/current.md", + link=FileLink(source_path="daily/a.md", target_path="daily/current.md"), + ) + with open(graph._graph_file, "wb") as file: + pickle.dump(persisted, file) + + asyncio.run(graph.load()) + + assert sorted(graph._graph.nodes) == ["daily/a.md", "daily/current.md"] + outlinks = asyncio.run(graph.get_outlinks("daily/a.md", LinkScopeEnum.ALL)) + assert [link.target_path for link in outlinks] == ["daily/current.md"]