Merge pull request #2 from jinliyl/review/tag-filter-followups

Refine tag filtering and automatic memory tagging
This commit is contained in:
xyf2020 2026-09-10 14:43:44 +08:00 • committed by GitHub
commit 481c98a42c
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
24 changed files with 629 additions and 544 deletions

3
.gitignore vendored
View file

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

View file

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

View file

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

View file

@ -27,8 +27,6 @@ CachedEmbedding = tuple[str, np.ndarray]
_EMBEDDING_F16_B64_FIELD = "_embedding_f16_b64"
_EMBEDDING_F16_DTYPE = np.dtype("<f2")
_VECTOR_SEARCH_BATCH_SIZE = 1024
_FILTERED_EXACT_RATIO = 0.05
_FILTERED_FULL_RANK_MULTIPLIER = 10
_PROGRESS_LOG_PERCENT_STEP = 10
_KEYWORD_REBUILD_BATCH_SIZE = 200
@ -83,6 +81,12 @@ class LocalFileStore(BaseFileStore):
self._tag_indexed_file_count = 0
self._closing = False
@property
def embedding_dimensions(self) -> 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:

View file

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

View file

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

View file

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

View file

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

View file

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

View file

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

View file

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

View file

@ -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}` 字段。

View file

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

View file

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

View file

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

View file

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

View file

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

View file

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

View file

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

View file

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

View file

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

View file

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

View file

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

View file

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