feat(Tag filter): Add an optional rebuildable tag index for file frontmatter (#517)
Some checks are pending
CI / Documentation / Test and build documentation (push) Waiting to run
CI / Python packages / Build and verify distributions (push) Waiting to run
CI / Python quality / Pre-commit (push) Waiting to run
CI / Python tests / Unit Tests - py3.11 (push) Waiting to run
CI / Python tests / Unit Tests - py3.12 (push) Waiting to run
CI / Python tests / Unit Tests - py3.13 (push) Waiting to run
CI / ReMe Studio / Studio checks (push) Waiting to run
CI / Windows / CLI smoke - py3.11 (push) Waiting to run
Deploy / Documentation / Build documentation (push) Waiting to run
Deploy / Documentation / deploy (push) Blocked by required conditions
Security / CodeQL / Analyze javascript-typescript (push) Waiting to run
Security / CodeQL / Analyze python (push) Waiting to run

* Add optional tag generation and normalization to auto memory

* Add tag index components and clean up temporary JSONL files

* Preserve tag index state when reconciliation fails

* Refactor and streamline application implementation

* Fix pylint C1803 warnings in tag normalization tests

* Document optional tag index configuration

* Make tag index failures non-blocking and disable auto-memory tags

* fix(tag-index): fail closed and support reindexing

* fix(tag-index): preserve complete query expressions

---------

Co-authored-by: jinli.yl <jinli.yl@alibaba-inc.com>
This commit is contained in:
xyf2020 2026-09-07 15:52:17 +08:00 committed by GitHub
parent 36e3a87c75
commit f04eedb3ab
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
19 changed files with 922 additions and 44 deletions

View file

@ -121,8 +121,9 @@ automatically.
Embedded integrations that have already verified a provider can call `resume_embedding(verified=True)` to repair
missing vectors in the same vector space. Vector-space changes must use the explicit `reindex` job with
`scope: embedding`; vector search remains unavailable until that job finishes successfully.
Use `scope: bm25` to rebuild only keyword search. `scope: all` runs the BM25 rebuild first and then the embedding
rebuild; all scopes use the current `file_chunks` snapshot.
Use `scope: bm25` to rebuild only keyword search, or `scope: tag` to rebuild the optional tag index from the current
file graph. `scope: all` rebuilds BM25 first, then embeddings, and finally tags. BM25 and embedding rebuilds use the
current `file_chunks` snapshot; the tag rebuild uses `FileNode` frontmatter from the file graph.
## How to Search

View file

@ -110,8 +110,9 @@ Embedding store 可通过 `health_check_timeout` 配置启动探测。临时失
已经完成真实服务验证的嵌入式集成可以调用 `resume_embedding(verified=True)`,修复同一向量空间内缺失的向量。
切换 Embedding 向量空间必须显式运行 `reindex` Job并传入 `scope: embedding`;该 Job 成功完成前向量搜索保持不可用。
`scope: bm25` 只重建关键词索引;`scope: all` 先重建 BM25再重建 Embedding。所有 scope 都使用当前的
`file_chunks` 快照。
`scope: bm25` 只重建关键词索引;`scope: tag` 从当前文件图重建可选的标签索引;`scope: all` 依次重建
BM25、Embedding 和标签索引。BM25 和 Embedding 使用当前的 `file_chunks` 快照,标签索引使用文件图中
`FileNode` 的 frontmatter。
## 怎么搜索

View file

@ -13,6 +13,7 @@ from . import job
from . import keyword_index
from . import outbound_proxy
from . import service
from . import tag_index
from . import tokenizer
from .application_context import ApplicationContext
from .base_component import BaseComponent, ComponentMixin
@ -43,5 +44,6 @@ __all__ = [
"keyword_index",
"outbound_proxy",
"service",
"tag_index",
"tokenizer",
]

View file

@ -16,6 +16,7 @@ from ..component_registry import R
from ..embedding_store import BaseEmbeddingStore
from ..file_graph import BaseFileGraph
from ..keyword_index import BaseKeywordIndex
from ..tag_index import BaseTagIndex
from ...enumeration import LinkScopeEnum
from ...schema import FileChunk, FileLink, FileNode
from ...utils import batch_cosine_similarity
@ -34,10 +35,10 @@ _KEYWORD_REBUILD_BATCH_SIZE = 200
class LocalFileStore(BaseFileStore):
"""In-memory file store with deferred JSONL persistence.
Composes three subcomponents: ``embedding_store`` for vector retrieval,
``keyword_index`` for full-text retrieval, and ``file_graph`` for node / link
storage. ``file_graph`` is mandatory; at least one of embedding / keyword
must be present.
Composes ``embedding_store`` for vector retrieval, ``keyword_index`` for
full-text retrieval, ``file_graph`` for node/link storage, and an optional
``tag_index`` derived from file-node frontmatter. ``file_graph`` is mandatory;
at least one of embedding / keyword must be present.
"""
def __init__(
@ -45,6 +46,7 @@ class LocalFileStore(BaseFileStore):
embedding_store: str = "default",
keyword_index: str = "default",
file_graph: str = "default",
tag_index: str = "",
encoding: str = "utf-8",
store_version: str = "v1",
embedding_rebuild_required: bool = False,
@ -54,6 +56,7 @@ class LocalFileStore(BaseFileStore):
from ..embedding_store import LocalEmbeddingStore
from ..file_graph import LocalFileGraph
from ..keyword_index import BM25Index
from ..tag_index import LocalTagIndex
if not embedding_store and not keyword_index:
raise ValueError("At least one of embedding_store or keyword_index must be set.")
@ -63,6 +66,7 @@ class LocalFileStore(BaseFileStore):
self.embedding_store = self.bind(embedding_store, BaseEmbeddingStore, default_factory=LocalEmbeddingStore)
self.keyword_index = self.bind(keyword_index, BaseKeywordIndex, default_factory=BM25Index)
self.file_graph = self.bind(file_graph, BaseFileGraph, default_factory=LocalFileGraph)
self.tag_index = self.bind(tag_index, BaseTagIndex, default_factory=LocalTagIndex)
self.encoding = encoding
self.store_version = store_version
@ -73,6 +77,8 @@ class LocalFileStore(BaseFileStore):
self._embedding_rebuild_pending = bool(embedding_rebuild_required)
self._embedding_space_generation = 0
self._mutation_generation = 0
self._tag_index_rebuild_required = False
self._tag_indexed_file_count = 0
self._closing = False
# -- lifecycle ------------------------------------------------------------
@ -239,6 +245,14 @@ class LocalFileStore(BaseFileStore):
f"elapsed={time.monotonic() - graph_repair_started_at:.3f}s",
)
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"healthy={tag_synced}, "
f"elapsed={time.monotonic() - tag_sync_started_at:.3f}s",
)
keyword_sync_started_at = time.monotonic()
await self._sync_keyword_index_from_chunks()
keyword_backend = type(self.keyword_index).__name__ if self.keyword_index is not None else "disabled"
@ -352,17 +366,20 @@ class LocalFileStore(BaseFileStore):
@BaseFileStore.serialized
async def reindex(self, scope: str) -> dict:
"""Rebuild derived search indexes from ``file_chunks`` without touching files or the graph."""
if scope not in {"all", "bm25", "embedding"}:
raise ValueError("reindex scope must be one of: all, bm25, embedding")
"""Rebuild derived search indexes without rescanning workspace files."""
if scope not in {"all", "bm25", "embedding", "tag"}:
raise ValueError("reindex scope must be one of: all, bm25, embedding, tag")
if scope == "bm25":
return await self._reindex_bm25()
if scope == "embedding":
return await self._reindex_embedding()
if scope == "tag":
return await self._reindex_tag()
return {
"scope": "all",
"bm25": await self._reindex_bm25(),
"embedding": await self._reindex_embedding(),
"tag": await self._reindex_tag(),
}
async def _reindex_bm25(self) -> dict:
@ -617,6 +634,76 @@ class LocalFileStore(BaseFileStore):
elapsed = time.monotonic() - started_at
self.logger.info(f"{self.name}: keyword index rebuild complete: total={total}, elapsed={elapsed:.2f}s")
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:
self._tag_index_rebuild_required = False
self._tag_indexed_file_count = 0
return True
try:
nodes = await self.file_graph.get_nodes()
await self.tag_index.rebuild(nodes)
except Exception:
self._tag_index_rebuild_required = True
self.logger.exception(
f"{self.name}: tag index rebuild failed during {reason}; keeping file store available",
)
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
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)
try:
await self.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)
async def _reindex_tag(self) -> dict:
"""Synchronously rebuild the optional tag index from the authoritative file graph."""
if not await self._rebuild_tag_index("explicit reindex"):
raise RuntimeError("tag index reindex failed")
return {"indexed": self._tag_indexed_file_count, "scope": "tag"}
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:
return
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)
except Exception:
self._tag_index_rebuild_required = True
self.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:
return
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)
except Exception:
self._tag_index_rebuild_required = True
self.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")
async def _dump_owned_state(self) -> None:
"""Persist state owned by this store, excluding dependency snapshots."""
try:
@ -671,6 +758,7 @@ class LocalFileStore(BaseFileStore):
new_nodes, needs_embed, keyword_docs = self._stage_upsert(files, old_map)
await self.file_graph.upsert_nodes(new_nodes)
await self._upsert_tag_nodes(new_nodes)
await self._embed_pending(needs_embed)
if self.keyword_index and old_chunk_ids:
await self.keyword_index.delete_docs(list(old_chunk_ids))
@ -783,6 +871,7 @@ class LocalFileStore(BaseFileStore):
for cid in deleted_chunk_ids:
self.file_chunks.pop(cid, None)
await self.file_graph.delete_nodes([str(n.path) for n in nodes])
await self._delete_tag_paths([str(n.path) for n in nodes])
if self.keyword_index and deleted_chunk_ids:
await self.keyword_index.delete_docs(deleted_chunk_ids)
@ -814,6 +903,17 @@ class LocalFileStore(BaseFileStore):
if self.keyword_index:
await self.keyword_index.clear()
await self.file_graph.clear()
if self.tag_index is not None:
try:
await self.tag_index.clear()
self._tag_index_rebuild_required = False
self._tag_indexed_file_count = 0
self.tag_index.set_healthy(True)
except Exception:
self._tag_index_rebuild_required = True
self.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 ---------------------------------------------------------------

View file

@ -0,0 +1,6 @@
"""File-level tag index components."""
from .base_tag_index import BaseTagIndex
from .local_tag_index import LocalTagIndex
__all__ = ["BaseTagIndex", "LocalTagIndex"]

View file

@ -0,0 +1,54 @@
"""Abstract interface for file-level tag indexes derived from graph nodes."""
from abc import abstractmethod
from ..base_component import BaseComponent
from ...enumeration import ComponentEnum
from ...schema import FileNode
class BaseTagIndex(BaseComponent):
"""A rebuildable index of normalized ``FileNode`` frontmatter tags."""
component_type = ComponentEnum.TAG_INDEX
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.is_healthy = True
def set_healthy(self, healthy: bool) -> None:
"""Mark whether lookups can safely use the current derived state."""
self.is_healthy = healthy
@property
@abstractmethod
def n_files(self) -> int:
"""Return the number of files that currently have indexed tags."""
@abstractmethod
def normalize_tags(self, value: object) -> list[str]:
"""Return canonical tags according to this index's configured limits."""
@abstractmethod
async def rebuild(self, nodes: list[FileNode]) -> None:
"""Replace the complete index with relationships derived from ``nodes``."""
@abstractmethod
async def upsert_nodes(self, nodes: list[FileNode]) -> None:
"""Insert or replace relationships derived from ``nodes``."""
@abstractmethod
async def delete(self, paths: list[str]) -> None:
"""Delete relationships by workspace-relative path."""
@abstractmethod
async def paths_for_tags(self, tags: object, *, match_all: bool = True) -> list[str]:
"""Return sorted paths matching all or any normalized tags."""
@abstractmethod
async def tags_for_path(self, path: str) -> list[str]:
"""Return normalized tags for one workspace-relative path."""
@abstractmethod
async def clear(self) -> None:
"""Clear memory and persisted state."""

View file

@ -0,0 +1,152 @@
"""In-memory tag index derived from ``FileNode.front_matter``."""
import asyncio
from pathlib import PurePosixPath
from .base_tag_index import BaseTagIndex
from ..component_registry import R
from ...schema import FileNode
@R.register("local")
class LocalTagIndex(BaseTagIndex):
"""Maintain bidirectional path/tag relationships without separate source I/O."""
def __init__(self, max_tags_per_file: int = 8, max_tag_length: int = 64, **kwargs):
super().__init__(**kwargs)
self.max_tags_per_file = self._positive_int("max_tags_per_file", max_tags_per_file)
self.max_tag_length = self._positive_int("max_tag_length", max_tag_length)
self.path_to_tags: dict[str, tuple[str, ...]] = {}
self.tag_to_paths: dict[str, set[str]] = {}
self._maintenance_lock = asyncio.Lock()
@property
def n_files(self) -> int:
return len(self.path_to_tags)
@staticmethod
def _positive_int(name: str, value: object) -> int:
if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
raise ValueError(f"{name} must be a positive integer")
return value
def _normalize_tags(self, value: object, *, limit: int | None) -> list[str]:
"""Normalize a strict tag list, optionally limiting the result count."""
if not isinstance(value, list):
return []
result: list[str] = []
seen: set[str] = set()
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):
continue
if not any(char.isalnum() for char in raw):
continue
canonical = raw.casefold()
if canonical in seen:
continue
seen.add(canonical)
result.append(canonical)
if limit is not None and len(result) >= limit:
break
return result
def normalize_tags(self, value: object) -> list[str]:
"""Normalize frontmatter tags according to the per-file count limit."""
return self._normalize_tags(value, limit=self.max_tags_per_file)
@staticmethod
def _validate_path(path: str) -> str:
if not isinstance(path, str) or not path or "\\" in path:
raise ValueError(f"Invalid workspace-relative tag-index path: {path!r}")
pure = PurePosixPath(path)
if pure.is_absolute() or path != pure.as_posix() or any(part in ("", ".", "..") for part in pure.parts):
raise ValueError(f"Invalid workspace-relative tag-index path: {path!r}")
return path
def _prepare_nodes(self, nodes: list[FileNode]) -> list[tuple[str, tuple[str, ...]]]:
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("tags"))
prepared.append((path, tuple(tags)))
return prepared
@staticmethod
def _replace(
path_to_tags: dict[str, tuple[str, ...]],
tag_to_paths: dict[str, set[str]],
path: str,
tags: tuple[str, ...],
) -> None:
old_tags = path_to_tags.pop(path, ())
for tag in old_tags:
paths = tag_to_paths[tag]
paths.discard(path)
if not paths:
del tag_to_paths[tag]
if not tags:
return
path_to_tags[path] = tags
for tag in tags:
tag_to_paths.setdefault(tag, set()).add(path)
async def rebuild(self, nodes: list[FileNode]) -> None:
prepared = self._prepare_nodes(nodes)
path_to_tags: dict[str, tuple[str, ...]] = {}
tag_to_paths: dict[str, set[str]] = {}
for path, tags in prepared:
self._replace(path_to_tags, tag_to_paths, path, tags)
async with self._maintenance_lock:
self.path_to_tags = path_to_tags
self.tag_to_paths = tag_to_paths
self.is_healthy = True
self.logger.info(f"Rebuilt tag index: files={len(path_to_tags)}, tags={len(tag_to_paths)}")
async def upsert_nodes(self, nodes: list[FileNode]) -> None:
prepared = self._prepare_nodes(nodes)
if not prepared:
return
async with self._maintenance_lock:
for path, tags in prepared:
self._replace(self.path_to_tags, self.tag_to_paths, path, tags)
async def delete(self, paths: list[str]) -> None:
validated = [self._validate_path(path) for path in paths]
if not validated:
return
async with self._maintenance_lock:
for path in validated:
self._replace(self.path_to_tags, self.tag_to_paths, path, ())
async def paths_for_tags(self, tags: object, *, match_all: bool = True) -> list[str]:
if not self.is_healthy:
return []
# ``max_tags_per_file`` constrains indexed documents, not lookup
# expressions. Truncating here would silently weaken AND queries and
# omit valid matches from OR queries.
normalized = self._normalize_tags(tags, limit=None)
if not normalized:
return []
async with self._maintenance_lock:
postings = [self.tag_to_paths.get(tag, set()) for tag in normalized]
matches = set.intersection(*postings) if match_all else set.union(*postings)
return sorted(matches)
async def tags_for_path(self, path: str) -> list[str]:
if not self.is_healthy:
return []
path = self._validate_path(path)
async with self._maintenance_lock:
return list(self.path_to_tags.get(path, ()))
async def clear(self) -> None:
async with self._maintenance_lock:
self.path_to_tags = {}
self.tag_to_paths = {}
self.is_healthy = True
async def _close(self) -> None:
await self.clear()

View file

@ -108,13 +108,13 @@ jobs:
# ── Reindex (derived search indexes only) ──
reindex:
backend: base
description: "rebuild BM25 and/or embedding indexes from current file_chunks"
description: "rebuild BM25, embedding, and/or tag indexes without rescanning workspace files"
parameters:
type: object
properties:
scope:
type: string
enum: [all, bm25, embedding]
enum: [all, bm25, embedding, tag]
default: all
steps:
- backend: reindex_step

View file

@ -152,6 +152,7 @@ jobs:
- messages
steps:
- backend: auto_memory_step
enable_tags: false
auto_memory_cc:
backend: base
@ -344,13 +345,13 @@ jobs:
reindex:
backend: base
description: "rebuild BM25 and/or embedding indexes from current file_chunks"
description: "rebuild BM25, embedding, and/or tag indexes without rescanning workspace files"
parameters:
type: object
properties:
scope:
type: string
enum: [all, bm25, embedding]
enum: [all, bm25, embedding, tag]
default: all
steps:
- backend: reindex_step
@ -847,6 +848,12 @@ components:
backend: bm25
tokenizer: default
# tag_index:
# default:
# backend: local
# max_tags_per_file: 8
# max_tag_length: 64
file_store:
default:
backend: local
@ -856,3 +863,4 @@ components:
embedding_store: ""
keyword_index: default
file_graph: default
# tag_index: default

View file

@ -24,6 +24,8 @@ class ComponentEnum(str, Enum):
KEYWORD_INDEX = "keyword_index"
TAG_INDEX = "tag_index"
SERVICE = "service"
CLIENT = "client"

View file

@ -17,6 +17,9 @@ from ...components import R
_SESSION_ID_KEY = "session_id"
_SOURCE_CONVERSATION_KEY = "source_conversation"
_TAGS_KEY = "tags"
_MAX_TAGS = 8
_MAX_TAG_LENGTH = 64
_MESSAGE_TIME_ALIASES = ("time_created", "timestamp", "createdAt", "timeCreated", "created_time")
@ -59,6 +62,39 @@ def _normalize_msg_timestamp(item: dict) -> dict:
return item
def _normalize_tags(value) -> list[str]:
"""Return up to eight unique, retrieval-friendly frontmatter tags.
Tags are stored uniformly as strings. Technical tokens such as ``GPT-5``,
``C++``, ``C#``, and ``.NET`` are valid; whitespace-delimited phrases and
punctuation-only values are not.
"""
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)):
continue
tag = str(item).strip()
if not tag or len(tag) > _MAX_TAG_LENGTH:
continue
if any(char.isspace() for char in tag):
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)
tags.append(tag)
if len(tags) >= _MAX_TAGS:
break
return tags
@R.register("auto_memory_step")
class AutoMemoryStep(BaseStep):
"""Record conversation facts into a daily note via an Agent."""
@ -83,6 +119,10 @@ class AutoMemoryStep(BaseStep):
def _daily_note_path(self, day: str, name: str) -> str:
return f"{self.config_value('daily_dir')}/{day}/{name}.md"
def _tags_enabled(self) -> bool:
"""Whether this step should generate and normalize frontmatter tags."""
return bool(self.kwargs.get("enable_tags", False))
def _frontmatter(self, path: str) -> dict:
post = frontmatter.loads((self.file_store.workspace_path / path).read_text(encoding="utf-8"))
return dict(post.metadata or {})
@ -118,12 +158,14 @@ class AutoMemoryStep(BaseStep):
notes = list_response.metadata.get("notes") or []
return self._find_session_note(notes, session_id)
async def _ensure_session_frontmatter(self, path: str, session_id: str) -> None:
async def _ensure_memory_frontmatter(self, path: str, session_id: str) -> None:
current = self._frontmatter(path)
metadata = {
_SESSION_ID_KEY: session_id,
_SOURCE_CONVERSATION_KEY: self._session_link(session_id),
}
current = self._frontmatter(path)
if self._tags_enabled():
metadata[_TAGS_KEY] = _normalize_tags(current.get(_TAGS_KEY))
if all(current.get(key) == value for key, value in metadata.items()):
return
response = await self.run_job(
@ -339,6 +381,7 @@ class AutoMemoryStep(BaseStep):
template_key = "user_message_create" if created else "user_message_update"
user_message = self.prompt_format(
template_key,
enable_tags=self._tags_enabled(),
today=day,
note=memory_hint or "(none)",
note_path=note_path,
@ -356,7 +399,7 @@ class AutoMemoryStep(BaseStep):
reply_kwargs["injected_job_kwargs"] = {"_allowed_paths": [note_path]}
result = await self.agent_wrapper.reply(
user_message,
system_prompt=self.prompt_format("system_prompt"),
system_prompt=self.prompt_format("system_prompt", enable_tags=self._tags_enabled()),
job_tools=self.create_tools if created else self.update_tools,
**reply_kwargs,
)
@ -382,24 +425,25 @@ 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"])
else:
try:
await self._ensure_session_frontmatter(note_path, session_id)
try:
if not created or self._tags_enabled():
await self._ensure_memory_frontmatter(note_path, session_id)
if not created:
note_path = await self._rename_from_frontmatter_name(note_path, day)
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
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-write failed path={note_path} answer={str(exc)!r}")
return
modified = self._note_modified(before_note_path, before_note_bytes, note_path)
daily_dir = self.config_value("daily_dir")

View file

@ -20,6 +20,7 @@ system_prompt: |
- `name` = a concise, stable topic/event filename stem, such as `cold-remedies` or `project-kickoff-decision`. Do not include today's date or the daily directory date; the outer daily path already records the date. For existing notes, update it when a better filename is clearly warranted.
- `description` = a thorough summary; vague descriptions like "notes" / "misc" are unacceptable.
[enable_tags] - `tags` = 08 unique retrieval keywords for the complete note. Always include the field, using `[]` when no useful tags exist. Each tag must be one string with no whitespace, may contain technical punctuation (for example `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, for example `"100"`.
- **Never set `status`** — it is a field reserved for downstream processing.
system_prompt_zh: |
你是自动记忆系统。你的职责是将最近对话中的核心信息记录到日记记忆中。思考人类会从这段对话中自然地记住什么——不是所有内容,而是真正重要的信息。
@ -43,6 +44,7 @@ system_prompt_zh: |
- `name` = 简洁、稳定的主题/事件文件名 stem例如 `cold-remedies` 或 `project-kickoff-decision`。不要包含今天日期或日记目录日期;外层日记路径已经记录日期。对已有笔记,如果明显有更好的文件名,就更新它。
- `description` = 详细总结;模糊的描述如 "notes" / "misc" 不可接受。
[enable_tags] - `tags` = 针对完整笔记的 08 个不重复检索关键词。该字段必须始终存在;没有合适关键词时使用 `[]`。每个 tag 必须是一个不含空白字符的字符串,可以包含技术名称中的标点(例如 `GPT-5`、`C++`、`C#` 或 `.NET`),必须至少包含一个字母或数字,且不得超过 64 个字符。数字 tag 也保存为字符串,例如 `"100"`。
- **永远不要设置 `status`**——它是下游处理保留的字段。
user_message_create: |
@ -68,11 +70,13 @@ user_message_create: |
Create the note in one shot:
`daily_write name=<name> description=<description> session_id={session_id} date={today} content=<body>`
[enable_tags] Also pass `metadata={{"tags": [<tag>, ...]}}` in the same `daily_write` call.
- Generate `name` as a concise, stable topic/event filename stem for this memory. Prefer a reusable topic or event summary, optionally in kebab-case.
- Do not include today's date or the daily directory date in `name`; the note already lives under today's daily path.
- `name` must be a valid single filename component: no slash, backslash, leading/trailing whitespace, or characters like `< > : " | ? *`.
- `description` must be a thorough summary of the body — specific enough that the description alone conveys all key information.
[enable_tags] - Generate 08 `tags` from the complete note. Always pass `metadata={{"tags": [...]}}`, including an empty list when there are no useful tags. Follow the system prompt's tag format exactly.
## Step 3 — Summary
@ -104,11 +108,13 @@ user_message_create_zh: |
一次性创建笔记:
`daily_write name=<name> description=<description> session_id={session_id} date={today} content=<正文>`
[enable_tags] 在同一次 `daily_write` 调用中另外传入 `metadata={{"tags": [<tag>, ...]}}`。
- 由你生成 `name`,作为这条记忆简洁、稳定的主题/事件文件名 stem。优先使用可复用的主题或事件总结可以采用 kebab-case。
- `name` 不要包含今天日期或日记目录日期;笔记已经位于当天日记路径下。
- `name` 必须是合法的单个文件名组件:不能包含 slash、反斜杠、首尾空白或 `< > : " | ? *` 等字符。
- `description` 必须是正文的详尽总结——具体到仅凭 description 就能传达全部核心信息。
[enable_tags] - 根据完整笔记生成 08 个 `tags`。必须始终传入 `metadata={{"tags": [...]}}`,没有合适关键词时也要传入空数组。严格遵守 system prompt 的 tag 格式。
## 步骤 3 — 总结
@ -155,18 +161,22 @@ user_message_update: |
Execution:
1. Use `edit path={note_path} old=<original fragment> new=<replacement fragment>` for each section that needs updating. You may call `edit` multiple times.
2. After body changes, refresh frontmatter with `frontmatter_update path={note_path} metadata={{"name": "<updated filename stem>", "description": "<updated summary>"}}`.
[enable_tags] Also regenerate 08 tags from the complete merged note and include `"tags": [<tag>, ...]` in the same metadata object. Always include `tags`, using `[]` when none are useful.
- Keep the existing `name` only when it is already the best concise topic/event filename stem. The system will rename the file after your final response.
- Do not add today's date or the daily directory date to `name`.
3. If `edit` fails repeatedly (e.g., cannot find the original text due to formatting mismatch), fall back to `write path={note_path} name=<name> description=<description> content=<full body>` for a complete rewrite.
[enable_tags] In this fallback, also pass `metadata={{"tags": [<tag>, ...]}}`.
## Step 3b — Full Write (Empty File Fallback)
The file exists but its body is empty. Write the full content in one shot:
`write path={note_path} name=<name> description=<description> content=<body>`
[enable_tags] Also pass `metadata={{"tags": [<tag>, ...]}}` in this call.
- Use a concise, stable topic/event `name`; filename changes are applied after your final response.
- Do not include today's date or the daily directory date in `name`.
- `description` must be a thorough summary of the body — specific enough that the description alone conveys all key information.
[enable_tags] - Generate 08 tags from the complete body and always include `metadata={{"tags": [...]}}`, using `[]` when none are useful.
## Step 4 — Summary
@ -214,18 +224,22 @@ user_message_update_zh: |
执行:
1. 对需要更新的每个部分使用 `edit path={note_path} old=<原文片段> new=<替换片段>`。可以多次调用 `edit`。
2. 正文变更后,刷新 frontmatter`frontmatter_update path={note_path} metadata={{"name": "<更新后的文件名 stem>", "description": "<更新后的总结>"}}`。
[enable_tags] 同时根据合并后的完整笔记重新生成 08 个 tags并在同一个 metadata 对象中包含 `"tags": [<tag>, ...]`。必须始终包含 `tags`,没有合适关键词时使用 `[]`。
- 只有当前 `name` 已经是最合适的简洁主题/事件文件名 stem 时才保留。系统会在你最终回复后负责重命名文件。
- 不要在 `name` 中加入今天日期或日记目录日期。
3. 如果 `edit` 多次失败(如因格式不匹配找不到原文),退回 `write path={note_path} name=<name> description=<description> content=<完整正文>` 全量重写。
[enable_tags] 在该 fallback 中还要传入 `metadata={{"tags": [<tag>, ...]}}`。
## 步骤 3b — 全量写入(空文件 fallback
文件存在但正文为空。一次性写入完整内容:
`write path={note_path} name=<name> description=<description> content=<正文>`
[enable_tags] 在该调用中另外传入 `metadata={{"tags": [<tag>, ...]}}`。
- 使用简洁、稳定的主题/事件 `name`;文件名变化会在你最终回复后应用。
- `name` 不要包含今天日期或日记目录日期。
- `description` 必须是正文的详尽总结——具体到仅凭 description 就能传达全部核心信息。
[enable_tags] - 根据完整正文生成 08 个 tags并始终包含 `metadata={{"tags": [...]}}`;没有合适关键词时使用 `[]`。
## 步骤 4 — 总结

View file

@ -6,7 +6,7 @@ from ...components import R
@R.register("reindex_step")
class ReindexStep(BaseStep):
"""Rebuild BM25 and/or embeddings without scanning files or changing the graph."""
"""Rebuild BM25, embeddings, and/or tags without scanning workspace files."""
async def execute(self):
assert self.context is not None

View file

@ -2008,7 +2008,12 @@ def test_auto_memory_reports_modified_for_create_and_false_for_skip():
"---\nname: memory\nsession_id: s1\n"
"source_conversation: '[[session/dialog/s1.jsonl]]'\n---\nbody\n",
)
step = AutoMemoryStep(app_context=app_ctx, file_store=fs, agent_wrapper=wrapper)
step = AutoMemoryStep(
app_context=app_ctx,
file_store=fs,
agent_wrapper=wrapper,
enable_tags=True,
)
resp = await step(
RuntimeContext(
messages=[{"name": "user", "role": "user", "content": "remember project detail"}],
@ -2020,6 +2025,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 "tags: []" in (cwd / "daily" / today / "memory.md").read_text(encoding="utf-8")
wrapper.on_reply = None
resp = await step(RuntimeContext(messages=[], session_id="s2"))
@ -2035,6 +2041,58 @@ def test_auto_memory_reports_modified_for_create_and_false_for_skip():
asyncio.run(run())
def test_auto_memory_normalizes_tags_after_existing_note_update():
"""Existing notes receive a refreshed, normalized tags field capped at eight entries."""
async def run():
with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
cwd = Path.cwd()
app_ctx = _make_app_context(cwd)
fs = LocalFileStore(name="test_store", embedding_store="")
wrapper = _FakeAgentWrapper()
await fs.start()
_install_file_jobs(app_ctx, fs)
try:
today = datetime.datetime.now().strftime("%Y-%m-%d")
note_path = cwd / "daily" / today / "memory.md"
write_file(
note_path,
"---\nname: memory\nsession_id: s1\n"
"source_conversation: '[[session/dialog/s1.jsonl]]'\n"
"tags: [old]\n---\nold body\n",
)
wrapper.on_reply = lambda *_: write_file(
note_path,
"---\nname: memory\nsession_id: s1\n"
"source_conversation: '[[session/dialog/s1.jsonl]]'\n"
"tags: [GPT-5, C++, C#, .NET, 100, 'memory system', '++', ReMe, reme, tag7, tag8, tag9]\n"
"---\nupdated body\n",
)
step = AutoMemoryStep(
app_context=app_ctx,
file_store=fs,
agent_wrapper=wrapper,
enable_tags=True,
)
resp = await step(
RuntimeContext(
messages=[{"name": "user", "role": "user", "content": "updated project detail"}],
session_id="s1",
),
)
resp = resp or step.context.response
assert resp.success is True
metadata = step._frontmatter(f"daily/{today}/memory.md")
assert metadata["tags"] == ["GPT-5", "C++", "C#", ".NET", "100", "ReMe", "tag7", "tag8"]
finally:
await fs.close()
asyncio.run(run())
def test_auto_memory_uses_message_day_for_historical_create():
"""AutoMemoryStep creates historical daily notes from message timestamps."""

View file

@ -9,7 +9,7 @@ from agentscope.message import Msg
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_memory import AutoMemoryStep, _normalize_tags, _sanitize_msg_for_save
def test_agent_reply_result_text_uses_last_text_block():
@ -75,6 +75,32 @@ def test_sanitize_msg_for_save_drops_tool_results_and_base64_data():
assert sanitized.content[1].name == "memory_search"
def test_auto_memory_normalizes_frontmatter_tags():
"""Tags keep technical punctuation, reject phrases, de-duplicate, and stop at eight."""
assert _normalize_tags(
[
"GPT-5",
"C++",
"C#",
".NET",
"100",
100,
"memory system",
"++",
"ReMe",
"reme",
"tag7",
"tag8",
"tag9",
],
) == ["GPT-5", "C++", "C#", ".NET", "100", "ReMe", "tag7", "tag8"]
# pylint: disable=use-implicit-booleaness-not-comparison
assert _normalize_tags(None) == []
assert _normalize_tags("GPT-5") == []
# pylint: enable=use-implicit-booleaness-not-comparison
assert _normalize_tags(["x" * 65, True, {}, "valid"]) == ["valid"]
def test_auto_memory_accepts_message_timestamp_aliases():
"""AutoMemoryStep preserves historical message timestamps from common benchmark fields."""
top_level = AutoMemoryStep._to_msg(

View file

@ -1038,12 +1038,16 @@ def test_all_reindex_composes_bm25_then_embedding_under_one_lock():
async def rebuild_embedding():
return await rebuild("embedding")
async def rebuild_tag():
return await rebuild("tag")
store._reindex_bm25 = rebuild_bm25
store._reindex_embedding = rebuild_embedding
store._reindex_tag = rebuild_tag
result = await store.reindex("all")
assert calls == ["bm25", "embedding"]
assert calls == ["bm25", "embedding", "tag"]
assert result["scope"] == "all"
run(go())

View file

@ -14,6 +14,7 @@ from reme.components.agent_wrapper import AsAgentWrapper, BaseAgentWrapper, CcAg
from reme.components.file_store import LocalFileStore
from reme.schema import Response
from reme.steps.evolve.auto_memory import AutoMemoryStep
from reme.steps.evolve.auto_memory_cc import AutoMemoryCCStep
class _Job:
@ -236,14 +237,60 @@ def test_auto_memory_keeps_original_tool_names():
assert step.update_tools == ["read", "edit", "frontmatter_update", "write"]
def test_auto_memory_tags_are_disabled_by_default_and_enabled_through_kwargs():
assert AutoMemoryStep()._tags_enabled() is False
assert AutoMemoryCCStep()._tags_enabled() is False
assert AutoMemoryStep(enable_tags=True)._tags_enabled() is True
def test_auto_memory_tags_prompt_follows_step_kwarg():
step = AutoMemoryStep()
prompt_kwargs = {
"today": "2026-09-01",
"note": "(none)",
"session_id": "s1",
"history": "user: remember GPT-5",
}
disabled_system = step.prompt_format("system_prompt", enable_tags=step._tags_enabled())
disabled_create = step.prompt_format("user_message_create", enable_tags=step._tags_enabled(), **prompt_kwargs)
disabled_update = step.prompt_format(
"user_message_update",
enable_tags=step._tags_enabled(),
note_path="daily/2026-09-01/memory.md",
**prompt_kwargs,
)
assert "`tags`" not in disabled_system
assert 'metadata={"tags"' not in disabled_create
assert '"tags"' not in disabled_update
enabled_step = AutoMemoryStep(enable_tags=True)
enabled_system = enabled_step.prompt_format("system_prompt", enable_tags=enabled_step._tags_enabled())
enabled_create = enabled_step.prompt_format(
"user_message_create",
enable_tags=enabled_step._tags_enabled(),
**prompt_kwargs,
)
enabled_update = enabled_step.prompt_format(
"user_message_update",
enable_tags=enabled_step._tags_enabled(),
note_path="daily/2026-09-01/memory.md",
**prompt_kwargs,
)
assert "`tags`" in enabled_system
assert 'metadata={"tags"' in enabled_create
assert '"tags"' in enabled_update
def test_auto_memory_create_prompts_match_upstream_date_arguments():
"""Auto-memory prompts keep the upstream model-supplied date argument."""
from pathlib import Path
prompt_files = (Path("reme/steps/evolve/auto_memory.yaml"),)
for prompt_file in prompt_files:
content = prompt_file.read_text(encoding="utf-8")
assert "date={today}" in content or "`date`: {today}" in content or "`date`{today}" in content
prompt_file = Path("reme/steps/evolve/auto_memory.yaml")
evolve_prompt = prompt_file.read_text(encoding="utf-8")
assert "date={today}" in evolve_prompt or "`date`: {today}" in evolve_prompt or "`date`{today}" in evolve_prompt
assert '"tags": [<tag>, ...]' in evolve_prompt
assert '"tags": []' in evolve_prompt or "using `[]`" in evolve_prompt
def test_configs_define_original_jobs_without_daily_variants():
@ -257,6 +304,9 @@ def test_configs_define_original_jobs_without_daily_variants():
for name in ("read_daily", "edit_daily", "write_daily"):
assert name not in jobs, f"{config_name} unexpectedly defines {name}"
default = resolve_app_config(config="default", log_config=False)
assert default["jobs"]["auto_memory"]["steps"][0]["enable_tags"] is False
if __name__ == "__main__":
pytest.main([__file__, "-v"])

View file

@ -6,11 +6,12 @@ import pytest
from reme.components.file_store import LocalFileStore
from reme.components.runtime_context import RuntimeContext
from reme.config import resolve_app_config
from reme.steps.index import ReindexStep
@pytest.mark.asyncio
@pytest.mark.parametrize("scope", ["bm25", "embedding"])
@pytest.mark.parametrize("scope", ["bm25", "embedding", "tag"])
async def test_reindex_step_delegates_scope(scope):
"""The step forwards each individual scope without clearing the store."""
store = LocalFileStore(name=f"test_reindex_{scope}", embedding_store="")
@ -32,6 +33,7 @@ async def test_reindex_step_delegates_all_once():
"scope": "all",
"bm25": {"scope": "bm25", "indexed": 3},
"embedding": {"scope": "embedding", "indexed": 3},
"tag": {"scope": "tag", "indexed": 3},
}
store.reindex = AsyncMock(return_value=details)
@ -39,3 +41,11 @@ async def test_reindex_step_delegates_all_once():
store.reindex.assert_awaited_once_with("all")
assert response.metadata == details
def test_reindex_job_schema_exposes_tag_scope():
"""The public job contract accepts every scope implemented by the file store."""
config = resolve_app_config(config="default", log_config=False)
scope = config["jobs"]["reindex"]["parameters"]["properties"]["scope"]
assert scope["enum"] == ["all", "bm25", "embedding", "tag"]

View file

@ -0,0 +1,346 @@
"""Focused contracts for the FileNode-derived tag index."""
# pylint: disable=protected-access
import asyncio
from pathlib import Path
import pytest
from reme.components.file_chunker import MarkdownFileChunker
from reme.components.file_store import LocalFileStore
from reme.components.tag_index import LocalTagIndex
from reme.config import resolve_app_config
from reme.schema import FileChunk, FileFrontMatter, FileNode
def _node(path: str, tags: object = None) -> FileNode:
metadata = {} if tags is None else {"tags": tags}
return FileNode(path=path, st_mtime=1.0, front_matter=FileFrontMatter(**metadata))
def _chunk(chunk_id: str, path: str, text: str) -> FileChunk:
return FileChunk(id=chunk_id, path=path, text=text, start_line=1, end_line=1)
def test_tag_normalization_and_bidirectional_mutations() -> None:
"""Normalize FileNode tags and keep both lookup directions consistent."""
async def run() -> None:
index = LocalTagIndex(max_tags_per_file=3)
await index.start()
await index.upsert_nodes([_node("daily/a.md", ["Python", "PYTHON", "C++", ".NET", "ignored"])])
assert await index.tags_for_path("daily/a.md") == ["python", "c++", ".net"]
assert await index.paths_for_tags(["PYTHON"]) == ["daily/a.md"]
assert index.tag_to_paths == {
"python": {"daily/a.md"},
"c++": {"daily/a.md"},
".net": {"daily/a.md"},
}
await index.upsert_nodes([_node("daily/a.md", ["ReMe"])])
assert await index.tags_for_path("daily/a.md") == ["reme"]
assert set(index.tag_to_paths) == {"reme"}
await index.delete(["daily/a.md", "daily/missing.md"])
assert await index.tags_for_path("daily/a.md") == []
assert not index.tag_to_paths
await index.close()
asyncio.run(run())
def test_rebuild_is_atomic_and_supports_all_or_any_queries() -> None:
"""Publish complete rebuilds atomically and support intersection and union lookup."""
async def run() -> None:
index = LocalTagIndex()
await index.rebuild(
[
_node("daily/a.md", ["python", "reme"]),
_node("digest/b.md", ["python"]),
_node("digest/untagged.md"),
],
)
assert await index.paths_for_tags(["python", "reme"]) == ["daily/a.md"]
assert await index.paths_for_tags(["python", "reme"], match_all=False) == [
"daily/a.md",
"digest/b.md",
]
assert "digest/untagged.md" not in index.path_to_tags
before_paths = dict(index.path_to_tags)
before_tags = {tag: set(paths) for tag, paths in index.tag_to_paths.items()}
with pytest.raises(ValueError, match="Invalid workspace-relative"):
await index.rebuild([_node("daily/new.md", ["new"]), _node("../escape.md", ["invalid"])])
assert index.path_to_tags == before_paths
assert index.tag_to_paths == before_tags
asyncio.run(run())
def test_queries_are_not_truncated_by_per_file_tag_limit() -> None:
"""Apply the count limit to indexed files without dropping lookup conditions."""
async def run() -> None:
index = LocalTagIndex(max_tags_per_file=2)
await index.rebuild(
[
_node("daily/a.md", ["a", "b"]),
_node("daily/c.md", ["c"]),
],
)
assert await index.paths_for_tags(["a", "b", "c"]) == []
assert await index.paths_for_tags(["a", "b", "c"], match_all=False) == [
"daily/a.md",
"daily/c.md",
]
asyncio.run(run())
def test_file_store_updates_tag_index_from_file_nodes(monkeypatch, tmp_path: Path) -> None:
"""Keep daily and digest tags aligned through file-store mutations."""
async def run() -> None:
monkeypatch.chdir(tmp_path)
store = LocalFileStore(name="test", embedding_store="", tag_index="default")
await store.start()
assert isinstance(store.tag_index, LocalTagIndex)
await store.upsert(
[
(_node("daily/a.md", ["Python"]), []),
(_node("digest/b.md", ["Digest"]), []),
],
)
assert await store.tag_index.paths_for_tags(["python"]) == ["daily/a.md"]
assert await store.tag_index.paths_for_tags(["digest"]) == ["digest/b.md"]
await store.upsert([(_node("daily/a.md", ["ReMe"]), [])])
assert await store.tag_index.paths_for_tags(["python"]) == []
assert await store.tag_index.paths_for_tags(["reme"]) == ["daily/a.md"]
await store.delete("daily/a.md")
assert await store.tag_index.paths_for_tags(["reme"]) == []
await store.clear()
assert store.tag_index.path_to_tags == {}
assert store.tag_index.tag_to_paths == {}
await store.close()
asyncio.run(run())
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."""
async def run() -> None:
monkeypatch.chdir(tmp_path)
store = LocalFileStore(name="test", embedding_store="", tag_index="default")
await store.start()
assert store.tag_index is not None
original_rebuild = store.tag_index.rebuild
async def fail_incremental(_nodes) -> None:
raise RuntimeError("incremental tag failure")
async def fail_rebuild(_nodes) -> None:
raise RuntimeError("tag rebuild failure")
monkeypatch.setattr(store.tag_index, "upsert_nodes", fail_incremental)
monkeypatch.setattr(store.tag_index, "rebuild", fail_rebuild)
first_chunk = _chunk("chunk-a", "daily/a.md", "alpha memory")
await store.upsert([(_node("daily/a.md", ["alpha"]), [first_chunk])])
assert [node.path for node in await store.get_nodes()] == ["daily/a.md"]
assert "chunk-a" in store.file_chunks
assert "chunk-a" in store.keyword_index.document_ids
assert store._tag_index_rebuild_required is True
monkeypatch.setattr(store.tag_index, "rebuild", original_rebuild)
second_chunk = _chunk("chunk-b", "digest/b.md", "beta memory")
await store.upsert([(_node("digest/b.md", ["beta"]), [second_chunk])])
assert store._tag_index_rebuild_required is False
assert await store.tag_index.paths_for_tags(["alpha"]) == ["daily/a.md"]
assert await store.tag_index.paths_for_tags(["beta"]) == ["digest/b.md"]
assert {"chunk-a", "chunk-b"}.issubset(store.keyword_index.document_ids)
await store.close()
asyncio.run(run())
def test_failed_tag_reconciliation_makes_queries_fail_closed(monkeypatch, tmp_path: Path) -> None:
"""Never expose stale tag matches while reconciliation is pending."""
async def run() -> None:
monkeypatch.chdir(tmp_path)
store = LocalFileStore(name="test", embedding_store="", tag_index="default")
await store.start()
assert store.tag_index is not None
await store.upsert([(_node("daily/a.md", ["old"]), [])])
async def fail(_items) -> None:
raise RuntimeError("tag failure")
monkeypatch.setattr(store.tag_index, "upsert_nodes", fail)
monkeypatch.setattr(store.tag_index, "rebuild", fail)
await store.upsert([(_node("daily/a.md", ["new"]), [])])
assert store._tag_index_rebuild_required is True
assert store.tag_index.is_healthy is False
assert await store.tag_index.paths_for_tags(["old"]) == []
assert await store.tag_index.paths_for_tags(["new"]) == []
assert await store.tag_index.tags_for_path("daily/a.md") == []
await store.close()
asyncio.run(run())
def test_tag_rebuild_graph_read_failure_does_not_block_upsert(monkeypatch, tmp_path: Path) -> None:
"""Keep core indexes consistent if the optional tag repair cannot read the graph snapshot."""
async def run() -> None:
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.file_graph is not None
store._tag_index_rebuild_required = True
original_get_nodes = store.file_graph.get_nodes
async def fail_full_snapshot(paths=None):
if paths is None:
raise RuntimeError("graph snapshot failure")
return await original_get_nodes(paths)
monkeypatch.setattr(store.file_graph, "get_nodes", fail_full_snapshot)
chunk = _chunk("chunk-a", "daily/a.md", "alpha memory")
await store.upsert([(_node("daily/a.md", ["alpha"]), [chunk])])
assert "chunk-a" in store.file_chunks
assert "chunk-a" in store.keyword_index.document_ids
assert store._tag_index_rebuild_required is True
assert store.tag_index.is_healthy is False
await store.close()
asyncio.run(run())
def test_explicit_reindex_restores_tag_index(monkeypatch, tmp_path: Path) -> None:
"""The tag scope and all scope rebuild tags from the authoritative graph."""
async def run() -> None:
monkeypatch.chdir(tmp_path)
store = LocalFileStore(name="test", embedding_store="", tag_index="default")
await store.start()
assert store.tag_index is not None
await store.upsert(
[
(_node("daily/a.md", ["ReMe"]), []),
(_node("daily/untagged.md"), []),
],
)
await store.tag_index.clear()
assert await store.tag_index.paths_for_tags(["reme"]) == []
assert await store.reindex("tag") == {"indexed": 1, "scope": "tag"}
assert await store.tag_index.paths_for_tags(["reme"]) == ["daily/a.md"]
await store.tag_index.clear()
result = await store.reindex("all")
assert result["tag"] == {"indexed": 1, "scope": "tag"}
assert await store.tag_index.paths_for_tags(["reme"]) == ["daily/a.md"]
await store.close()
asyncio.run(run())
def test_tag_delete_failures_do_not_block_core_deletion(monkeypatch, tmp_path: Path) -> None:
"""Complete graph, chunk, and keyword deletion when tag deletion and recovery fail."""
async def run() -> None:
monkeypatch.chdir(tmp_path)
store = LocalFileStore(name="test", embedding_store="", tag_index="default")
await store.start()
assert store.tag_index is not None
chunk = _chunk("chunk-a", "daily/a.md", "alpha memory")
await store.upsert([(_node("daily/a.md", ["alpha"]), [chunk])])
async def fail_delete(_paths) -> None:
raise RuntimeError("incremental tag delete failure")
async def fail_rebuild(_nodes) -> None:
raise RuntimeError("tag rebuild failure")
monkeypatch.setattr(store.tag_index, "delete", fail_delete)
monkeypatch.setattr(store.tag_index, "rebuild", fail_rebuild)
await store.delete("daily/a.md")
assert await store.get_nodes() == []
assert "chunk-a" not in store.file_chunks
assert "chunk-a" not in store.keyword_index.document_ids
assert store._tag_index_rebuild_required is True
await store.close()
asyncio.run(run())
def test_existing_markdown_chunker_supplies_frontmatter_tags(monkeypatch, tmp_path: Path) -> None:
"""Use the FileNode produced by the existing chunker without reading frontmatter again."""
async def run() -> None:
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")
node, chunks = await MarkdownFileChunker().chunk(note)
store = LocalFileStore(name="test", embedding_store="", tag_index="default")
await store.start()
await store.upsert([(node, chunks)])
assert await store.tag_index.tags_for_path("daily/a.md") == ["python", "reme"]
await store.close()
asyncio.run(run())
def test_file_store_rebuilds_non_persistent_tag_index_from_graph(monkeypatch, tmp_path: Path) -> None:
"""Restore tag relationships from the persisted file graph on startup."""
async def run() -> None:
monkeypatch.chdir(tmp_path)
first = LocalFileStore(name="test", embedding_store="", tag_index="default")
await first.start()
await first.upsert([(_node("daily/a.md", ["ReMe"]), [])])
await first.close()
assert not list((tmp_path / "metadata").glob("tag_index/**/*"))
restored = LocalFileStore(name="test", embedding_store="", tag_index="default")
await restored.start()
assert await restored.tag_index.paths_for_tags(["reme"]) == ["daily/a.md"]
await restored.close()
asyncio.run(run())
def test_default_config_documents_optional_tag_index_without_enabling_it() -> None:
"""Document tag indexing in the default config without enabling another index or watcher."""
config = resolve_app_config(config="default", log_config=False)
assert config["jobs"]["index_update_loop"]["watch_dirs"] == ["daily_dir", "digest_dir"]
assert "tag_index_loop" not in config["jobs"]
assert "tag_index" not in config["components"]
assert "tag_index" not in config["components"]["file_store"]["default"]
default_yaml = Path("reme/config/default.yaml").read_text(encoding="utf-8")
assert "# tag_index:" in default_yaml
assert "# tag_index: default" in default_yaml