From 82c1395341ea100d3e0b1b8481f08f1eed87c594 Mon Sep 17 00:00:00 2001 From: "jinli.yl" Date: Thu, 10 Sep 2026 12:27:52 +0800 Subject: [PATCH] Refine tag filtering and automatic memory tagging --- .gitignore | 3 - reme/components/file_store/base_file_store.py | 18 ++ .../file_store/faiss_local_file_store.py | 7 +- .../components/file_store/local_file_store.py | 197 +++++------------- .../file_store/zvec_local_file_store.py | 2 +- reme/components/keyword_index/bm25_index.py | 69 +++--- reme/components/tag_index/base_tag_index.py | 45 ++-- reme/components/tag_index/local_tag_index.py | 47 +++-- reme/config/default.yaml | 8 +- reme/steps/evolve/auto_memory.py | 46 ++-- reme/steps/evolve/auto_tag.py | 182 +++++++++------- reme/steps/evolve/auto_tag.yaml | 49 +++-- reme/steps/evolve/dream/utils.py | 3 +- reme/steps/index/list_tags.py | 4 +- reme/steps/index/search.py | 89 ++++---- reme/steps/index/update_changes.py | 8 +- tests/unit/test_auto_tag.py | 144 +++++++++---- tests/unit/test_background_steps.py | 5 +- tests/unit/test_evolve_utils.py | 32 ++- tests/unit/test_faiss_index_maintenance.py | 1 + tests/unit/test_file_store_consistency.py | 26 ++- tests/unit/test_search_step.py | 93 +++++---- tests/unit/test_tag_index.py | 94 +++++---- tests/unit/test_zvec_file_store.py | 1 + 24 files changed, 629 insertions(+), 544 deletions(-) diff --git a/.gitignore b/.gitignore index 64b6f32b..da160dc1 100644 --- a/.gitignore +++ b/.gitignore @@ -77,9 +77,6 @@ benchmark/*/dataset/ # Benchmark outputs (created on demand by run.py via mkdir) benchmark/*/results/ -# Meta-ReMe generated bundles (build artifacts) -meta-reme/bundles/ - # integration tests outputs tests/integration/logs/ daily/ diff --git a/reme/components/file_store/base_file_store.py b/reme/components/file_store/base_file_store.py index 91bcaa73..414be787 100644 --- a/reme/components/file_store/base_file_store.py +++ b/reme/components/file_store/base_file_store.py @@ -6,6 +6,7 @@ from contextlib import asynccontextmanager from functools import wraps from ..base_component import BaseComponent +from ..tag_index import BaseTagIndex from ...enumeration import ComponentEnum, LinkScopeEnum from ...schema import FileChunk, FileLink, FileNode @@ -23,9 +24,26 @@ class BaseFileStore(BaseComponent): def __init__(self, **kwargs): super().__init__(**kwargs) + self.tag_index: BaseTagIndex | None = None self._maintenance_lock = asyncio.Lock() self._maintenance_lock_owner = None + @property + def tag_index_enabled(self) -> bool: + """Whether this file-store backend has a tag index configured.""" + return self.tag_index is not None + + def require_tag_index(self) -> BaseTagIndex: + """Return the configured tag index or fail with one consistent error.""" + if self.tag_index is None: + raise RuntimeError("tag index is not configured") + return self.tag_index + + @property + def embedding_dimensions(self) -> int: + """Vector dimensions used for memory estimates; zero when unavailable.""" + return 0 + @asynccontextmanager async def _maintenance_guard(self): """Serialize maintenance and mutations, allowing nested backend overrides.""" diff --git a/reme/components/file_store/faiss_local_file_store.py b/reme/components/file_store/faiss_local_file_store.py index 12684e81..02613eab 100644 --- a/reme/components/file_store/faiss_local_file_store.py +++ b/reme/components/file_store/faiss_local_file_store.py @@ -648,12 +648,7 @@ class FaissLocalFileStore(LocalFileStore): async def vector_search(self, query: str, limit: int, search_filter: dict) -> list[FileChunk]: index_empty = self._faiss_index is None or self._faiss_index.ntotal == 0 embedding_unavailable = self.embedding_store is None or self._embedding_rebuild_pending - if ( - embedding_unavailable - or not query - or limit <= 0 - or (index_empty and getattr(self.embedding_store, "is_healthy", True)) - ): + if embedding_unavailable or not query or limit <= 0 or (index_empty and self.embedding_store.is_healthy): return [] query_embedding = await self._get_query_embedding(query) diff --git a/reme/components/file_store/local_file_store.py b/reme/components/file_store/local_file_store.py index 21f6fd28..bccd8fbf 100644 --- a/reme/components/file_store/local_file_store.py +++ b/reme/components/file_store/local_file_store.py @@ -27,8 +27,6 @@ CachedEmbedding = tuple[str, np.ndarray] _EMBEDDING_F16_B64_FIELD = "_embedding_f16_b64" _EMBEDDING_F16_DTYPE = np.dtype(" int: + if self.embedding_store is None: + return 0 + return max(0, int(getattr(self.embedding_store, "dimensions", 0))) + # -- lifecycle ------------------------------------------------------------ async def _start(self) -> None: @@ -138,7 +142,7 @@ class LocalFileStore(BaseFileStore): self.embedding_store is None or self._embedding_rebuild_pending or was_healthy - or not getattr(self.embedding_store, "is_healthy", True) + or not self.embedding_store.is_healthy ): return self.logger.info(f"{self.name}: embedding provider recovered; scheduling missing-vector backfill") @@ -198,7 +202,7 @@ class LocalFileStore(BaseFileStore): return None embedding_generation = self._embedding_space_generation - was_healthy = bool(getattr(embedding_store, "is_healthy", True)) + was_healthy = embedding_store.is_healthy try: query_embedding = await embedding_store.get_embedding(query) except Exception as e: @@ -250,7 +254,7 @@ class LocalFileStore(BaseFileStore): tag_sync_started_at = time.monotonic() tag_synced = await self._rebuild_tag_index("startup synchronization") self.logger.info( - f"{self.name}: tag index sync complete: enabled={self.tag_index is not None}, " + f"{self.name}: tag index sync complete: enabled={self.tag_index_enabled}, " f"healthy={tag_synced}, " f"elapsed={time.monotonic() - tag_sync_started_at:.3f}s", ) @@ -504,7 +508,7 @@ class LocalFileStore(BaseFileStore): return total = len(missing) - batch_size = max(1, int(getattr(self.embedding_store, "max_batch_size", 10))) + batch_size = max(1, int(self.embedding_store.max_batch_size)) self.logger.info(f"{self.name}: embedding backfill started: total={total}, batch_size={batch_size}") try: if not skip_health_check: @@ -638,13 +642,14 @@ class LocalFileStore(BaseFileStore): async def _rebuild_tag_index(self, reason: str) -> bool: """Best-effort rebuild that never lets an optional tag index block the file store.""" - if self.tag_index is None: + if not self.tag_index_enabled: self._tag_index_rebuild_required = False self._tag_indexed_file_count = 0 return True + tag_index = self.require_tag_index() try: nodes = await self.file_graph.get_nodes() - await self.tag_index.rebuild(nodes) + await tag_index.rebuild(nodes) except Exception: self._tag_index_rebuild_required = True self.logger.exception( @@ -653,20 +658,20 @@ class LocalFileStore(BaseFileStore): await self._quarantine_tag_index(reason) return False self._tag_index_rebuild_required = False - self.tag_index.set_healthy(True) - self._tag_indexed_file_count = self.tag_index.n_files + tag_index.set_healthy(True) + self._tag_indexed_file_count = tag_index.n_files return True async def _quarantine_tag_index(self, reason: str) -> None: """Fail closed after a reconciliation error without propagating cleanup failures.""" - assert self.tag_index is not None - self.tag_index.set_healthy(False) + tag_index = self.require_tag_index() + tag_index.set_healthy(False) try: - await self.tag_index.clear() + await tag_index.clear() except Exception: self.logger.exception(f"{self.name}: failed to clear unhealthy tag index during {reason}") finally: - self.tag_index.set_healthy(False) + tag_index.set_healthy(False) async def _reindex_tag(self) -> dict: """Synchronously rebuild the optional tag index from the authoritative file graph.""" @@ -676,33 +681,35 @@ class LocalFileStore(BaseFileStore): async def _upsert_tag_nodes(self, nodes: list[FileNode]) -> None: """Update tags without allowing optional-index failures to block other indexes.""" - if self.tag_index is None: + if not self.tag_index_enabled: return + tag_index = self.require_tag_index() if self._tag_index_rebuild_required: await self._rebuild_tag_index("retry before incremental update") return try: - await self.tag_index.upsert_nodes(nodes) - self.tag_index.set_healthy(True) + await tag_index.upsert_nodes(nodes) + tag_index.set_healthy(True) except Exception: self._tag_index_rebuild_required = True - self.tag_index.set_healthy(False) + tag_index.set_healthy(False) self.logger.exception(f"{self.name}: incremental tag index update failed; rebuilding from graph") await self._rebuild_tag_index("incremental update recovery") async def _delete_tag_paths(self, paths: list[str]) -> None: """Delete tag paths without allowing optional-index failures to block other indexes.""" - if self.tag_index is None: + if not self.tag_index_enabled: return + tag_index = self.require_tag_index() if self._tag_index_rebuild_required: await self._rebuild_tag_index("retry before incremental delete") return try: - await self.tag_index.delete(paths) - self.tag_index.set_healthy(True) + await tag_index.delete(paths) + tag_index.set_healthy(True) except Exception: self._tag_index_rebuild_required = True - self.tag_index.set_healthy(False) + tag_index.set_healthy(False) self.logger.exception(f"{self.name}: incremental tag index delete failed; rebuilding from graph") await self._rebuild_tag_index("incremental delete recovery") @@ -833,7 +840,7 @@ class LocalFileStore(BaseFileStore): return embedding_store = self.embedding_store embedding_generation = self._embedding_space_generation - was_healthy = bool(getattr(embedding_store, "is_healthy", True)) + was_healthy = embedding_store.is_healthy try: await embedding_store.get_node_embeddings(chunks) except Exception as e: @@ -905,125 +912,22 @@ class LocalFileStore(BaseFileStore): if self.keyword_index: await self.keyword_index.clear() await self.file_graph.clear() - if self.tag_index is not None: + if self.tag_index_enabled: + tag_index = self.require_tag_index() try: - await self.tag_index.clear() + await tag_index.clear() self._tag_index_rebuild_required = False self._tag_indexed_file_count = 0 - self.tag_index.set_healthy(True) + tag_index.set_healthy(True) except Exception: self._tag_index_rebuild_required = True - self.tag_index.set_healthy(False) + tag_index.set_healthy(False) self.logger.exception(f"{self.name}: tag index clear failed; rebuilding from empty graph") await self._rebuild_tag_index("clear recovery") self._mutation_generation += 1 # -- search --------------------------------------------------------------- - def resolve_filtered_chunk_ids(self, allowed_paths: set[str], search_filter: dict) -> set[str]: - """Resolve a tag-derived path set and existing filters to chunk IDs.""" - return { - chunk_id - for chunk_id, chunk in self.file_chunks.items() - if chunk.path in allowed_paths and self._matches_search_filter(chunk, search_filter) - } - - async def filtered_vector_search( - self, - query: str, - limit: int, - eligible_chunk_ids: set[str], - ) -> list[FileChunk]: - """Search a tag-filtered vector domain, choosing exact scan or ANN.""" - if limit <= 0 or not eligible_chunk_ids or self.embedding_store is None or self._embedding_rebuild_pending: - return [] - - all_vector_chunks = [ - chunk for chunk in self.file_chunks.values() if self._embedding_dim_matches(chunk.embedding) - ] - eligible = [chunk for chunk in all_vector_chunks if chunk.id in eligible_chunk_ids] - if not eligible: - return [] - - exact = len(eligible) < _FILTERED_EXACT_RATIO * len(all_vector_chunks) or len(eligible) < ( - _FILTERED_FULL_RANK_MULTIPLIER * limit - ) - if not exact: - return await self.vector_search( - query, - limit, - {"_eligible_chunk_ids": frozenset(eligible_chunk_ids)}, - ) - - query_embedding = await self._get_query_embedding(query) - if query_embedding is None: - return [] - - return_all = len(eligible) <= _FILTERED_FULL_RANK_MULTIPLIER * limit - scored: list[tuple[float, int, FileChunk]] = [] - for start in range(0, len(eligible), _VECTOR_SEARCH_BATCH_SIZE): - batch = eligible[start : start + _VECTOR_SEARCH_BATCH_SIZE] - matrix = np.stack([chunk.embedding for chunk in batch]) - similarities = batch_cosine_similarity(query_embedding.reshape(1, -1), matrix)[0] - scored.extend( - (float(score), -(start + offset), chunk) - for offset, (chunk, score) in enumerate(zip(batch, similarities)) - ) - scored.sort(key=lambda item: (-item[0], -item[1])) - if not return_all: - scored = scored[:limit] - return [ - chunk.model_copy(update={"scores": {"vector": score, "score": score}}) - for score, _neg_order, chunk in scored - ] - - async def filtered_keyword_search( - self, - query: str, - limit: int, - eligible_chunk_ids: set[str], - ) -> list[FileChunk]: - """Compute exact global-statistics BM25 scores inside a chunk domain.""" - if not self.keyword_index or limit <= 0 or not eligible_chunk_ids: - return [] - query = query.strip() - if not query: - return [] - - try: - eligible_keyword_ids = eligible_chunk_ids.intersection(self.keyword_index.document_ids) - except NotImplementedError: - # Compatibility path for a third-party keyword index that cannot - # expose its live document IDs. - return await self.keyword_search( - query, - limit, - {"_eligible_chunk_ids": frozenset(eligible_chunk_ids)}, - ) - if not eligible_keyword_ids: - return [] - - return_all = len(eligible_keyword_ids) <= _FILTERED_FULL_RANK_MULTIPLIER * limit - try: - if return_all: - doc_scores = await self.keyword_index.score_documents(query, eligible_keyword_ids) - else: - doc_scores = await self.keyword_index.retrieve_filtered(query, limit, eligible_keyword_ids) - except NotImplementedError: - # A third-party index may expose document IDs but not implement an - # efficient filtered scoring extension. - return await self.keyword_search( - query, - limit, - {"_eligible_chunk_ids": frozenset(eligible_chunk_ids)}, - ) - - return [ - chunk.model_copy(update={"scores": {"keyword": score, "score": score}}) - for doc_id, score in doc_scores.items() - if (chunk := self.file_chunks.get(doc_id)) is not None - ] - async def vector_search(self, query: str, limit: int, search_filter: dict) -> list[FileChunk]: if limit <= 0: return [] @@ -1073,17 +977,32 @@ class LocalFileStore(BaseFileStore): ] async def keyword_search(self, query: str, limit: int, search_filter: dict) -> list[FileChunk]: - if not self.keyword_index: + if not self.keyword_index or limit <= 0: return [] query = query.strip() if not query: return [] - retrieve_limit = limit if search_filter: - retrieve_limit = max(limit, getattr(self.keyword_index, "n_docs", limit)) - doc_id_score_dict = await self.keyword_index.retrieve(query, limit=retrieve_limit) + eligible_ids = { + chunk.id for chunk in self.file_chunks.values() if self._matches_search_filter(chunk, search_filter) + } + if not eligible_ids: + return [] + try: + eligible_ids.intersection_update(self.keyword_index.document_ids) + if not eligible_ids: + return [] + doc_id_score_dict = await self.keyword_index.retrieve_filtered(query, limit, eligible_ids) + except NotImplementedError: + # Compatibility path for third-party indexes implementing only retrieve(). + doc_id_score_dict = await self.keyword_index.retrieve( + query, + limit=max(limit, getattr(self.keyword_index, "n_docs", limit)), + ) + else: + doc_id_score_dict = await self.keyword_index.retrieve(query, limit=limit) results = [] for doc_id, score in doc_id_score_dict.items(): chunk = self.file_chunks.get(doc_id) @@ -1138,14 +1057,13 @@ class LocalFileStore(BaseFileStore): if not search_filter: return True - eligible_chunk_ids = search_filter.get("_eligible_chunk_ids") 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 (eligible_chunk_ids is not None and chunk.id not in eligible_chunk_ids) or ( - exact_paths and chunk.path not in exact_paths - ): + if has_exact_path_filter and chunk.path not in exact_paths: return False prefixes = [] @@ -1181,7 +1099,6 @@ class LocalFileStore(BaseFileStore): "start_date", "end_date", "strict_date_filter", - "_eligible_chunk_ids", } for key, value in search_filter.items(): if key not in reserved: diff --git a/reme/components/file_store/zvec_local_file_store.py b/reme/components/file_store/zvec_local_file_store.py index afad4b4d..58a0300a 100644 --- a/reme/components/file_store/zvec_local_file_store.py +++ b/reme/components/file_store/zvec_local_file_store.py @@ -441,7 +441,7 @@ class ZvecLocalFileStore(LocalFileStore): if self.embedding_store is None or self._embedding_rebuild_pending or not query or limit <= 0: return [] index_empty = self._collection is None or not self._indexed_ids - if index_empty and getattr(self.embedding_store, "is_healthy", True): + if index_empty and self.embedding_store.is_healthy: return [] query_embedding = await self._get_query_embedding(query) diff --git a/reme/components/keyword_index/bm25_index.py b/reme/components/keyword_index/bm25_index.py index d19c1362..4bd4c14d 100644 --- a/reme/components/keyword_index/bm25_index.py +++ b/reme/components/keyword_index/bm25_index.py @@ -280,29 +280,50 @@ class BM25Index(BaseKeywordIndex): self._remove_doc(doc_id) self._idf_cache = {} - def _score_query(self, query_ids: list[int], n_docs: int) -> np.ndarray: - """Compute BM25 scores across all docs; deleted docs zeroed out.""" + def _score_query(self, query_ids: list[int], candidate_idxs: np.ndarray | None = None) -> np.ndarray: + """Compute BM25 scores globally or for selected live document indexes.""" + n_docs = self.n_docs + if n_docs == 0: + size = self._doc_lens.size if candidate_idxs is None else candidate_idxs.size + return np.zeros(size, dtype=np.float32) + avg_len = self.total_len / n_docs k1, b = self.k1, self.b denom_base = k1 * (1.0 - b) denom_norm = k1 * b / avg_len if avg_len > 0 else 0.0 - scores = np.zeros(self._doc_lens.size, dtype=np.float32) + size = self._doc_lens.size if candidate_idxs is None else candidate_idxs.size + scores = np.zeros(size, dtype=np.float32) for tid in query_ids: - doc_idxs = self._posting_doc_idxs.get(tid) - if doc_idxs is None or doc_idxs.size == 0: + posting_idxs = self._posting_doc_idxs.get(tid) + if posting_idxs is None or posting_idxs.size == 0: continue idf = self._get_idf(tid, n_docs) if idf == 0.0: continue - tfs = self._posting_tfs[tid].astype(np.float32) - d_lens = self._doc_lens[doc_idxs].astype(np.float32) + + if candidate_idxs is None: + posting_positions = slice(None) + score_positions = posting_idxs + else: + _common, posting_positions, score_positions = np.intersect1d( + posting_idxs, + candidate_idxs, + assume_unique=True, + return_indices=True, + ) + if score_positions.size == 0: + continue + + doc_idxs = posting_idxs[posting_positions] + tfs = self._posting_tfs[tid][posting_positions].astype(np.float32) + doc_lens = self._doc_lens[doc_idxs].astype(np.float32) # Each doc_idx appears at most once per posting list (Counter dedups # within a doc, and updates allocate fresh idxs), so fancy-index # accumulation is safe here. - scores[doc_idxs] += idf * tfs * (k1 + 1.0) / (tfs + denom_base + denom_norm * d_lens) + scores[score_positions] += idf * tfs * (k1 + 1.0) / (tfs + denom_base + denom_norm * doc_lens) - if self._deleted.any(): + if candidate_idxs is None and self._deleted.any(): scores[self._deleted] = 0.0 return scores @@ -316,33 +337,7 @@ class BM25Index(BaseKeywordIndex): sorted({self._doc_id_to_idx[doc_id] for doc_id in document_ids if doc_id in self._doc_id_to_idx}), dtype=np.int32, ) - scores = np.zeros(candidate_idxs.size, dtype=np.float32) - n_docs = self.n_docs - if candidate_idxs.size == 0 or n_docs == 0: - return candidate_idxs, scores - - avg_len = self.total_len / n_docs - denom_base = self.k1 * (1.0 - self.b) - denom_norm = self.k1 * self.b / avg_len if avg_len > 0 else 0.0 - for tid in query_ids: - posting_idxs = self._posting_doc_idxs.get(tid) - if posting_idxs is None or posting_idxs.size == 0: - continue - _common, posting_positions, candidate_positions = np.intersect1d( - posting_idxs, - candidate_idxs, - assume_unique=True, - return_indices=True, - ) - if posting_positions.size == 0: - continue - idf = self._get_idf(tid, n_docs) - if idf == 0.0: - continue - tfs = self._posting_tfs[tid][posting_positions].astype(np.float32) - doc_lens = self._doc_lens[candidate_idxs[candidate_positions]].astype(np.float32) - scores[candidate_positions] += idf * tfs * (self.k1 + 1.0) / (tfs + denom_base + denom_norm * doc_lens) - return candidate_idxs, scores + return candidate_idxs, self._score_query(query_ids, candidate_idxs) async def retrieve(self, query: str, limit: int = 3) -> dict[str, float]: """BM25 retrieval; returns {doc_id: score} sorted by score descending.""" @@ -353,7 +348,7 @@ class BM25Index(BaseKeywordIndex): if not query_ids: return {} - scores = self._score_query(query_ids, n_docs) + scores = self._score_query(query_ids) top_idxs = self._top_k(scores, limit) return {self._doc_ids[int(i)]: float(scores[int(i)]) for i in top_idxs} diff --git a/reme/components/tag_index/base_tag_index.py b/reme/components/tag_index/base_tag_index.py index c7f2f5ec..95882945 100644 --- a/reme/components/tag_index/base_tag_index.py +++ b/reme/components/tag_index/base_tag_index.py @@ -1,20 +1,36 @@ """Abstract interface for file-level tag indexes derived from graph nodes.""" from abc import abstractmethod +from typing import ClassVar, Literal, TypedDict from ..base_component import BaseComponent from ...enumeration import ComponentEnum from ...schema import FileNode +TagOrderBy = Literal["tag", "file_count"] +TagOrder = Literal["asc", "desc"] +TagListItem = tuple[str, int] + + +class TagListResult(TypedDict): + """One page of active tags and their indexed-file counts.""" + + total_tags: int + total_pages: int + page: int + range: tuple[int, int] + items: list[TagListItem] + class BaseTagIndex(BaseComponent): """A rebuildable index of normalized ``FileNode`` frontmatter tags.""" component_type = ComponentEnum.TAG_INDEX + reserved_tag_keys: ClassVar[frozenset[str]] = frozenset() - def __init__(self, tag_key="tags", **kwargs): + def __init__(self, tag_key: object = "memory_tags", **kwargs): super().__init__(**kwargs) - self.tag_key = tag_key + self._tag_key = self._validate_tag_key(tag_key) self.is_healthy = True @property @@ -24,13 +40,19 @@ class BaseTagIndex(BaseComponent): @tag_key.setter def tag_key(self, value: object) -> None: - self._tag_key = self._validate_tag_key(value) + tag_key = self._validate_tag_key(value) + if tag_key != self._tag_key: + self._tag_key = tag_key + self.set_healthy(False) def _validate_tag_key(self, value: object) -> str: """Validate and normalize the configured frontmatter tag field.""" if not isinstance(value, str) or not value.strip(): raise ValueError("tag_key must be a non-empty string") - return value.strip() + tag_key = value.strip() + if tag_key in self.reserved_tag_keys: + raise ValueError(f"tag_key must not be a reserved frontmatter key: {tag_key!r}") + return tag_key def set_healthy(self, healthy: bool) -> None: """Mark whether lookups can safely use the current derived state.""" @@ -45,14 +67,9 @@ class BaseTagIndex(BaseComponent): def normalize_tags(self, value: object) -> list[str]: """Return canonical tags according to this index's configured limits.""" + @abstractmethod def normalize_query_tags(self, value: object) -> list[str]: - """Return canonical lookup tags. - - Existing third-party indexes remain compatible by inheriting their - normal tag canonicalization. Implementations with per-file count limits - should override this method so lookup expressions are not truncated. - """ - return self.normalize_tags(value) + """Return canonical lookup tags without per-file count limits.""" @abstractmethod async def rebuild(self, nodes: list[FileNode]) -> None: @@ -79,10 +96,10 @@ class BaseTagIndex(BaseComponent): self, *, page: int = 1, - order_by: str = "tag", - order: str | None = None, + order_by: TagOrderBy = "tag", + order: TagOrder | None = None, page_size: int = 100, - ) -> dict[str, object]: + ) -> TagListResult: """Return a page of active tags with counts and a 1-based result range.""" @abstractmethod diff --git a/reme/components/tag_index/local_tag_index.py b/reme/components/tag_index/local_tag_index.py index f1e3eb0f..541201a8 100644 --- a/reme/components/tag_index/local_tag_index.py +++ b/reme/components/tag_index/local_tag_index.py @@ -3,7 +3,7 @@ import asyncio from pathlib import PurePosixPath -from .base_tag_index import BaseTagIndex +from .base_tag_index import BaseTagIndex, TagListItem, TagListResult, TagOrder, TagOrderBy from ..component_registry import R from ...schema import FileFrontMatter, FileNode @@ -12,10 +12,12 @@ from ...schema import FileFrontMatter, FileNode class LocalTagIndex(BaseTagIndex): """Maintain bidirectional path/tag relationships without separate source I/O.""" + reserved_tag_keys = frozenset(FileFrontMatter.model_fields) + def __init__( self, - tag_key: object = "tags", - max_tags_per_file: int = 8, + tag_key: object = "memory_tags", + max_tags_per_file: int = 3, max_tag_length: int = 64, **kwargs, ): @@ -26,12 +28,6 @@ class LocalTagIndex(BaseTagIndex): self.tag_to_paths: dict[str, set[str]] = {} self._maintenance_lock = asyncio.Lock() - def _validate_tag_key(self, value: object) -> str: - tag_key = super()._validate_tag_key(value) - if tag_key in FileFrontMatter.model_fields: - raise ValueError(f"tag_key must not be a reserved frontmatter key: {tag_key!r}") - return tag_key - @property def n_files(self) -> int: return len(self.path_to_tags) @@ -51,8 +47,8 @@ class LocalTagIndex(BaseTagIndex): for item in value: if isinstance(item, bool) or not isinstance(item, (str, int)): continue - raw = str(item).strip() - if not raw or len(raw) > self.max_tag_length or any(char.isspace() for char in raw): + raw = " ".join(str(item).split()) + if not raw or len(raw) > self.max_tag_length: continue if not any(char.isalnum() for char in raw): continue @@ -86,7 +82,8 @@ class LocalTagIndex(BaseTagIndex): prepared: list[tuple[str, tuple[str, ...]]] = [] for node in nodes: path = self._validate_path(node.path) - tags = self.normalize_tags(node.front_matter.model_dump().get(self.tag_key)) + frontmatter = node.front_matter.model_extra or {} + tags = self.normalize_tags(frontmatter.get(self.tag_key)) prepared.append((path, tuple(tags))) return prepared @@ -162,41 +159,45 @@ class LocalTagIndex(BaseTagIndex): self, *, page: int = 1, - order_by: str = "tag", - order: str | None = None, + order_by: TagOrderBy = "tag", + order: TagOrder | None = None, page_size: int = 100, - ) -> dict[str, object]: + ) -> TagListResult: """Return a deterministic page of active tags, counts, and its 1-based range.""" page = self._positive_int("page", page) page_size = self._positive_int("page_size", page_size) if page_size > 1000: raise ValueError("page_size must be less than or equal to 1000") - order_by = str(order_by).lower() + if not isinstance(order_by, str): + raise ValueError("order_by must be one of ['file_count', 'tag']") + order_by = order_by.lower() if order_by not in {"tag", "file_count"}: raise ValueError("order_by must be one of ['file_count', 'tag']") if order is None: order = "asc" if order_by == "tag" else "desc" - order = str(order).lower() + if not isinstance(order, str): + raise ValueError("order must be one of ['asc', 'desc']") + order = order.lower() if order not in {"asc", "desc"}: raise ValueError("order must be one of ['asc', 'desc']") async with self._maintenance_lock: - items = [[tag, len(paths)] for tag, paths in self.tag_to_paths.items()] if self.is_healthy else [] + items: list[TagListItem] = ( + [(tag, len(paths)) for tag, paths in self.tag_to_paths.items()] if self.is_healthy else [] + ) if order_by == "tag": items.sort(key=lambda item: item[0], reverse=order == "desc") else: - # Keep tag ascending as a stable, deterministic tiebreaker. - items.sort(key=lambda item: item[0]) - items.sort(key=lambda item: item[1], reverse=order == "desc") + direction = 1 if order == "asc" else -1 + items.sort(key=lambda item: (direction * item[1], item[0])) total_tags = len(items) total_pages = (total_tags + page_size - 1) // page_size - page = min(page, max(total_pages, 1)) start = (page - 1) * page_size page_items = items[start : start + page_size] - item_range = [start + 1, start + len(page_items)] if page_items else [0, 0] + item_range = (start + 1, start + len(page_items)) if page_items else (0, 0) return { "total_tags": total_tags, "total_pages": total_pages, diff --git a/reme/config/default.yaml b/reme/config/default.yaml index c6ea233f..561d3962 100644 --- a/reme/config/default.yaml +++ b/reme/config/default.yaml @@ -392,7 +392,7 @@ jobs: backend: base description: >- List active tags with file counts. Returns {total_tags, total_pages, page, range, items}; range is the inclusive - 1-based item range and each item is [tag, file_count]. An oversized page returns the last page. + 1-based item range and each item is [tag, file_count]. A page past the end returns empty items and range [0, 0]. parameters: type: object properties: @@ -400,7 +400,7 @@ jobs: type: integer minimum: 1 default: 1 - description: "1-based page; values past the end use the last page" + description: "1-based page; values past the end return an empty page" order_by: type: string enum: [tag, file_count] @@ -936,8 +936,8 @@ components: tag_index: default: backend: local - tag_key: tags - max_tags_per_file: 8 + tag_key: memory_tags + max_tags_per_file: 3 max_tag_length: 64 file_store: diff --git a/reme/steps/evolve/auto_memory.py b/reme/steps/evolve/auto_memory.py index a244391d..8cf29a1f 100644 --- a/reme/steps/evolve/auto_memory.py +++ b/reme/steps/evolve/auto_memory.py @@ -118,12 +118,12 @@ class AutoMemoryStep(BaseStep): notes = list_response.metadata.get("notes") or [] return self._find_session_note(notes, session_id) - async def _ensure_memory_frontmatter(self, path: str, session_id: str) -> None: - current = self._frontmatter(path) + async def _ensure_session_frontmatter(self, path: str, session_id: str) -> None: metadata = { _SESSION_ID_KEY: session_id, _SOURCE_CONVERSATION_KEY: self._session_link(session_id), } + current = self._frontmatter(path) if all(current.get(key) == value for key, value in metadata.items()): return response = await self.run_job( @@ -276,7 +276,7 @@ class AutoMemoryStep(BaseStep): # pylint: disable=too-many-return-statements async def execute(self): assert self.context is not None - self.context["modified_paths"] = [] + self.context["changes"] = [] raw_messages = self.context.get("messages") or [] session_id: str = self.context.get("session_id", "") memory_hint: str = self.context.get("memory_hint", "") @@ -383,32 +383,28 @@ class AutoMemoryStep(BaseStep): self.logger.info(f"[{self.name}] done without note session_id={session_id!r} modified=False") return note_path = str(note["path"]) - try: - if not created: - await self._ensure_memory_frontmatter(note_path, session_id) - if not created: + else: + try: + await self._ensure_session_frontmatter(note_path, session_id) note_path = await self._rename_from_frontmatter_name(note_path, day) - except RuntimeError as exc: - modified = self._note_modified(before_note_path, before_note_bytes, note_path) - if modified and note_path: - self.context["modified_paths"] = [note_path] - self.context.response.success = False - self.context.response.answer = str(exc) - self.context.response.metadata.update( - { - "date": day, - "path": note_path, - "created": created, - "modified": modified, - "n_messages": len(messages), - }, - ) - self.logger.info(f"[{self.name}] post-write failed path={note_path} answer={str(exc)!r}") - return + except RuntimeError as exc: + self.context.response.success = False + self.context.response.answer = str(exc) + self.context.response.metadata.update( + { + "date": day, + "path": note_path, + "created": created, + "modified": self._note_modified(before_note_path, before_note_bytes, note_path), + "n_messages": len(messages), + }, + ) + self.logger.info(f"[{self.name}] post-update failed path={note_path} answer={str(exc)!r}") + return modified = self._note_modified(before_note_path, before_note_bytes, note_path) if modified: - self.context["modified_paths"] = [note_path] + self.context["changes"] = [{"change": "added" if created else "modified", "path": note_path}] daily_dir = self.config_value("daily_dir") self.logger.info(f"[{self.name}] refresh index start date={day} daily_dir={daily_dir}") index_payload = await refresh_day_index(self.file_store, day, daily_dir) diff --git a/reme/steps/evolve/auto_tag.py b/reme/steps/evolve/auto_tag.py index 63c5c730..3324507e 100644 --- a/reme/steps/evolve/auto_tag.py +++ b/reme/steps/evolve/auto_tag.py @@ -1,6 +1,8 @@ -"""auto_tag — generate frontmatter tags for modified Markdown files.""" +"""Generate entity-oriented memory tags for added or modified Markdown files.""" +from dataclasses import dataclass from pathlib import Path +from typing import Literal import frontmatter @@ -11,88 +13,107 @@ from ..file_io._path import display_path, resolve_path from ..index import normalize_posix_path from ...components import R -_MAX_TAGS = 8 -_MAX_TAG_LENGTH = 64 +_MAX_MEMORY_TAGS = 3 +_MAX_MEMORY_TAG_LENGTH = 64 +_SUPPORTED_CHANGES = {"added", "modified"} -def normalize_tags(value) -> list[str]: - """Return unique retrieval-friendly tags that satisfy the storage contract.""" +@dataclass(frozen=True) +class _TagTarget: + change: Literal["added", "modified"] + path: str + + +def normalize_memory_tags(value: object) -> list[str]: + """Normalize up to three human-readable entity labels for frontmatter storage.""" if not isinstance(value, list): return [] tags: list[str] = [] seen: set[str] = set() for item in value: - if isinstance(item, bool) or not isinstance(item, (str, int)): + if not isinstance(item, str): continue - tag = str(item).strip() - if not tag or len(tag) > _MAX_TAG_LENGTH: + tag = " ".join(item.split()) + if not tag or len(tag) > _MAX_MEMORY_TAG_LENGTH or not any(char.isalnum() for char in tag): continue - if any(char.isspace() for char in tag): + canonical = tag.casefold() + if canonical in seen: continue - if not any(char.isalnum() for char in tag): - continue - dedupe_key = tag.casefold() - if dedupe_key in seen: - continue - seen.add(dedupe_key) + seen.add(canonical) tags.append(tag) - if len(tags) >= _MAX_TAGS: + if len(tags) >= _MAX_MEMORY_TAGS: break return tags @R.register("auto_tag_step") class AutoTagStep(BaseStep): - """Generate tags for each eligible path produced by an earlier Step.""" + """Update memory tags for Markdown files described by the common ``changes`` contract.""" def __init__(self, **kwargs): super().__init__(**kwargs) self.tools = ["read", "list_tags", "frontmatter_read", "frontmatter_update"] - def _eligible_paths(self) -> tuple[list[str], list[str]]: + def _targets(self) -> tuple[list[_TagTarget], list[dict[str, str]]]: + """Validate, normalize, and de-duplicate added/modified Markdown changes.""" assert self.context is not None - raw_paths = self.context.get("modified_paths") or [] - if not isinstance(raw_paths, list): - raw_paths = [raw_paths] + raw_changes = self.context.get("changes") or [] + if not isinstance(raw_changes, list): + raise ValueError("AutoTagStep requires changes: list[dict]") workspace = Path(self.file_store.workspace_path or ".").resolve() - eligible: list[str] = [] - ignored: list[str] = [] - seen: set[str] = set() - for raw_path in raw_paths: - raw = str(raw_path or "").strip() - target, err = resolve_path(workspace, raw) - if err or target is None or not target.is_file() or target.suffix.lower() != ".md": - ignored.append(raw) + targets: dict[str, _TagTarget] = {} + ignored: list[dict[str, str]] = [] + for item in raw_changes: + if not isinstance(item, dict): + ignored.append({"path": "", "reason": "change must be an object"}) continue - path = normalize_posix_path(display_path(workspace, target)) - if path in seen: - continue - seen.add(path) - eligible.append(path) - return eligible, ignored - async def _process_path(self, path: str, tag_key: str) -> str: + change = str(item.get("change") or "").strip().lower() + raw_path = str(item.get("path") or "").strip() + if change not in _SUPPORTED_CHANGES: + ignored.append({"path": raw_path, "reason": f"unsupported change: {change or 'missing'}"}) + continue + + target, error = resolve_path(workspace, raw_path) + if error or target is None: + ignored.append({"path": raw_path, "reason": error or "invalid path"}) + continue + if not target.is_file(): + ignored.append({"path": raw_path, "reason": "not a file"}) + continue + if target.suffix.lower() != ".md": + ignored.append({"path": raw_path, "reason": "not a Markdown file"}) + continue + + path = normalize_posix_path(display_path(workspace, target)) + previous = targets.get(path) + was_added = previous is not None and previous.change == "added" + normalized_change: Literal["added", "modified"] = "added" if change == "added" or was_added else "modified" + targets[path] = _TagTarget(change=normalized_change, path=path) + return list(targets.values()), ignored + + async def _process_target(self, target: _TagTarget, tag_key: str) -> str: result = await self.agent_wrapper.reply( - self.prompt_format("user_message", path=path, tags_key=tag_key), - system_prompt=self.prompt_format("system_prompt", tags_key=tag_key), + self.prompt_format("user_message", path=target.path, change=target.change, tag_key=tag_key), + system_prompt=self.prompt_format("system_prompt", tag_key=tag_key), job_tools=self.tools, injected_job_kwargs={ - "_allowed_paths": [path], + "_allowed_paths": [target.path], "_allowed_frontmatter_keys": [tag_key], }, ) - target = Path(self.file_store.workspace_path or ".") / path - metadata = dict(frontmatter.loads(target.read_text(encoding="utf-8")).metadata or {}) - normalized = normalize_tags(metadata.get(tag_key)) + path = Path(self.file_store.workspace_path or ".") / target.path + metadata = dict(frontmatter.loads(path.read_text(encoding="utf-8")).metadata or {}) + normalized = normalize_memory_tags(metadata.get(tag_key)) if metadata.get(tag_key) != normalized: response = await self.run_job( "frontmatter_update", - path=path, + path=target.path, metadata={tag_key: normalized}, - _allowed_paths=[path], + _allowed_paths=[target.path], _allowed_frontmatter_keys=[tag_key], ) if not response.success: @@ -104,49 +125,62 @@ class AutoTagStep(BaseStep): prefix = f"{daily_dir}/" if not path.startswith(prefix): return None - remainder = path[len(prefix) :] - parts = remainder.split("/") - if len(parts) != 2: - return None - return parse_daily_date(parts[0]) + parts = path[len(prefix) :].split("/") + return parse_daily_date(parts[0]) if len(parts) == 2 else None async def execute(self): assert self.context is not None initial_success = self.context.response.success - tag_index = getattr(self.file_store, "tag_index", None) - if tag_index is None: + initial_answer = self.context.response.answer + try: + targets, ignored = self._targets() + except ValueError as exc: self.context.response.success = False - self.context.response.answer = "Error: tag index is not configured" + self.context.response.answer = str(exc) return self.context.response - eligible, ignored = self._eligible_paths() - tagged: list[str] = [] - failed: list[dict[str, str]] = [] - summaries: list[str] = [] + results: list[dict] = [] + indexes: list[dict] = [] + if targets and not self.file_store.tag_index_enabled: + self.context.response.success = False + if initial_success: + self.context.response.answer = "Error: tag index is not configured" + return self.context.response + + tag_index = self.file_store.require_tag_index() if targets else None dates: set[str] = set() - for path in eligible: - if day := self._daily_date(path): + for target in targets: + if day := self._daily_date(target.path): dates.add(day) try: - summaries.append(await self._process_path(path, tag_index.tag_key)) - tagged.append(path) + summary = await self._process_target(target, tag_index.tag_key) + results.append( + {"change": target.change, "path": target.path, "success": True, "summary": summary}, + ) except Exception as exc: # pylint: disable=broad-exception-caught - failed.append({"path": path, "error": str(exc)}) - self.logger.warning(f"[{self.name}] failed path={path}: {exc}") + results.append( + {"change": target.change, "path": target.path, "success": False, "error": str(exc)}, + ) + self.logger.warning(f"[{self.name}] failed path={target.path}: {exc}") - indexes = [ - await refresh_day_index(self.file_store, day, self.config_value("daily_dir")) for day in sorted(dates) - ] - self.context.response.success = initial_success and not failed - if failed: - self.context.response.answer = f"Tagged {len(tagged)} file(s); {len(failed)} failed" - elif tagged: - self.context.response.answer = f"Tagged {len(tagged)} file(s)" + for day in sorted(dates): + indexes.append(await refresh_day_index(self.file_store, day, self.config_value("daily_dir"))) + + failed = sum(not item["success"] for item in results) + succeeded = len(results) - failed + self.context.response.success = initial_success and failed == 0 + if initial_success and failed: + self.context.response.answer = f"Tagged {succeeded} file(s); {failed} failed" + elif initial_success and not initial_answer and succeeded: + self.context.response.answer = f"Tagged {succeeded} file(s)" + else: + self.context.response.answer = initial_answer self.context.response.metadata["auto_tag"] = { - "tagged_paths": tagged, - "ignored_paths": ignored, - "failed_paths": failed, + "processed": len(results), + "succeeded": succeeded, + "failed": failed, + "ignored": ignored, + "results": results, "indexes": indexes, - "summaries": [summary for summary in summaries if summary], } return self.context.response diff --git a/reme/steps/evolve/auto_tag.yaml b/reme/steps/evolve/auto_tag.yaml index eecfcd7a..c1f0b299 100644 --- a/reme/steps/evolve/auto_tag.yaml +++ b/reme/steps/evolve/auto_tag.yaml @@ -1,29 +1,44 @@ system_prompt: | - You add retrieval tags to one Markdown memory file at a time. + You maintain entity-oriented memory tags for one Markdown memory file at a time. - Read the complete target file and its frontmatter, then inspect existing tags with `list_tags`. Update only the - `{tags_key}` frontmatter field by calling `frontmatter_update`. Do not change the body or any other frontmatter field. + The `{tag_key}` field answers: "Who or what real-world entity is this memory about?" - Generate 0–8 unique retrieval keywords for the complete note. Always write `{tags_key}`, using `[]` when no useful - tags exist. Each tag must be one string with no whitespace, may contain technical punctuation such as `GPT-5`, `C++`, - `C#`, or `.NET`, must contain at least one letter or digit, and must be at most 64 characters. Store numeric tags as - strings. Prefer an existing tag when it accurately expresses the same concept. + Required workflow: + 1. Use `read` and `frontmatter_read` to inspect the complete target document and its current frontmatter. + 2. Use `list_tags` to inspect tags already used in the workspace. Check additional pages when useful. + 3. Prefer one primary entity that best answers the question. Use two or three tags only when the memory is genuinely + about multiple independent central entities, and never use more than three. Prefer an existing tag whenever it + represents the same entity. Examples include a person (`Sam Altman`), an organization (`宁德时代`), or an + asset/entity (`黄金`). + 4. Use `frontmatter_update` to write the complete `{tag_key}` list, including `[]` when no entity is central. + + Choose entities, not broad topics, activities, attributes, dates, or generic keywords. Keep the canonical real-world + name and use one concise string per entity. Store multiple entities as separate YAML list items, for example + `[宁德时代, 黄金]`; never combine them into one comma-delimited string. Do not create aliases or near-duplicate + spellings for an existing tag. Update only `{tag_key}`; never change the document body or another frontmatter field. system_prompt_zh: | - 你每次为一个 Markdown 记忆文件添加检索 tag。 + 你负责为每一份 Markdown 记忆维护以实体为中心的记忆标签。 - 阅读目标文件的完整内容和 frontmatter，然后通过 `list_tags` 查看已有 tag。只允许调用 `frontmatter_update` - 更新 frontmatter 的 `{tags_key}` 字段，不得修改正文或其他 frontmatter 字段。 + `{tag_key}` 字段回答的问题是：“这份记忆是关于谁或什么现实实体的？” - 根据完整笔记生成 0–8 个不重复检索关键词。必须始终写入 `{tags_key}`，没有合适关键词时使用 `[]`。每个 tag - 必须是一个不含空白字符的字符串，可以包含 `GPT-5`、`C++`、`C#`、`.NET` 等技术标点，必须至少包含一个 - 字母或数字，且不得超过 64 个字符。数字 tag 也保存为字符串。含义相同时优先复用已有 tag。 + 必须遵循以下流程： + 1. 使用 `read` 和 `frontmatter_read` 阅读目标文档的完整内容及当前 frontmatter。 + 2. 使用 `list_tags` 查看工作区已经使用的标签；必要时继续查看后续分页。 + 3. 默认只选择一个最能回答上述问题的主实体。只有当记忆确实同时围绕多个相互独立的核心实体时，才选择 + 两个或三个标签，并且绝不能超过三个。同一实体已有标签时必须优先复用。实体可以是人物（如 + `Sam Altman`）、组织或公司（如 `宁德时代`），也可以是资产等实体（如 `黄金`）。 + 4. 使用 `frontmatter_update` 写入完整的 `{tag_key}` 列表；没有核心实体时也要写入 `[]`。 + + 只选择实体，不要使用宽泛主题、行为、属性、日期或普通关键词。每个实体使用一个简洁、规范的现实名称。 + 多个实体必须保存为独立的 YAML 列表项，例如 `[宁德时代, 黄金]`；不要把它们拼成一个逗号分隔的字符串。 + 不要为已有实体创造别名或近似拼写。只能更新 `{tag_key}`，绝不能修改正文或其他 frontmatter 字段。 user_message: | + Change: {change} Target path: {path} - Tag field: {tags_key} - Read this file, select tags for its complete content, and update only its `{tags_key}` frontmatter field. + Follow the required workflow and update this document's `{tag_key}` field. user_message_zh: | + 变更类型：{change} 目标路径：{path} - Tag 字段：{tags_key} - 阅读该文件，根据完整内容选择 tag，并且只更新 frontmatter 的 `{tags_key}` 字段。 + 按照规定流程更新这份文档的 `{tag_key}` 字段。 diff --git a/reme/steps/evolve/dream/utils.py b/reme/steps/evolve/dream/utils.py index 7701d005..2f41d27d 100644 --- a/reme/steps/evolve/dream/utils.py +++ b/reme/steps/evolve/dream/utils.py @@ -31,8 +31,7 @@ def store_state(step: BaseStep, state: DreamState) -> None: def workspace_dir(step: BaseStep) -> Path: """Get workspace directory.""" - vr = getattr(step.file_store, "workspace_path", None) - return Path(vr).resolve() if vr else Path.cwd().resolve() + return step.file_store.workspace_path.resolve() def daily_dir(step: BaseStep) -> str: diff --git a/reme/steps/index/list_tags.py b/reme/steps/index/list_tags.py index b03f4a3c..aaa172d9 100644 --- a/reme/steps/index/list_tags.py +++ b/reme/steps/index/list_tags.py @@ -10,9 +10,7 @@ class ListTagsStep(BaseStep): async def execute(self): assert self.context is not None - tag_index = getattr(self.file_store, "tag_index", None) - if tag_index is None: - raise RuntimeError("tag index is not configured") + tag_index = self.file_store.require_tag_index() result = await tag_index.list_tags( page=self.context.get("page", 1), diff --git a/reme/steps/index/search.py b/reme/steps/index/search.py index b0b257ff..d30aef3d 100644 --- a/reme/steps/index/search.py +++ b/reme/steps/index/search.py @@ -27,6 +27,13 @@ def _default_limit() -> int: return _DEFAULT_LIMIT +def _filter_values(value: object) -> set: + """Normalize a scalar or collection-valued search filter.""" + if isinstance(value, (list, tuple, set, frozenset)): + return set(value) + return {value} + + @R.register("search_step") class SearchStep(BaseStep): """Hybrid search: run vector + keyword in parallel, fuse via RRF, filter, truncate.""" @@ -138,43 +145,61 @@ class SearchStep(BaseStep): self, raw_tags: object, search_filter: dict, - ) -> tuple[set[str] | None, dict | None, str | None]: - """Build a tag-aware chunk domain, or describe a safe fallback.""" + ) -> tuple[dict, dict | None, str | None]: + """Merge tag-derived paths into the ordinary file-store filter.""" if not raw_tags: - return None, None, None + return search_filter, None, None - tag_index = getattr(self.file_store, "tag_index", None) - tag_index_available = tag_index is not None and bool(getattr(tag_index, "is_healthy", True)) - resolve_chunk_ids = getattr(self.file_store, "resolve_filtered_chunk_ids", None) - filtered_search_available = callable(resolve_chunk_ids) and all( - callable(getattr(self.file_store, method, None)) - for method in ("filtered_vector_search", "filtered_keyword_search") - ) - if not tag_index_available or not filtered_search_available: - reason = "tag_index_unavailable" if not tag_index_available else "file_store_unsupported" - self.logger.warning(f"[{self.name}] tags requested but {reason}; falling back to unfiltered search") - return None, {"requested": True, "applied": False, "reason": reason}, None + if not self.file_store.tag_index_enabled: + self.logger.warning( + f"[{self.name}] tags requested but tag_index_unavailable", + ) + return ( + search_filter, + {"requested": True, "applied": False, "reason": "tag_index_unavailable"}, + "Error: tag index unavailable", + ) + tag_index = self.file_store.require_tag_index() + if not tag_index.is_healthy: + self.logger.warning(f"[{self.name}] tags requested but tag_index_unavailable") + return ( + search_filter, + {"requested": True, "applied": False, "reason": "tag_index_unavailable"}, + "Error: tag index unavailable", + ) normalized_tags = tag_index.normalize_query_tags(raw_tags) if not normalized_tags: - return None, None, "Error: tags contained no valid values" + return search_filter, None, "Error: tags contained no valid values" allowed_paths = set(await tag_index.paths_for_tags(normalized_tags, match_all=False)) - if not bool(getattr(tag_index, "is_healthy", True)): + matched_path_count = len(allowed_paths) + if not tag_index.is_healthy: self.logger.warning( - f"[{self.name}] tag index became unhealthy during lookup; falling back to unfiltered search", + f"[{self.name}] tag index became unhealthy during lookup", + ) + return ( + search_filter, + {"requested": True, "applied": False, "reason": "tag_index_unavailable"}, + "Error: tag index unavailable", ) - return None, {"requested": True, "applied": False, "reason": "tag_index_unavailable"}, None - eligible_chunk_ids = resolve_chunk_ids(allowed_paths, search_filter) + 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(_filter_values(search_filter.pop(key))) + if has_exact_path_filter: + allowed_paths.intersection_update(exact_paths) + search_filter["paths"] = sorted(allowed_paths) metadata = { "requested": True, "applied": True, "tags": normalized_tags, - "matched_paths": len(allowed_paths), - "eligible_chunks": len(eligible_chunk_ids), + "matched_paths": matched_path_count, } - return eligible_chunk_ids, metadata, None + return search_filter, metadata, None async def execute(self): assert self.context is not None @@ -251,31 +276,19 @@ class SearchStep(BaseStep): if strict_date_filter: search_filter["strict_date_filter"] = True - eligible_chunk_ids, tag_filter_metadata, tag_error = await self._resolve_tag_filter(raw_tags, search_filter) + search_filter, tag_filter_metadata, tag_error = await self._resolve_tag_filter(raw_tags, search_filter) if tag_error is not None: self.context.response.success = False self.context.response.answer = tag_error + if tag_filter_metadata is not None: + self.context.response.metadata["tag_filter"] = tag_filter_metadata return self.context.response text_weight = 1.0 - vector_weight use_vector = vector_weight > 0.0 use_keyword = text_weight > 0.0 - if eligible_chunk_ids is not None: - if not eligible_chunk_ids: - vector_results, keyword_results = [], [] - elif use_vector and use_keyword: - vector_results, keyword_results = await asyncio.gather( - self.file_store.filtered_vector_search(query, candidates, eligible_chunk_ids), - self.file_store.filtered_keyword_search(query, candidates, eligible_chunk_ids), - ) - elif use_vector: - vector_results = await self.file_store.filtered_vector_search(query, candidates, eligible_chunk_ids) - keyword_results = [] - else: - vector_results = [] - keyword_results = await self.file_store.filtered_keyword_search(query, candidates, eligible_chunk_ids) - elif use_vector and use_keyword: + if use_vector and use_keyword: vector_results, keyword_results = await asyncio.gather( self.file_store.vector_search(query, candidates, search_filter), self.file_store.keyword_search(query, candidates, search_filter), diff --git a/reme/steps/index/update_changes.py b/reme/steps/index/update_changes.py index 72c7f6c9..6ce461f0 100644 --- a/reme/steps/index/update_changes.py +++ b/reme/steps/index/update_changes.py @@ -351,13 +351,7 @@ class UpdateIndexStep(ChangeApplyStep): return self._estimate_index_memory(size_bytes, len(item[1])) def _estimate_index_memory(self, size_bytes: int, chunk_count: int) -> int: - embedding_bytes = 0 - embedding_store = getattr(self.file_store, "embedding_store", None) - if embedding_store is not None: - try: - embedding_bytes = max(0, int(embedding_store.dimensions)) * self.float16_bytes - except (AttributeError, TypeError, ValueError): - embedding_bytes = 0 + embedding_bytes = self.file_store.embedding_dimensions * self.float16_bytes expanded_content = int(size_bytes * self.batch_memory_expansion_factor) per_chunk = self.chunk_memory_overhead_bytes + embedding_bytes return expanded_content + self.file_memory_overhead_bytes + max(0, chunk_count) * per_chunk diff --git a/tests/unit/test_auto_tag.py b/tests/unit/test_auto_tag.py index 95037205..021b6244 100644 --- a/tests/unit/test_auto_tag.py +++ b/tests/unit/test_auto_tag.py @@ -12,7 +12,7 @@ from reme.components.file_store import LocalFileStore from reme.components.runtime_context import RuntimeContext from reme.components.tag_index import LocalTagIndex from reme.schema import Response -from reme.steps.evolve.auto_tag import AutoTagStep, normalize_tags +from reme.steps.evolve.auto_tag import AutoTagStep, normalize_memory_tags class _TaggingWrapper(BaseAgentWrapper): @@ -21,14 +21,14 @@ class _TaggingWrapper(BaseAgentWrapper): workspace: Path, *, fail_name: str = "", - tag_key: str = "tags", + tag_key: str = "memory_tags", tags: list[object] | None = None, ) -> None: super().__init__(name="tagger") self.workspace = workspace self.fail_name = fail_name self.tag_key = tag_key - self.tags = ["ReMe", "Python"] if tags is None else tags + self.tags = ["宁德时代", "黄金"] if tags is None else tags self.calls: list[tuple[str, dict]] = [] async def reply(self, inputs, **kwargs) -> dict: @@ -59,7 +59,7 @@ async def test_auto_tag_requires_tag_index_before_modifying_files(tmp_path, monk wrapper = _TaggingWrapper(tmp_path) step = AutoTagStep(file_store=store, agent_wrapper=wrapper) - response = await step(RuntimeContext(modified_paths=["daily/2026-09-09/note.md"])) + response = await step(RuntimeContext(changes=[{"change": "added", "path": "daily/2026-09-09/note.md"}])) assert response.success is False assert response.answer == "Error: tag index is not configured" @@ -67,6 +67,44 @@ async def test_auto_tag_requires_tag_index_before_modifying_files(tmp_path, monk assert note.read_bytes() == before +@pytest.mark.asyncio +async def test_auto_tag_with_no_changes_preserves_upstream_response(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + store = LocalFileStore(name="store", embedding_store="", tag_index="") + wrapper = _TaggingWrapper(tmp_path) + step = AutoTagStep(file_store=store, agent_wrapper=wrapper) + context = RuntimeContext(changes=[]) + context.response.answer = "Skipped: no messages" + + response = await step(context) + + assert response.success is True + assert response.answer == "Skipped: no messages" + assert response.metadata["auto_tag"] == { + "processed": 0, + "succeeded": 0, + "failed": 0, + "ignored": [], + "results": [], + "indexes": [], + } + assert not wrapper.calls + + +@pytest.mark.asyncio +async def test_auto_tag_rejects_a_non_list_changes_payload(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + step = AutoTagStep( + file_store=LocalFileStore(name="store", embedding_store="", tag_index=""), + agent_wrapper=_TaggingWrapper(tmp_path), + ) + + response = await step(RuntimeContext(changes="daily/note.md")) + + assert response.success is False + assert response.answer == "AutoTagStep requires changes: list[dict]" + + @pytest.mark.asyncio async def test_auto_tag_filters_paths_and_continues_after_one_file_fails(tmp_path, monkeypatch): monkeypatch.chdir(tmp_path) @@ -82,12 +120,13 @@ async def test_auto_tag_filters_paths_and_continues_after_one_file_fails(tmp_pat wrapper = _TaggingWrapper(tmp_path, fail_name="failed.md") step = AutoTagStep(file_store=store, agent_wrapper=wrapper) context = RuntimeContext( - modified_paths=[ - "daily/2026-09-09/failed.md", - "daily/2026-09-09/notes", - "daily/2026-09-09/plain.txt", - "daily/2026-09-09/first.md", - "daily/2026-09-09/first.md", + changes=[ + {"change": "modified", "path": "daily/2026-09-09/failed.md"}, + {"change": "added", "path": "daily/2026-09-09/notes"}, + {"change": "added", "path": "daily/2026-09-09/plain.txt"}, + {"change": "modified", "path": "daily/2026-09-09/first.md"}, + {"change": "added", "path": "daily/2026-09-09/first.md"}, + {"change": "deleted", "path": "daily/2026-09-09/deleted.md"}, ], ) @@ -97,28 +136,42 @@ async def test_auto_tag_filters_paths_and_continues_after_one_file_fails(tmp_pat assert [call[1]["injected_job_kwargs"] for call in wrapper.calls] == [ { "_allowed_paths": ["daily/2026-09-09/failed.md"], - "_allowed_frontmatter_keys": ["tags"], + "_allowed_frontmatter_keys": ["memory_tags"], }, { "_allowed_paths": ["daily/2026-09-09/first.md"], - "_allowed_frontmatter_keys": ["tags"], + "_allowed_frontmatter_keys": ["memory_tags"], }, ] assert all( call[1]["job_tools"] == ["read", "list_tags", "frontmatter_read", "frontmatter_update"] for call in wrapper.calls ) - assert frontmatter.loads(first.read_text(encoding="utf-8")).metadata["tags"] == ["ReMe", "Python"] - assert "tags" not in frontmatter.loads(failed.read_text(encoding="utf-8")).metadata - assert response.metadata["auto_tag"]["tagged_paths"] == ["daily/2026-09-09/first.md"] - assert response.metadata["auto_tag"]["ignored_paths"] == [ - "daily/2026-09-09/notes", - "daily/2026-09-09/plain.txt", + assert frontmatter.loads(first.read_text(encoding="utf-8")).metadata["memory_tags"] == ["宁德时代", "黄金"] + assert "memory_tags" not in frontmatter.loads(failed.read_text(encoding="utf-8")).metadata + assert response.metadata["auto_tag"]["processed"] == 2 + assert response.metadata["auto_tag"]["succeeded"] == 1 + assert response.metadata["auto_tag"]["failed"] == 1 + assert response.metadata["auto_tag"]["ignored"] == [ + {"path": "daily/2026-09-09/notes", "reason": "not a file"}, + {"path": "daily/2026-09-09/plain.txt", "reason": "not a Markdown file"}, + {"path": "daily/2026-09-09/deleted.md", "reason": "unsupported change: deleted"}, ] - assert response.metadata["auto_tag"]["failed_paths"] == [ - {"path": "daily/2026-09-09/failed.md", "error": "tagging failed"}, + assert response.metadata["auto_tag"]["results"] == [ + { + "change": "modified", + "path": "daily/2026-09-09/failed.md", + "success": False, + "error": "tagging failed", + }, + { + "change": "added", + "path": "daily/2026-09-09/first.md", + "success": True, + "summary": "tagged daily/2026-09-09/first.md", + }, ] - assert "tags: ['ReMe', 'Python']" in (tmp_path / "daily/2026-09-09.md").read_text(encoding="utf-8") + assert "memory_tags: ['宁德时代', '黄金']" in (tmp_path / "daily/2026-09-09.md").read_text(encoding="utf-8") @pytest.mark.asyncio @@ -131,7 +184,7 @@ async def test_auto_tag_uses_configured_key_and_normalizes_agent_output(tmp_path wrapper = _TaggingWrapper( tmp_path, tag_key="keywords", - tags=["ReMe", "reme", "memory system", "++", 100], + tags=["OpenAI", "openai", "Sam Altman", "++", 100, "宁德时代", "黄金"], ) step = AutoTagStep(file_store=store, agent_wrapper=wrapper) @@ -145,28 +198,41 @@ async def test_auto_tag_uses_configured_key_and_normalizes_agent_output(tmp_path monkeypatch.setattr(step, "run_job", update_frontmatter) - response = await step(RuntimeContext(modified_paths=["memory/note.md"])) + context = RuntimeContext(changes=[{"change": "modified", "path": "memory/note.md"}]) + context.response.answer = "Created memory/note.md" + response = await step(context) assert response.success is True - assert frontmatter.loads(note.read_text(encoding="utf-8")).metadata["keywords"] == ["ReMe", "100"] - assert "Tag field: keywords" in wrapper.calls[0][0] + assert response.answer == "Created memory/note.md" + assert frontmatter.loads(note.read_text(encoding="utf-8")).metadata["keywords"] == [ + "OpenAI", + "Sam Altman", + "宁德时代", + ] + assert "Change: modified" in wrapper.calls[0][0] assert "`keywords`" in wrapper.calls[0][1]["system_prompt"] -def test_normalize_tags_enforces_storage_contract(): - assert normalize_tags( +def test_normalize_memory_tags_enforces_entity_storage_contract(): + assert normalize_memory_tags( [ - "GPT-5", - "C++", - "C#", - ".NET", - 100, - "memory system", + "OpenAI", + "openai", + "Sam Altman", "++", - "ReMe", - "reme", - "tag7", - "tag8", - "tag9", + 100, + "宁德时代", + "黄金", ], - ) == ["GPT-5", "C++", "C#", ".NET", "100", "ReMe", "tag7", "tag8"] + ) == ["OpenAI", "Sam Altman", "宁德时代"] + + +def test_auto_tag_prompt_prefers_one_existing_entity_and_yaml_list_storage(): + step = AutoTagStep() + prompt = step.prompt_format("system_prompt", tag_key="memory_tags") + + assert "Prefer one primary entity" in prompt + assert "never use more than three" in prompt + assert "list_tags" in prompt + assert "[宁德时代, 黄金]" in prompt + assert "never combine them into one comma-delimited string" in prompt diff --git a/tests/unit/test_background_steps.py b/tests/unit/test_background_steps.py index 5cc5c2d1..9cdbfe22 100644 --- a/tests/unit/test_background_steps.py +++ b/tests/unit/test_background_steps.py @@ -837,6 +837,7 @@ def test_update_catalog_yields_to_event_loop_while_building_batch(): class _CountingEmbeddingStore: dimensions = 2 max_batch_size = 10 + is_healthy = True def __init__(self): self.calls = 0 @@ -2023,7 +2024,7 @@ def test_auto_memory_reports_modified_for_create_and_false_for_skip(): assert resp.success is True assert resp.metadata["created"] is True assert resp.metadata["modified"] is True - assert context["modified_paths"] == [f"daily/{today}/memory.md"] + assert context["changes"] == [{"change": "added", "path": f"daily/{today}/memory.md"}] assert "tags:" not in (cwd / "daily" / today / "memory.md").read_text(encoding="utf-8") wrapper.on_reply = None @@ -2034,7 +2035,7 @@ def test_auto_memory_reports_modified_for_create_and_false_for_skip(): assert resp.success is True assert resp.metadata["modified"] is False assert resp.metadata["n_messages"] == 0 - assert context["modified_paths"] == [] + assert context["changes"] == [] finally: await fs.close() print("✓ test_auto_memory_reports_modified_for_create_and_false_for_skip passed") diff --git a/tests/unit/test_evolve_utils.py b/tests/unit/test_evolve_utils.py index 8c890b23..b0356ecf 100644 --- a/tests/unit/test_evolve_utils.py +++ b/tests/unit/test_evolve_utils.py @@ -10,7 +10,7 @@ import pytest from reme.steps.evolve._evolve import agent_reply_result_text, format_history from reme.steps.evolve.auto_memory import AutoMemoryStep, _sanitize_msg_for_save -from reme.steps.evolve.auto_tag import normalize_tags +from reme.steps.evolve.auto_tag import normalize_memory_tags def test_agent_reply_result_text_uses_last_text_block(): @@ -77,29 +77,23 @@ def test_sanitize_msg_for_save_drops_tool_results_and_base64_data(): def test_auto_tag_normalizes_frontmatter_tags(): - """Tags keep technical punctuation, reject phrases, de-duplicate, and stop at eight.""" - assert normalize_tags( + """Memory tags preserve entity names, de-duplicate, and stop at three.""" + assert normalize_memory_tags( [ - "GPT-5", - "C++", - "C#", - ".NET", - "100", - 100, - "memory system", + "OpenAI", + "openai", + "Sam Altman", "++", - "ReMe", - "reme", - "tag7", - "tag8", - "tag9", + 100, + "宁德时代", + "黄金", ], - ) == ["GPT-5", "C++", "C#", ".NET", "100", "ReMe", "tag7", "tag8"] + ) == ["OpenAI", "Sam Altman", "宁德时代"] # pylint: disable=use-implicit-booleaness-not-comparison - assert normalize_tags(None) == [] - assert normalize_tags("GPT-5") == [] + assert normalize_memory_tags(None) == [] + assert normalize_memory_tags("OpenAI") == [] # pylint: enable=use-implicit-booleaness-not-comparison - assert normalize_tags(["x" * 65, True, {}, "valid"]) == ["valid"] + assert normalize_memory_tags(["x" * 65, True, {}, "宁德时代"]) == ["宁德时代"] def test_auto_memory_accepts_message_timestamp_aliases(): diff --git a/tests/unit/test_faiss_index_maintenance.py b/tests/unit/test_faiss_index_maintenance.py index 15960f18..ca2a490c 100644 --- a/tests/unit/test_faiss_index_maintenance.py +++ b/tests/unit/test_faiss_index_maintenance.py @@ -36,6 +36,7 @@ class FakeEmbeddingStore: dimensions = 2 max_batch_size = 10 + is_healthy = True def _embed(self, text: str) -> np.ndarray: if "beta" in text or "fresh" in text: diff --git a/tests/unit/test_file_store_consistency.py b/tests/unit/test_file_store_consistency.py index 267026ec..b7750017 100644 --- a/tests/unit/test_file_store_consistency.py +++ b/tests/unit/test_file_store_consistency.py @@ -49,6 +49,7 @@ class FakeEmbeddingStore: dimensions = 2 max_batch_size = 10 + is_healthy = True def _embed(self, text: str) -> np.ndarray: if "beta" in text or "fresh" in text: @@ -1519,9 +1520,16 @@ def test_search_filter_applies_to_vector_and_keyword_results(store_factory): run(go()) +def test_empty_exact_path_filter_matches_nothing(): + """An explicitly empty path domain must not widen into an unfiltered search.""" + candidate = chunk("a", "daily/a.md", "fresh topic") + + assert LocalFileStore._matches_search_filter(candidate, {"paths": []}) is False + + @pytest.mark.parametrize("store_factory", [_new_local_store, _new_zvec_store]) -def test_filtered_search_uses_one_chunk_id_domain_for_vector_and_keyword(store_factory): - """Tag-aware vector and BM25 branches search the same resolved chunk IDs.""" +def test_path_filter_uses_one_domain_for_vector_and_keyword(store_factory): + """Vector and BM25 branches apply the same ordinary path and metadata filters.""" async def go(): with tempfile.TemporaryDirectory() as tmp, temp_chdir(tmp): @@ -1537,14 +1545,12 @@ def test_filtered_search_uses_one_chunk_id_domain_for_vector_and_keyword(store_f ], ) - eligible = store.resolve_filtered_chunk_ids( - {"daily/a.md", "daily/b.md"}, - {"metadata": {"kind": "daily"}}, - ) - - assert eligible == {"a"} - assert [item.id for item in await store.filtered_vector_search("fresh", 5, eligible)] == ["a"] - assert [item.id for item in await store.filtered_keyword_search("fresh", 5, eligible)] == ["a"] + search_filter = { + "paths": ["daily/a.md", "daily/b.md"], + "metadata": {"kind": "daily"}, + } + assert [item.id for item in await store.vector_search("fresh", 5, search_filter)] == ["a"] + assert [item.id for item in await store.keyword_search("fresh", 5, search_filter)] == ["a"] await store.close() run(go()) diff --git a/tests/unit/test_search_step.py b/tests/unit/test_search_step.py index 20786080..ec7a483b 100644 --- a/tests/unit/test_search_step.py +++ b/tests/unit/test_search_step.py @@ -81,43 +81,24 @@ class FakeSearchStore(BaseFileStore): class TaggedFakeSearchStore(FakeSearchStore): - """Fake store exposing the filtered-search extension used by LocalFileStore.""" + """Fake store with a tag index and ordinary file-store filtering.""" def __init__(self, *, chunks: list[FileChunk]): super().__init__(vector_results=chunks, keyword_results=chunks) self.file_chunks = {chunk.id: chunk for chunk in chunks} self.tag_index = LocalTagIndex() - self.filtered_ids: set[str] | None = None - def resolve_filtered_chunk_ids(self, allowed_paths: set[str], search_filter: dict) -> set[str]: - """Resolve paths and the existing filters like LocalFileStore.""" - return { - chunk.id - for chunk in self.file_chunks.values() - if chunk.path in allowed_paths and LocalFileStore._matches_search_filter(chunk, search_filter) - } + async def vector_search(self, query: str, limit: int, search_filter: dict) -> list[FileChunk]: + self.calls.append(("vector", query, limit, search_filter)) + return [chunk for chunk in self.vector_results if LocalFileStore._matches_search_filter(chunk, search_filter)][ + :limit + ] - async def filtered_vector_search( - self, - query: str, - limit: int, - eligible_chunk_ids: set[str], - ) -> list[FileChunk]: - """Return the vector-side members of the resolved domain.""" - self.filtered_ids = set(eligible_chunk_ids) - self.calls.append(("filtered_vector", query, limit, {})) - return [chunk for chunk in self.vector_results if chunk.id in eligible_chunk_ids] - - async def filtered_keyword_search( - self, - query: str, - limit: int, - eligible_chunk_ids: set[str], - ) -> list[FileChunk]: - """Return the keyword-side members of the resolved domain.""" - self.filtered_ids = set(eligible_chunk_ids) - self.calls.append(("filtered_keyword", query, limit, {})) - return [chunk for chunk in self.keyword_results if chunk.id in eligible_chunk_ids] + async def keyword_search(self, query: str, limit: int, search_filter: dict) -> list[FileChunk]: + self.calls.append(("keyword", query, limit, search_filter)) + return [chunk for chunk in self.keyword_results if LocalFileStore._matches_search_filter(chunk, search_filter)][ + :limit + ] def _chunk( @@ -942,8 +923,8 @@ def test_search_step_empty_query_fails_before_store_calls(): asyncio.run(run()) -def test_search_step_missing_tag_index_preserves_legacy_search(): - """A requested tag filter degrades to the unchanged search path without an index.""" +def test_search_step_missing_tag_index_fails_closed(): + """A requested tag filter cannot silently widen search without an index.""" async def run(): hit = _chunk("hit", "daily/a.md", "text", "keyword", 3.0) @@ -952,8 +933,9 @@ def test_search_step_missing_tag_index_preserves_legacy_search(): resp = await step(RuntimeContext(query="hello", limit=5, tags=["python"])) - assert resp.success is True - assert [call[0] for call in store.calls] == ["vector", "keyword"] + assert resp.success is False + assert resp.answer == "Error: tag index unavailable" + assert not store.calls assert resp.metadata["tag_filter"] == { "requested": True, "applied": False, @@ -963,8 +945,8 @@ def test_search_step_missing_tag_index_preserves_legacy_search(): asyncio.run(run()) -def test_search_step_unhealthy_tag_index_preserves_legacy_search(): - """An unhealthy derived tag index is treated like an unavailable index.""" +def test_search_step_unhealthy_tag_index_fails_closed(): + """An unhealthy derived tag index cannot silently widen search.""" async def run(): hit = _chunk("hit", "daily/a.md", "text", "keyword", 3.0) @@ -974,8 +956,9 @@ def test_search_step_unhealthy_tag_index_preserves_legacy_search(): resp = await step(RuntimeContext(query="hello", limit=5, tags=["python"])) - assert resp.success is True - assert {call[0] for call in store.calls} == {"vector", "keyword"} + assert resp.success is False + assert resp.answer == "Error: tag index unavailable" + assert not store.calls assert resp.metadata["tag_filter"]["reason"] == "tag_index_unavailable" asyncio.run(run()) @@ -988,7 +971,7 @@ def test_search_step_rejects_nonempty_tags_that_normalize_to_empty(): store = TaggedFakeSearchStore(chunks=[]) step = SearchStep(file_store=store, expand_links=False) - resp = await step(RuntimeContext(query="hello", limit=5, tags=["has whitespace", "!", "x" * 65])) + resp = await step(RuntimeContext(query="hello", limit=5, tags=["!", "++", "x" * 65])) assert resp.success is False assert resp.answer == "Error: tags contained no valid values" @@ -1007,9 +990,9 @@ def test_search_step_combines_tags_with_existing_chunk_filters(): store = TaggedFakeSearchStore(chunks=[matching, old, wrong_prefix]) await store.tag_index.rebuild( [ - FileNode(path=matching.path, st_mtime=1.0, front_matter={"tags": ["Python"]}), - FileNode(path=old.path, st_mtime=1.0, front_matter={"tags": ["ReMe"]}), - FileNode(path=wrong_prefix.path, st_mtime=1.0, front_matter={"tags": ["python"]}), + FileNode(path=matching.path, st_mtime=1.0, front_matter={"memory_tags": ["Python"]}), + FileNode(path=old.path, st_mtime=1.0, front_matter={"memory_tags": ["ReMe"]}), + FileNode(path=wrong_prefix.path, st_mtime=1.0, front_matter={"memory_tags": ["python"]}), ], ) step = SearchStep(file_store=store, expand_links=False) @@ -1025,16 +1008,36 @@ def test_search_step_combines_tags_with_existing_chunk_filters(): ) assert resp.success is True - assert store.filtered_ids == {"a"} assert [result["id"] for result in resp.metadata["results"]] == ["a"] assert resp.metadata["tag_filter"] == { "requested": True, "applied": True, "tags": ["python", "reme"], "matched_paths": 3, - "eligible_chunks": 1, } - assert {call[0] for call in store.calls} == {"filtered_vector", "filtered_keyword"} + assert {call[0] for call in store.calls} == {"vector", "keyword"} + assert all(set(call[3]["paths"]) == {matching.path, old.path, wrong_prefix.path} for call in store.calls) + + asyncio.run(run()) + + +def test_search_step_tag_filter_with_no_matching_paths_returns_nothing(): + """An empty tag-derived path domain remains restrictive in both search branches.""" + + async def run(): + hit = _chunk("a", "daily/a.md", "match", "keyword", 3.0) + store = TaggedFakeSearchStore(chunks=[hit]) + await store.tag_index.rebuild( + [FileNode(path=hit.path, st_mtime=1.0, front_matter={"memory_tags": ["python"]})], + ) + step = SearchStep(file_store=store, expand_links=False) + + resp = await step(RuntimeContext(query="hello", limit=5, tags=["missing"])) + + assert resp.success is True + assert resp.metadata["results"] == [] + assert resp.metadata["tag_filter"]["matched_paths"] == 0 + assert all(call[3]["paths"] == [] for call in store.calls) asyncio.run(run()) diff --git a/tests/unit/test_tag_index.py b/tests/unit/test_tag_index.py index 6e1ccc41..18aeba15 100644 --- a/tests/unit/test_tag_index.py +++ b/tests/unit/test_tag_index.py @@ -15,7 +15,7 @@ from reme.schema import FileChunk, FileFrontMatter, FileNode from reme.steps.index.list_tags import ListTagsStep -def _node(path: str, tags: object = None, *, key: str = "tags") -> FileNode: +def _node(path: str, tags: object = None, *, key: str = "memory_tags") -> FileNode: metadata = {} if tags is None else {key: tags} return FileNode(path=path, st_mtime=1.0, front_matter=FileFrontMatter(**metadata)) @@ -131,33 +131,33 @@ def test_list_tags_paginates_and_applies_default_sort_orders() -> None: "total_tags": 4, "total_pages": 2, "page": 1, - "range": [1, 2], - "items": [["alpha", 1], ["beta", 2]], + "range": (1, 2), + "items": [("alpha", 1), ("beta", 2)], } assert await index.list_tags(page=2, page_size=2) == { "total_tags": 4, "total_pages": 2, "page": 2, - "range": [3, 4], - "items": [["delta", 1], ["gamma", 1]], + "range": (3, 4), + "items": [("delta", 1), ("gamma", 1)], } assert (await index.list_tags(order_by="file_count"))["items"] == [ - ["beta", 2], - ["alpha", 1], - ["delta", 1], - ["gamma", 1], + ("beta", 2), + ("alpha", 1), + ("delta", 1), + ("gamma", 1), ] assert (await index.list_tags(order_by="file_count", order="asc"))["items"] == [ - ["alpha", 1], - ["delta", 1], - ["gamma", 1], - ["beta", 2], + ("alpha", 1), + ("delta", 1), + ("gamma", 1), + ("beta", 2), ] assert (await index.list_tags(order="desc"))["items"] == [ - ["gamma", 1], - ["delta", 1], - ["beta", 2], - ["alpha", 1], + ("gamma", 1), + ("delta", 1), + ("beta", 2), + ("alpha", 1), ] await index.delete(["daily/c.md"]) @@ -165,17 +165,17 @@ def test_list_tags_paginates_and_applies_default_sort_orders() -> None: assert result == { "total_tags": 3, "total_pages": 2, - "page": 2, - "range": [3, 3], - "items": [["gamma", 1]], + "page": 3, + "range": (0, 0), + "items": [], } await index.clear() assert await index.list_tags(page=9) == { "total_tags": 0, "total_pages": 0, - "page": 1, - "range": [0, 0], + "page": 9, + "range": (0, 0), "items": [], } @@ -217,8 +217,8 @@ def test_list_tags_step_and_tool_schema_expose_compact_result_contract() -> None "total_tags": 1, "total_pages": 1, "page": 1, - "range": [1, 1], - "items": [["reme", 1]], + "range": (1, 1), + "items": [("reme", 1)], } asyncio.run(run()) @@ -229,6 +229,7 @@ def test_list_tags_step_and_tool_schema_expose_compact_result_contract() -> None assert job["parameters"]["properties"]["page_size"]["default"] == 100 assert "[tag, file_count]" in job["description"] assert "range" in job["description"] + assert "empty page" in job["parameters"]["properties"]["page"]["description"] def test_tag_index_reads_configured_frontmatter_key() -> None: @@ -269,14 +270,22 @@ def test_tag_index_allows_model_config_as_frontmatter_key() -> None: assert LocalTagIndex(tag_key="model_config").tag_key == "model_config" +def test_changing_tag_key_invalidates_the_derived_index() -> None: + """Do not serve stale relationships after changing the source field.""" + index = LocalTagIndex() + index.tag_key = "keywords" + + assert index.tag_key == "keywords" + assert not index.is_healthy + + def test_tag_index_rejects_invalid_runtime_frontmatter_key() -> None: """Apply the same validation when a live component key is updated.""" index = LocalTagIndex() with pytest.raises(ValueError, match="tag_key must be a non-empty string"): index.tag_key = "" - - assert index.tag_key == "tags" + assert index.tag_key == "memory_tags" @pytest.mark.parametrize("tag_key", ["name", "description"]) @@ -287,7 +296,7 @@ def test_tag_index_rejects_reserved_runtime_frontmatter_key(tag_key: str) -> Non with pytest.raises(ValueError, match="tag_key must not be a reserved frontmatter key"): index.tag_key = tag_key - assert index.tag_key == "tags" + assert index.tag_key == "memory_tags" def test_file_store_updates_tag_index_from_file_nodes(monkeypatch, tmp_path: Path) -> None: @@ -323,6 +332,20 @@ def test_file_store_updates_tag_index_from_file_nodes(monkeypatch, tmp_path: Pat asyncio.run(run()) +def test_file_store_exposes_tag_index_capability() -> None: + """Expose optional tag-index configuration without backend-specific getattr checks.""" + store = LocalFileStore(name="test", embedding_store="", tag_index="") + + assert store.tag_index_enabled is False + with pytest.raises(RuntimeError, match="tag index is not configured"): + store.require_tag_index() + + index = LocalTagIndex() + store.tag_index = index + assert store.tag_index_enabled is True + assert store.require_tag_index() is index + + def test_tag_failures_do_not_block_other_indexes_and_retry_rebuild(monkeypatch, tmp_path: Path) -> None: """Keep core indexes writable and rebuild tags on the next mutation after a failed recovery.""" @@ -330,7 +353,7 @@ def test_tag_failures_do_not_block_other_indexes_and_retry_rebuild(monkeypatch, monkeypatch.chdir(tmp_path) store = LocalFileStore(name="test", embedding_store="", tag_index="default") await store.start() - assert store.tag_index is not None + assert store.tag_index_enabled original_rebuild = store.tag_index.rebuild async def fail_incremental(_nodes) -> None: @@ -370,7 +393,7 @@ def test_failed_tag_reconciliation_makes_queries_fail_closed(monkeypatch, tmp_pa monkeypatch.chdir(tmp_path) store = LocalFileStore(name="test", embedding_store="", tag_index="default") await store.start() - assert store.tag_index is not None + assert store.tag_index_enabled await store.upsert([(_node("daily/a.md", ["old"]), [])]) async def fail(_items) -> None: @@ -397,7 +420,7 @@ def test_tag_rebuild_graph_read_failure_does_not_block_upsert(monkeypatch, tmp_p monkeypatch.chdir(tmp_path) store = LocalFileStore(name="test", embedding_store="", tag_index="default") await store.start() - assert store.tag_index is not None + assert store.tag_index_enabled assert store.file_graph is not None store._tag_index_rebuild_required = True original_get_nodes = store.file_graph.get_nodes @@ -427,7 +450,7 @@ def test_explicit_reindex_restores_tag_index(monkeypatch, tmp_path: Path) -> Non monkeypatch.chdir(tmp_path) store = LocalFileStore(name="test", embedding_store="", tag_index="default") await store.start() - assert store.tag_index is not None + assert store.tag_index_enabled await store.upsert( [ (_node("daily/a.md", ["ReMe"]), []), @@ -456,7 +479,7 @@ def test_explicit_reindex_uses_updated_tag_key(monkeypatch, tmp_path: Path) -> N monkeypatch.chdir(tmp_path) store = LocalFileStore(name="test", embedding_store="", tag_index="default") await store.start() - assert store.tag_index is not None + assert store.tag_index_enabled await store.upsert([(_node("daily/a.md", ["old"]), [])]) await store.file_graph.upsert_nodes([_node("daily/a.md", ["new"], key="keywords")]) @@ -477,7 +500,7 @@ def test_tag_delete_failures_do_not_block_core_deletion(monkeypatch, tmp_path: P monkeypatch.chdir(tmp_path) store = LocalFileStore(name="test", embedding_store="", tag_index="default") await store.start() - assert store.tag_index is not None + assert store.tag_index_enabled chunk = _chunk("chunk-a", "daily/a.md", "alpha memory") await store.upsert([(_node("daily/a.md", ["alpha"]), [chunk])]) @@ -507,7 +530,7 @@ def test_existing_markdown_chunker_supplies_frontmatter_tags(monkeypatch, tmp_pa monkeypatch.chdir(tmp_path) note = tmp_path / "daily" / "a.md" note.parent.mkdir() - note.write_text("---\ntags: [Python, ReMe]\n---\nbody\n", encoding="utf-8") + note.write_text("---\nmemory_tags: [Python, ReMe]\n---\nbody\n", encoding="utf-8") node, chunks = await MarkdownFileChunker().chunk(note) store = LocalFileStore(name="test", embedding_store="", tag_index="default") @@ -547,7 +570,8 @@ def test_default_config_enables_tag_index_with_explicit_key() -> None: assert config["jobs"]["index_update_loop"]["watch_dirs"] == ["daily_dir", "digest_dir"] assert "tag_index_loop" not in config["jobs"] - assert config["components"]["tag_index"]["default"]["tag_key"] == "tags" + assert config["components"]["tag_index"]["default"]["tag_key"] == "memory_tags" + assert config["components"]["tag_index"]["default"]["max_tags_per_file"] == 3 assert config["components"]["file_store"]["default"]["tag_index"] == "default" assert config["jobs"]["search"]["parameters"]["properties"]["tags"]["default"] == [] assert config["jobs"]["auto_memory"]["steps"] == [ diff --git a/tests/unit/test_zvec_file_store.py b/tests/unit/test_zvec_file_store.py index ba1566f4..037486ac 100644 --- a/tests/unit/test_zvec_file_store.py +++ b/tests/unit/test_zvec_file_store.py @@ -38,6 +38,7 @@ class FakeEmbeddingStore: dimensions = 2 max_batch_size = 10 + is_healthy = True def _embed(self, text: str) -> np.ndarray: if "beta" in text or "fresh" in text: