diff --git a/docs/en/memory_search.md b/docs/en/memory_search.md
index 2560467d..5e634128 100644
--- a/docs/en/memory_search.md
+++ b/docs/en/memory_search.md
@@ -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
 
diff --git a/docs/zh/memory_search.md b/docs/zh/memory_search.md
index 702bd18b..eb53ed7d 100644
--- a/docs/zh/memory_search.md
+++ b/docs/zh/memory_search.md
@@ -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。
 
 ## 怎么搜索
 
diff --git a/reme/components/__init__.py b/reme/components/__init__.py
index eada5ce5..f3b630c6 100644
--- a/reme/components/__init__.py
+++ b/reme/components/__init__.py
@@ -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",
 ]
diff --git a/reme/components/file_store/local_file_store.py b/reme/components/file_store/local_file_store.py
index cab8024f..6075aaee 100644
--- a/reme/components/file_store/local_file_store.py
+++ b/reme/components/file_store/local_file_store.py
@@ -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 ---------------------------------------------------------------
diff --git a/reme/components/tag_index/__init__.py b/reme/components/tag_index/__init__.py
new file mode 100644
index 00000000..d32ef0da
--- /dev/null
+++ b/reme/components/tag_index/__init__.py
@@ -0,0 +1,6 @@
+"""File-level tag index components."""
+
+from .base_tag_index import BaseTagIndex
+from .local_tag_index import LocalTagIndex
+
+__all__ = ["BaseTagIndex", "LocalTagIndex"]
diff --git a/reme/components/tag_index/base_tag_index.py b/reme/components/tag_index/base_tag_index.py
new file mode 100644
index 00000000..01a9a9d8
--- /dev/null
+++ b/reme/components/tag_index/base_tag_index.py
@@ -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."""
diff --git a/reme/components/tag_index/local_tag_index.py b/reme/components/tag_index/local_tag_index.py
new file mode 100644
index 00000000..c3e97780
--- /dev/null
+++ b/reme/components/tag_index/local_tag_index.py
@@ -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()
diff --git a/reme/config/benchmark.yaml b/reme/config/benchmark.yaml
index 643c3a62..4b38238f 100644
--- a/reme/config/benchmark.yaml
+++ b/reme/config/benchmark.yaml
@@ -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
diff --git a/reme/config/default.yaml b/reme/config/default.yaml
index 2565ae79..5daa15af 100644
--- a/reme/config/default.yaml
+++ b/reme/config/default.yaml
@@ -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
diff --git a/reme/enumeration/component_enum.py b/reme/enumeration/component_enum.py
index 04d29e65..22095464 100644
--- a/reme/enumeration/component_enum.py
+++ b/reme/enumeration/component_enum.py
@@ -24,6 +24,8 @@ class ComponentEnum(str, Enum):
 
     KEYWORD_INDEX = "keyword_index"
 
+    TAG_INDEX = "tag_index"
+
     SERVICE = "service"
 
     CLIENT = "client"
diff --git a/reme/steps/evolve/auto_memory.py b/reme/steps/evolve/auto_memory.py
index 977a6736..9961371f 100644
--- a/reme/steps/evolve/auto_memory.py
+++ b/reme/steps/evolve/auto_memory.py
@@ -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")
diff --git a/reme/steps/evolve/auto_memory.yaml b/reme/steps/evolve/auto_memory.yaml
index 0767a502..76df7878 100644
--- a/reme/steps/evolve/auto_memory.yaml
+++ b/reme/steps/evolve/auto_memory.yaml
@@ -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` = 0–8 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` = 针对完整笔记的 0–8 个不重复检索关键词。该字段必须始终存在；没有合适关键词时使用 `[]`。每个 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 0–8 `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] - 根据完整笔记生成 0–8 个 `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 0–8 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 0–8 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]    同时根据合并后的完整笔记重新生成 0–8 个 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] - 根据完整正文生成 0–8 个 tags，并始终包含 `metadata={{"tags": [...]}}`；没有合适关键词时使用 `[]`。
 
   ## 步骤 4 — 总结
 
diff --git a/reme/steps/index/reindex.py b/reme/steps/index/reindex.py
index 1f003a2b..6ba44655 100644
--- a/reme/steps/index/reindex.py
+++ b/reme/steps/index/reindex.py
@@ -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
diff --git a/tests/unit/test_background_steps.py b/tests/unit/test_background_steps.py
index 4ac42542..6de14302 100644
--- a/tests/unit/test_background_steps.py
+++ b/tests/unit/test_background_steps.py
@@ -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."""
 
diff --git a/tests/unit/test_evolve_utils.py b/tests/unit/test_evolve_utils.py
index 85d5ab69..b8e3312c 100644
--- a/tests/unit/test_evolve_utils.py
+++ b/tests/unit/test_evolve_utils.py
@@ -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(
diff --git a/tests/unit/test_file_store_consistency.py b/tests/unit/test_file_store_consistency.py
index 26f2dce6..a2693ed7 100644
--- a/tests/unit/test_file_store_consistency.py
+++ b/tests/unit/test_file_store_consistency.py
@@ -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())
diff --git a/tests/unit/test_injected_job_kwargs.py b/tests/unit/test_injected_job_kwargs.py
index d92482c5..678ae0e7 100644
--- a/tests/unit/test_injected_job_kwargs.py
+++ b/tests/unit/test_injected_job_kwargs.py
@@ -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"])
diff --git a/tests/unit/test_reindex.py b/tests/unit/test_reindex.py
index 2dc428ba..64bd1a43 100644
--- a/tests/unit/test_reindex.py
+++ b/tests/unit/test_reindex.py
@@ -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"]
diff --git a/tests/unit/test_tag_index.py b/tests/unit/test_tag_index.py
new file mode 100644
index 00000000..407bd02b
--- /dev/null
+++ b/tests/unit/test_tag_index.py
@@ -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