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