up

docs4/reme_design.md

@@ -1,18 +1,22 @@
 # 基础Job
+
 @jinli
-| 分类     | 能力      | 参数                                                        |
-|--------|---------|-----------------------------------------------------------|
-| 通用     | help    |                                                           |
-| 通用     | start   |   支持后台 |
-| 通用     | restart |                                                           |
-| 通用     | version |                                                           |
-| 通用     | reindex |                                                           |
-| search | search  | query="search term" limit=10 tag="[]" score=0.1 copy=true |
+说明：📥 输入参数 ｜ 📤 输出 ｜ ⭐ 必填 ｜ 🎚️ 默认值 ｜ 🛠️ 内部行为
+
+| 分类        | 能力 (register name)                               | 参数 & 行为                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                         |
+|-----------|--------------------------------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| 🌐 通用     | 🆘 `help` (`help_step`)                          | 📥 无 ｜ 📤 `answer` 一行一个 job：`🛠️ \`{name}\` — {description} 📥 {params}`，参数渲染为 `name:type*`(必填) / `name:type={default}` / `name:type` ｜ 📊 `metadata.job_count` ｜ 🛠️ 自动跳过名为 `help` 的 job                                                                                                                                                                                                                                                                                                                                                                                                                                                                                       |
+| 🌐 通用     | 🩺 `health_check` (`health_check_step`)          | 📥 无 ｜ 📤 `answer = "✅/❌ ReMe v{version} - healthy/unhealthy"` ｜ 📊 `metadata.health = {version, healthy, components}` ｜ 🧩 覆盖组件：`embedding_model`(🟢 is_started/is_healthy/model_name/dimensions/cache_size/memory) · `file_graph`(🕸️ n_nodes/n_edges/n_virtual\|n_pending/memory) · `file_store`(📦 n_chunks/n_chunks_with_embedding/memory) · `file_watcher`(👀 background_running/watch_paths) · `keyword_index`(🔤 n_docs/vocab_size/memory) ｜ 🛠️ deep sizeof（含 numpy.nbytes），未启动 / 后台未跑 / embedding 不健康 → ❌                                                                                                                                                             |
+| 🌐 通用     | 🏷️ `version` (`version_step`)                   | 📥 无 ｜ 📤 `answer = reme4.__version__` ｜ 📊 `metadata.version`                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                  |
+| 🌐 通用     | 🔄 `reindex` (`reindex_step`)                    | 📥 无 ｜ 📤 `answer = "🔄 Reindexed {added} file(s)"` ｜ 📊 `metadata.counts = {added, ...}` ｜ 🛠️ 流程：`file_watcher.close()` → `file_store.clear()` → `file_watcher.update_store()` → `file_watcher.start()`（finally 保证重启）                                                                                                                                                                                                                                                                                                                                                                                                                                                         |
+| 🔎 search | 🔍 `search` (`search_step`)                      | 📥 `query:str` ⭐ ｜ 🎚️ `limit:int=5`(>0) ｜ 🎚️ `min_score:float=0.0` ｜ ⚖️ `vector_weight:float=0.7` ∈[0,1]（keyword 权 = 1-vw）｜ 🔀 `candidate_multiplier:float=3.0`（candidates = min(200, limit×mult)）｜ 🔗 `expand_links:bool=True` ｜ 🔢 `max_links_per_direction:int=10` ｜ 🎚️ `search_filter:dict={}` ｜ 📤 `answer` 每命中一行 `path:start-end [score=… vector=… keyword=…] text` + 缩进的 `→ outlinks (n)` / `← inlinks (n)` + `via predicate=… anchor=#…` ｜ 📊 `metadata.results` / `metadata.link_expansion` / `metadata.counts={vector,keyword,returned,hybrid}` ｜ 🛠️ 并行 `vector_search` + `keyword_search` → RRF 融合（K=60，按 chunk.id 合并）→ `min_score` 过滤 → `limit` 截断 → 邻居 meta 注入 |
+| 🧪 demo   | 🪄 `demo_echo` (`demo_echo_step1` + `step2`)     | 📥 `query:str=""` ｜ 🎚️ `min_score:float=0.5` ｜ 🛠️ step1：`processed_query = query.strip().lower()`，`adjusted_min_score = min_score * 0.9`，写回 context ｜ 📤 step2：`answer = "echo: {processed_query} (min_score={adjusted_min_score})"` ｜ 📊 `metadata = {step, query, min_score, processed_query, adjusted_min_score}`                                                                                                                                                                                                                                                                                                                                                          |
+| 🌊 demo   | 🌊 `stream_demo` (`stream_demo_step1` + `step2`) | 📥 `query:str=""` ｜ 🎚️ `repeat:int=10` ｜ 🎚️ `interval:float=0.1`（秒/字符）｜ 🛠️ step1：`stream_text = query * repeat` 写回 context ｜ 📤 step2：按字符 `add_stream_string(ch, ChunkEnum.CONTENT)` 流式输出，`asyncio.sleep(interval)` 节流                                                                                                                                                                                                                                                                                                                                                                                                                                                       |
 
 @sen
-| tags | stat |  返回特定tag信息 |
-| tags | list |  返回所有tag列表 |
-| crud | upload/download | 其他文件                                                   |
+| tags | stat | 返回特定tag信息 |
+| tags | list | 返回所有tag列表 |
+| crud | upload/download | 其他文件 |
 | file | stat | path |
 | file | list | path |
 | property | property:read | |
@@ -29,7 +33,6 @@
 | crud | delete | path="My Note
 | daily:crud | daily:xxx | 与 crud 参数保持一致 |
 
-
 # 日记类型
 
 | 类型        | 路径                                            | 说明                          |
@@ -46,73 +49,72 @@
 | 主题dream  + 生成链接 @sen    | daily/xxx     | knowledge/xxx                                 | /dream                      | 把 daily 目录的内容按主题聚类合并到 topic 目录, 主动在文档中建立 [[link]] 关联 |
 | 主动proactive     @wangce | daily / topic | proactive_query                               | pre_query                   | 思考 daily / topic 信息，主动决定推送给用户的消息                     |
 
-
-
-
 2. file_parser
-  a. 抽象基类 parse: @jinli
-    ⅰ. 输入是path：相对路径
-    ⅱ. 输出是FileMetadata & list[FileChunks] & list[FileEdge]
-  b. default parser 兼容老方案 @jinli
-    ⅰ. 带overlap的chunking策略 ，不输出FileEdge
-  c. markdown parser @sen
-    ⅰ. 根据markdown ast做chunk，不需要overlap
-    ⅱ. 增加一个索引的chunk chunk_type @锦鲤 file_chunk_type content/index
-    ⅲ. 增加link的正则解析：predicate:: [[path#anchor]]
+   a. 抽象基类 parse: @jinli
+   ⅰ. 输入是path：相对路径
+   ⅱ. 输出是FileMetadata & list[FileChunks] & list[FileEdge]
+   b. default parser 兼容老方案 @jinli
+   ⅰ. 带overlap的chunking策略 ，不输出FileEdge
+   c. markdown parser @sen
+   ⅰ. 根据markdown ast做chunk，不需要overlap
+   ⅱ. 增加一个索引的chunk chunk_type @锦鲤 file_chunk_type content/index
+   ⅲ. 增加link的正则解析：predicate:: [[path#anchor]]
 3. file_store @sen
-  a. 抽象存储：
-    ⅰ. filenode = file + path + st_mtime + metadata + list[FileEdge]
-    ⅱ. graph=dict[str, filenode] 内存+json
-    ⅲ. list[FileChunk] 存db
-  b. 抽象基类
-    ⅰ. graph：fellow dict的操作 update/get/set
-    ⅱ. chunks 	dict[str, list[chunk]]
-      1. delete_chunks_by_path
-      2. update_chunks_by_path
-      3. list_chunks_by_path
-      4. vector_search/keyword_search
-    ⅲ. 手写一个bm25检索
-    ⅳ. 【核心】检索机制 vector bm25 graph 如何进行融合
+   a. 抽象存储：
+   ⅰ. filenode = file + path + st_mtime + metadata + list[FileEdge]
+   ⅱ. graph=dict[str, filenode] 内存+json
+   ⅲ. list[FileChunk] 存db
+   b. 抽象基类
+   ⅰ. graph：fellow dict的操作 update/get/set
+   ⅱ. chunks dict[str, list[chunk]]
+    1. delete_chunks_by_path
+    2. update_chunks_by_path
+    3. list_chunks_by_path
+    4. vector_search/keyword_search
+       ⅲ. 手写一个bm25检索
+       ⅳ. 【核心】检索机制 vector bm25 graph 如何进行融合
 4. file_watcher @jinli
-  a. 抽象基类
-    ⅰ. on_start:
-      1. file_store 的start 在前，加载graph，file_watcher在后，递归扫描目录
-        a. 通过ms_time对比graph，on_change 进行改动
-    ⅱ. on_change:
-      1. 更新/增加:
-        a. delete_chunks_by_path 更新数据库
-        b. upate_chunks_by_path 更新数据库
-        c. 更新graph
-      2. 删除
-        a. delete_chunks_by_path 更新数据库
+   a. 抽象基类
+   ⅰ. on_start:
+    1. file_store 的start 在前，加载graph，file_watcher在后，递归扫描目录
+       a. 通过ms_time对比graph，on_change 进行改动
+       ⅱ. on_change:
+    1. 更新/增加:
+       a. delete_chunks_by_path 更新数据库
+       b. upate_chunks_by_path 更新数据库
+       c. 更新graph
+    2. 删除
+       a. delete_chunks_by_path 更新数据库
 
 MemorySchema
+
 1. markdown文件结构 @sen
-  a. formatter：
-    ⅰ. title
-    ⅱ. desc
-    ⅲ. tags
-    ⅳ.
+   a. formatter：
+   ⅰ. title
+   ⅱ. desc
+   ⅲ. tags
+   ⅳ.
 2. memory文件结构目录
-  a. MEMORY.md
-  b. msg/files -> daily/YYYYMMDD/YYYYMMDD.md + xxxx.md
-    ⅰ. YYYYMMDD.md
-      1. xxx -> xxxx.md
-      2. xxx -> xxxd.md
-    ⅱ.
-  c. daily -> topic/topic_l1/topic_l1.md + xxx.md + topic_l2
-  d. proactive
+   a. MEMORY.md
+   b. msg/files -> daily/YYYYMMDD/YYYYMMDD.md + xxxx.md
+   ⅰ. YYYYMMDD.md
+    1. xxx -> xxxx.md
+    2. xxx -> xxxd.md
+       ⅱ.
+       c. daily -> topic/topic_l1/topic_l1.md + xxx.md + topic_l2
+       d. proactive
 
 steps:
+
 1. 治理（算法+LLM）：
-  a. 节点关联P0：现有的链接做补充，挖掘新的LLM的link
-    ⅰ. /Users/yuli/workspace/ReMe/reme2/component/edge_extractor/llm_edge_extractor.py
-    ⅱ. 移动到steps
-  b. 节点整合/节点拆分/节点归档
-  c. 健康度检查
+   a. 节点关联P0：现有的链接做补充，挖掘新的LLM的link
+   ⅰ. /Users/yuli/workspace/ReMe/reme2/component/edge_extractor/llm_edge_extractor.py
+   ⅱ. 移动到steps
+   b. 节点整合/节点拆分/节点归档
+   c. 健康度检查
 2. retrieve 调用store的检索
 3. 原子steps：reme edit
 4. 组合steps：总结：
-  a. - freq (every_n_turn、compact) -> daily_summarizer
-  b. topic (/dream ) -> topic_summarizer(daily_xx -> topic_xx)
-  c. proactive -> proactive_summarizer(personal_xxx -> proactive_query - pre_query
+   a. - freq (every_n_turn、compact) -> daily_summarizer
+   b. topic (/dream ) -> topic_summarizer(daily_xx -> topic_xx)
+   c. proactive -> proactive_summarizer(personal_xxx -> proactive_query - pre_query

reme4/components/client/http_client.py

@@ -50,7 +50,11 @@ class HttpClient(BaseClient):
             self.client = httpx.AsyncClient(base_url=self.base_url, timeout=self.timeout)
 
     async def _iter_stream_chunks(self) -> AsyncGenerator[StreamChunk, None]:
-        """Send request and yield StreamChunks; auto-detects JSON vs SSE via Content-Type."""
+        """Send request and yield raw StreamChunks; auto-detects JSON vs SSE via Content-Type.
+
+        For JSON responses: yields a single CONTENT chunk with the raw response body.
+        For SSE responses: yields each streaming chunk as it arrives.
+        """
         if self.client is None:
             raise RuntimeError("Client not initialized. Call _start() first.")
 
@@ -79,16 +83,10 @@ class HttpClient(BaseClient):
                     yield chunk
             else:
                 body = await resp.aread()
-                text = body.decode()
-                try:
-                    data = json.loads(text)
-                    pretty = json.dumps(data, indent=2, ensure_ascii=False)
-                except json.JSONDecodeError:
-                    pretty = text
-                yield StreamChunk(chunk_type=ChunkEnum.CONTENT, chunk=pretty)
+                yield StreamChunk(chunk_type=ChunkEnum.CONTENT, chunk=body.decode())
 
     async def stream_chunks(self) -> AsyncGenerator[StreamChunk, None]:
-        """HTTP-specific richer access: yield full StreamChunk objects with chunk_type/metadata."""
+        """HTTP-specific richer access: yield raw StreamChunk objects (no display formatting)."""
         async for chunk in self._iter_stream_chunks():
             yield chunk
 
@@ -105,12 +103,38 @@ class HttpClient(BaseClient):
                 actions.append({"action": path.lstrip("/"), "method": method.upper(), **op})
         return actions
 
+    @staticmethod
+    def _format_for_display(text: str) -> str:
+        """Render a JSON response as human-friendly CLI text; pass through unrecognized payloads."""
+        try:
+            data = json.loads(text)
+        except (ValueError, json.JSONDecodeError):
+            return text
+        if not (isinstance(data, dict) and isinstance(data.get("answer"), str)):
+            return json.dumps(data, indent=2, ensure_ascii=False) if isinstance(data, (dict, list)) else text
+        d = dict(data)
+        answer = d.pop("answer")
+        success = d.pop("success", None)
+        metadata = d.pop("metadata", None)
+        parts = [answer]
+        status_pieces = []
+        if success is not None:
+            status_pieces.append("✅" if success else "❌")
+        if metadata:
+            status_pieces.append(json.dumps(metadata, ensure_ascii=False))
+        if status_pieces:
+            parts.append(" ".join(status_pieces))
+        if d:
+            parts.append(json.dumps(d, indent=2, ensure_ascii=False))
+        return "\n".join(parts)
+
     # pylint: disable=invalid-overridden-method
     async def _execute(self) -> AsyncGenerator[str, None]:
-        """Yield text chunks; one yield for JSON endpoints, many for SSE."""
+        """Yield text chunks for CLI display; JSON responses are pretty-formatted."""
         async for chunk in self._iter_stream_chunks():
             payload = chunk.chunk
-            yield payload if isinstance(payload, str) else json.dumps(payload, ensure_ascii=False)
+            text = payload if isinstance(payload, str) else json.dumps(payload, ensure_ascii=False)
+            yield self._format_for_display(text)
 
     async def _close(self) -> None:
         """Close the HTTP client."""

reme4/components/embedding/base_embedding_model.py

@@ -198,6 +198,7 @@ class BaseEmbeddingModel(BaseComponent):
             if len(self._embedding_cache) >= self.max_cache_size:
                 break
             self._embedding_cache[str(key)] = emb.astype(np.float16)
+        self.logger.info(f"Loaded {len(self._embedding_cache)} embeddings from {self.cache_path}")
 
     async def dump(self) -> None:
         """Persist in-memory cache to disk (npz format)."""
@@ -208,5 +209,6 @@ class BaseEmbeddingModel(BaseComponent):
         embeddings = np.stack(list(self._embedding_cache.values()))
         try:
             np.savez(self.cache_path, keys=np.array(keys, dtype=str), embeddings=embeddings)
+            self.logger.info(f"Saved {len(self._embedding_cache)} embeddings to {self.cache_path}")
         except Exception:
             self.logger.exception("Failed to save embedding cache")

reme4/components/file_graph/local_file_graph.py

@@ -24,11 +24,6 @@ class LocalFileGraph(BaseFileGraph):
         await super()._start()
         await self.load()
         await self.rebuild_links()
-        self.logger.info(
-            f"LocalFileGraph '{self.graph_name}' ready: "
-            f"{len(self._nodes)} nodes, {sum(len(s) for s in self._inverse.values())} edges, "
-            f"{sum(len(s) for s in self._pending.values())} pending",
-        )
 
     async def _close(self) -> None:
         await self.dump()
@@ -43,6 +38,7 @@ class LocalFileGraph(BaseFileGraph):
                 self._nodes.update(
                     (n.path, n) for line in f if line.strip() for n in [FileNode.model_validate_json(line)]
                 )
+            self.logger.info(f"Loaded {len(self._nodes)} nodes from {self._graph_file}")
         except Exception as e:
             self.logger.exception(f"Failed to load {self._graph_file}: {e}")
 
@@ -53,6 +49,7 @@ class LocalFileGraph(BaseFileGraph):
             with open(tmp, "w", encoding="utf-8") as f:
                 f.writelines(f"{n.model_dump_json()}\n" for n in self._nodes.values())
             tmp.replace(self._graph_file)
+            self.logger.info(f"Saved {len(self._nodes)} nodes to {self._graph_file}")
         except Exception as e:
             self.logger.exception(f"Failed to write {self._graph_file}: {e}")
 
@@ -123,6 +120,7 @@ class LocalFileGraph(BaseFileGraph):
         self._nodes.clear()
         self._inverse.clear()
         self._pending.clear()
+        self._graph_file.unlink(missing_ok=True)
 
     # -- Link access -------------------------------------------------------
 

reme4/components/file_graph/nx_file_graph.py

@@ -29,12 +29,6 @@ class NxFileGraph(BaseFileGraph):
     async def _start(self) -> None:
         await super()._start()
         await self.load()
-        n_real = sum(1 for _, d in self._graph.nodes(data=True) if "node" in d)
-        self.logger.info(
-            f"NxFileGraph '{self.graph_name}' ready: "
-            f"{n_real} nodes, {self._graph.number_of_edges()} edges, "
-            f"{self._graph.number_of_nodes() - n_real} virtual",
-        )
 
     async def _close(self) -> None:
         await self.dump()
@@ -47,6 +41,8 @@ class NxFileGraph(BaseFileGraph):
         try:
             with open(self._graph_file, "rb") as f:
                 self._graph = pickle.load(f)
+            n_real = sum(1 for _, d in self._graph.nodes(data=True) if "node" in d)
+            self.logger.info(f"Loaded {n_real} nodes from {self._graph_file}")
         except Exception as e:
             self.logger.exception(f"Failed to load {self._graph_file}: {e}")
 
@@ -57,6 +53,8 @@ class NxFileGraph(BaseFileGraph):
             with open(tmp, "wb") as f:
                 pickle.dump(self._graph, f, protocol=pickle.HIGHEST_PROTOCOL)
             tmp.replace(self._graph_file)
+            n_real = sum(1 for _, d in self._graph.nodes(data=True) if "node" in d)
+            self.logger.info(f"Saved {n_real} nodes to {self._graph_file}")
         except Exception as e:
             self.logger.exception(f"Failed to write {self._graph_file}: {e}")
 
@@ -101,8 +99,9 @@ class NxFileGraph(BaseFileGraph):
         )
 
     async def clear(self):
-        """Remove all nodes and edges."""
+        """Remove all nodes and edges, and remove persisted file."""
         self._graph.clear()
+        self._graph_file.unlink(missing_ok=True)
 
     # -- Link access -------------------------------------------------------
 

reme4/components/file_store/local_file_store.py

@@ -24,7 +24,6 @@ class LocalFileStore(BaseFileStore):
     async def _start(self) -> None:
         await super()._start()
         await self.load()
-        self.logger.info(f"LocalFileStore '{self.store_name}' ready: {len(self.file_chunks)} chunks")
 
     async def _close(self) -> None:
         await self.dump()
@@ -42,6 +41,7 @@ class LocalFileStore(BaseFileStore):
                     if line:
                         chunk = FileChunk.model_validate_json(line)
                         self.file_chunks[chunk.id] = chunk
+            self.logger.info(f"Loaded {len(self.file_chunks)} chunks from {self.chunks_path}")
         except Exception as e:
             self.logger.exception(f"Failed to load {self.chunks_path}: {e}")
 
@@ -52,6 +52,7 @@ class LocalFileStore(BaseFileStore):
             async with aiofiles.open(tmp, "w", encoding=self.encoding) as f:
                 await f.write("\n".join(c.model_dump_json() for c in self.file_chunks.values()))
             tmp.replace(self.chunks_path)
+            self.logger.info(f"Saved {len(self.file_chunks)} chunks to {self.chunks_path}")
         except Exception as e:
             self.logger.exception(f"Failed to write {self.chunks_path}: {e}")
         if self.keyword_index:
@@ -125,6 +126,7 @@ class LocalFileStore(BaseFileStore):
         if not self.file_graph:
             raise RuntimeError("file_graph is required for clear")
         self.file_chunks.clear()
+        self.chunks_path.unlink(missing_ok=True)
         if self.keyword_index:
             await self.keyword_index.clear()
         await self.file_graph.clear()

reme4/components/keyword_index/base_keyword_index.py

@@ -25,12 +25,10 @@ class BaseKeywordIndex(BaseComponent):
     async def _start(self) -> None:
         """Load existing index from disk if available."""
         await self.load()
-        self.logger.info(f"Loaded index from {self.index_path}")
 
     async def _close(self) -> None:
         """Save index to disk on shutdown."""
         await self.dump()
-        self.logger.info(f"Saved index to {self.index_path}")
 
     @property
     def index_file(self) -> Path:

reme4/components/keyword_index/bm25_index.py

@@ -143,6 +143,7 @@ class BM25Index(BaseKeywordIndex):
                     f,
                 )
             tmp.replace(self.index_file)
+            self.logger.info(f"Saved {self.n_docs} docs to {self.index_file}")
         except Exception as e:
             self.logger.exception(f"Failed to write {self.index_file}: {e}")
 
@@ -160,18 +161,20 @@ class BM25Index(BaseKeywordIndex):
             self.k1 = data.get("k1", 1.5)
             self.b = data.get("b", 0.75)
             self._idf_cache = {}
+            self.logger.info(f"Loaded {self.n_docs} docs from {self.index_file}")
         except Exception as e:
             self.logger.exception(f"Failed to load index: {e}")
             self.index_file.unlink(missing_ok=True)
             await self.clear()
 
     async def clear(self) -> None:
-        """Reset index to empty state."""
+        """Reset index to empty state and remove persisted file."""
         self.vocab = {}
         self.inverted_index = {}
         self.doc_meta = {}
         self.total_len = 0
         self._idf_cache = {}
+        self.index_file.unlink(missing_ok=True)
 
     async def optimize_index(self) -> None:
         """Rebuild vocab to remove unused tokens and compact token IDs."""

reme4/config/default.yaml

@@ -49,6 +49,15 @@ jobs:
     steps:
       - backend: help_step
 
+  - backend: base
+    name: reindex
+    description: "wipe the file store and rebuild it from the watcher's tracked files"
+    parameters:
+      type: object
+      properties: {}
+    steps:
+      - backend: reindex_step
+
   - backend: base
     name: search
     description: "hybrid search over file_store: vector + keyword fused via RRF"
@@ -144,8 +153,8 @@ components:
     default:
       backend: local
       store_name: default
-      embedding_model: default
-#      embedding_model: ""
+#      embedding_model: default
+      embedding_model: ""
       keyword_index: default
       file_graph: default
 

reme4/steps/common/health_check.py

@@ -58,6 +58,18 @@ def _embedding_status(comp) -> dict:
 
 
 def _file_graph_status(comp) -> dict:
+    # Nx backend: single _graph attribute holds nodes/edges, virtuals are nodes without "node" payload.
+    g = getattr(comp, "_graph", None)
+    if g is not None:
+        n_real = sum(1 for _, d in g.nodes(data=True) if "node" in d)
+        return {
+            "is_started": comp.is_started,
+            "n_nodes": n_real,
+            "n_edges": g.number_of_edges(),
+            "n_virtual": g.number_of_nodes() - n_real,
+            "memory": _mb_str(g),
+        }
+    # Local backend: separate dicts for nodes, resolved inverse edges, and pending edges.
     nodes = getattr(comp, "_nodes", {}) or {}
     inverse = getattr(comp, "_inverse", {}) or {}
     pending = getattr(comp, "_pending", {}) or {}
@@ -65,6 +77,7 @@ def _file_graph_status(comp) -> dict:
         "is_started": comp.is_started,
         "n_nodes": len(nodes),
         "n_edges": sum(len(s) for s in inverse.values()),
+        "n_pending": sum(len(s) for s in pending.values()),
         "memory": _mb_str(nodes, inverse, pending),
     }
 

reme4/steps/common/search.py

@@ -208,12 +208,15 @@ class SearchStep(BaseStep):
         answer_lines: list[str] = []
         for c in fused:
             answer_lines.append(
-                f"{c.path}:{c.start_line}-{c.end_line} [{self._format_scores(c.scores, hybrid)}] {c.text}",
+                f"========== {c.path}:{c.start_line}-{c.end_line} "
+                f"[{self._format_scores(c.scores, hybrid)}] ==========\n{c.text}",
             )
             answer_lines.extend(self._render_expansion_lines(link_expansion.get(c.path, {})))
 
         self.context.response.answer = "\n".join(answer_lines)
-        self.context.response.metadata["results"] = [c.model_dump(exclude_none=True) for c in fused]
+        self.context.response.metadata["results"] = [
+            c.model_dump(exclude_none=True, exclude={"embedding"}) for c in fused
+        ]
         self.context.response.metadata["link_expansion"] = link_expansion
         self.context.response.metadata["counts"] = {
             "vector": len(vector_results),

reme4/utils/common_utils.py

@@ -210,8 +210,9 @@ async def call_action(
 
     pieces: list[str] = []
     async with HttpClient(action=action, host=host, port=port, timeout=timeout, **kwargs) as client:
-        async for chunk in client():
-            pieces.append(chunk)
+        async for chunk in client.stream_chunks():
+            payload = chunk.chunk
+            pieces.append(payload if isinstance(payload, str) else json.dumps(payload, ensure_ascii=False))
     raw = "".join(pieces)
     try:
         return json.loads(raw)

tests4/unittest/test_common_steps.py

@@ -167,6 +167,28 @@ def test_search_job_missing_query():
     _run(run())
 
 
+def test_reindex_job():
+    """reindex job should wipe the file store and rebuild from tracked files."""
+
+    async def run():
+        with tempfile.TemporaryDirectory() as tmp, _temp_chdir(tmp):
+            async with mock_reme_server() as (host, port):
+                await call_and_check(
+                    "reindex",
+                    host=host,
+                    port=port,
+                    validator=lambda r: (
+                        isinstance(r, dict)
+                        and r.get("success") is True
+                        and isinstance(r.get("metadata", {}).get("counts"), dict)
+                        and "added" in r["metadata"]["counts"]
+                    ),
+                )
+        print("✓ test_reindex_job passed")
+
+    _run(run())
+
+
 def test_demo_job():
     """demo job should echo back the normalized query and adjusted min_score."""
 
@@ -235,6 +257,13 @@ def test_all_jobs_one_server():
                     query="anything",
                     validator=lambda r: isinstance(r, dict) and r.get("success") is True,
                 )
+                # reindex
+                await call_and_check(
+                    "reindex",
+                    host=host,
+                    port=port,
+                    validator=lambda r: isinstance(r, dict) and isinstance(r.get("metadata", {}).get("counts"), dict),
+                )
                 # demo
                 await call_and_check(
                     "demo",
@@ -255,6 +284,7 @@ if __name__ == "__main__":
     test_health_check_job()
     test_search_job_empty_store()
     test_search_job_missing_query()
+    test_reindex_job()
     test_demo_job()
     test_all_jobs_one_server()
     print("\n所有测试通过!")

tests4/unittest/test_file_graph.py

@@ -0,0 +1,333 @@
+"""Tests for FileGraph backends (LocalFileGraph + NxFileGraph)."""
+
+# pylint: disable=protected-access
+
+import asyncio
+import os
+import tempfile
+
+import pytest
+
+from reme4.components.file_graph import LocalFileGraph, NxFileGraph
+from reme4.schema import FileLink, FileNode
+
+
+class temp_chdir:
+    """Context manager to temporarily chdir into a path and restore on exit."""
+
+    def __init__(self, path):
+        self.path = path
+        self.old = None
+
+    def __enter__(self):
+        self.old = os.getcwd()
+        os.chdir(self.path)
+        return self
+
+    def __exit__(self, *exc):
+        os.chdir(self.old)
+
+
+def make_node(path: str, links: list[tuple[str, str | None]] | None = None) -> FileNode:
+    """Build a FileNode with the given outgoing (target_path, target_anchor) pairs."""
+    return FileNode(
+        path=path,
+        st_mtime=1.0,
+        links=[FileLink(source_path=path, target_path=t, target_anchor=a) for t, a in (links or [])],
+    )
+
+
+# Both backends should satisfy the same BaseFileGraph contract.
+BACKENDS = [LocalFileGraph, NxFileGraph]
+
+
+@pytest.mark.parametrize("backend_cls", BACKENDS)
+def test_upsert_and_get_nodes(backend_cls):
+    """upsert_nodes stores nodes; get_nodes returns them by path or all."""
+
+    async def run():
+        with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
+            graph = backend_cls()
+            await graph.start()
+
+            n1 = make_node("a.md", [("b.md", None)])
+            n2 = make_node("b.md")
+            await graph.upsert_nodes([n1, n2])
+
+            got_all = await graph.get_nodes()
+            assert {n.path for n in got_all} == {"a.md", "b.md"}
+
+            got_one = await graph.get_nodes(["a.md"])
+            assert len(got_one) == 1
+            assert got_one[0].path == "a.md"
+
+            got_missing = await graph.get_nodes(["nope.md"])
+            assert got_missing == []
+
+            await graph.close()
+            print(f"✓ test_upsert_and_get_nodes[{backend_cls.__name__}] passed")
+
+    asyncio.run(run())
+
+
+@pytest.mark.parametrize("backend_cls", BACKENDS)
+def test_upsert_replaces_old_links(backend_cls):
+    """Re-upserting a node with new links replaces the old outgoing edges."""
+
+    async def run():
+        with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
+            graph = backend_cls()
+            await graph.start()
+
+            await graph.upsert_nodes(
+                [
+                    make_node("a.md", [("b.md", None)]),
+                    make_node("b.md"),
+                    make_node("c.md"),
+                ]
+            )
+            assert {lnk.target_path for lnk in await graph.get_outlinks("a.md")} == {"b.md"}
+
+            # Replace a's link target from b → c
+            await graph.upsert_nodes([make_node("a.md", [("c.md", None)])])
+            assert {lnk.target_path for lnk in await graph.get_outlinks("a.md")} == {"c.md"}
+            # b should no longer have a as an inlink
+            assert await graph.get_inlinks("b.md") == []
+            assert {lnk.source_path for lnk in await graph.get_inlinks("c.md")} == {"a.md"}
+
+            await graph.close()
+            print(f"✓ test_upsert_replaces_old_links[{backend_cls.__name__}] passed")
+
+    asyncio.run(run())
+
+
+@pytest.mark.parametrize("backend_cls", BACKENDS)
+def test_outlinks_skip_virtual_targets(backend_cls):
+    """get_outlinks only returns links pointing to real (existing) nodes."""
+
+    async def run():
+        with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
+            graph = backend_cls()
+            await graph.start()
+
+            # a links to b (real) and ghost (virtual)
+            await graph.upsert_nodes(
+                [
+                    make_node("a.md", [("b.md", None), ("ghost.md", None)]),
+                    make_node("b.md"),
+                ]
+            )
+
+            outs = await graph.get_outlinks("a.md")
+            targets = {lnk.target_path for lnk in outs}
+            assert targets == {"b.md"}
+
+            await graph.close()
+            print(f"✓ test_outlinks_skip_virtual_targets[{backend_cls.__name__}] passed")
+
+    asyncio.run(run())
+
+
+@pytest.mark.parametrize("backend_cls", BACKENDS)
+def test_inlinks_promotion_after_upsert(backend_cls):
+    """Edges to virtual targets become real inlinks once the target is upserted."""
+
+    async def run():
+        with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
+            graph = backend_cls()
+            await graph.start()
+
+            # b doesn't exist yet — link is pending
+            await graph.upsert_nodes([make_node("a.md", [("b.md", None)])])
+            assert await graph.get_inlinks("b.md") == []  # b not real yet
+
+            # Now create b — pending edge promotes
+            await graph.upsert_nodes([make_node("b.md")])
+            inlinks = await graph.get_inlinks("b.md")
+            assert {lnk.source_path for lnk in inlinks} == {"a.md"}
+
+            await graph.close()
+            print(f"✓ test_inlinks_promotion_after_upsert[{backend_cls.__name__}] passed")
+
+    asyncio.run(run())
+
+
+@pytest.mark.parametrize("backend_cls", BACKENDS)
+def test_delete_node_demotes_inbound(backend_cls):
+    """Deleting a node makes it virtual; sources still hold the link, but get_inlinks([deleted]) is []."""
+
+    async def run():
+        with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
+            graph = backend_cls()
+            await graph.start()
+
+            await graph.upsert_nodes(
+                [
+                    make_node("a.md", [("b.md", None)]),
+                    make_node("b.md"),
+                ]
+            )
+            assert {lnk.source_path for lnk in await graph.get_inlinks("b.md")} == {"a.md"}
+
+            await graph.delete_nodes(["b.md"])
+            # b is no longer a real node
+            assert await graph.get_nodes(["b.md"]) == []
+            # inlinks query for a non-real node returns []
+            assert await graph.get_inlinks("b.md") == []
+            # a's outlink to b is hidden because b is virtual
+            assert await graph.get_outlinks("a.md") == []
+
+            # Re-upsert b — pending should re-promote
+            await graph.upsert_nodes([make_node("b.md")])
+            assert {lnk.source_path for lnk in await graph.get_inlinks("b.md")} == {"a.md"}
+
+            await graph.close()
+            print(f"✓ test_delete_node_demotes_inbound[{backend_cls.__name__}] passed")
+
+    asyncio.run(run())
+
+
+@pytest.mark.parametrize("backend_cls", BACKENDS)
+def test_delete_outgoing_links_cleared(backend_cls):
+    """Deleting a source node drops its outgoing edges (no inlink left on its targets)."""
+
+    async def run():
+        with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
+            graph = backend_cls()
+            await graph.start()
+
+            await graph.upsert_nodes(
+                [
+                    make_node("a.md", [("b.md", None)]),
+                    make_node("b.md"),
+                ]
+            )
+            await graph.delete_nodes(["a.md"])
+
+            assert await graph.get_inlinks("b.md") == []
+
+            await graph.close()
+            print(f"✓ test_delete_outgoing_links_cleared[{backend_cls.__name__}] passed")
+
+    asyncio.run(run())
+
+
+@pytest.mark.parametrize("backend_cls", BACKENDS)
+def test_clear(backend_cls):
+    """clear() removes all nodes and edges."""
+
+    async def run():
+        with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
+            graph = backend_cls()
+            await graph.start()
+
+            await graph.upsert_nodes(
+                [
+                    make_node("a.md", [("b.md", None)]),
+                    make_node("b.md"),
+                ]
+            )
+            await graph.clear()
+            assert await graph.get_nodes() == []
+
+            await graph.close()
+            print(f"✓ test_clear[{backend_cls.__name__}] passed")
+
+    asyncio.run(run())
+
+
+@pytest.mark.parametrize("backend_cls", BACKENDS)
+def test_rebuild_links_idempotent(backend_cls):
+    """rebuild_links produces the same outlink/inlink view as the original upserts."""
+
+    async def run():
+        with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
+            graph = backend_cls()
+            await graph.start()
+
+            await graph.upsert_nodes(
+                [
+                    make_node("a.md", [("b.md", None), ("c.md", "h")]),
+                    make_node("b.md"),
+                    make_node("c.md"),
+                ]
+            )
+
+            before_out = sorted((lnk.target_path, lnk.target_anchor) for lnk in await graph.get_outlinks("a.md"))
+            before_in = sorted(lnk.source_path for lnk in await graph.get_inlinks("b.md"))
+
+            await graph.rebuild_links()
+
+            after_out = sorted((lnk.target_path, lnk.target_anchor) for lnk in await graph.get_outlinks("a.md"))
+            after_in = sorted(lnk.source_path for lnk in await graph.get_inlinks("b.md"))
+
+            assert before_out == after_out
+            assert before_in == after_in
+
+            await graph.close()
+            print(f"✓ test_rebuild_links_idempotent[{backend_cls.__name__}] passed")
+
+    asyncio.run(run())
+
+
+@pytest.mark.parametrize("backend_cls", BACKENDS)
+def test_persistence_roundtrip(backend_cls):
+    """close() dumps; a fresh instance loads the same nodes from disk."""
+
+    async def run():
+        with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
+            g1 = backend_cls()
+            await g1.start()
+            await g1.upsert_nodes(
+                [
+                    make_node("a.md", [("b.md", None)]),
+                    make_node("b.md"),
+                ]
+            )
+            await g1.close()  # triggers dump
+
+            g2 = backend_cls()
+            await g2.start()  # triggers load
+            paths = sorted(n.path for n in await g2.get_nodes())
+            assert paths == ["a.md", "b.md"]
+            # Inlink relationship should also be reconstructable.
+            assert {lnk.source_path for lnk in await g2.get_inlinks("b.md")} == {"a.md"}
+            await g2.close()
+            print(f"✓ test_persistence_roundtrip[{backend_cls.__name__}] passed")
+
+    asyncio.run(run())
+
+
+@pytest.mark.parametrize("backend_cls", BACKENDS)
+def test_get_nodes_empty_inputs(backend_cls):
+    """get_nodes([]) returns []; get_nodes(None) returns all."""
+
+    async def run():
+        with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
+            graph = backend_cls()
+            await graph.start()
+
+            await graph.upsert_nodes([make_node("a.md")])
+            assert await graph.get_nodes([]) == []
+            assert len(await graph.get_nodes(None)) == 1
+
+            await graph.close()
+            print(f"✓ test_get_nodes_empty_inputs[{backend_cls.__name__}] passed")
+
+    asyncio.run(run())
+
+
+if __name__ == "__main__":
+    print("\n=== FileGraph Tests ===")
+    for backend in BACKENDS:
+        test_upsert_and_get_nodes(backend)
+        test_upsert_replaces_old_links(backend)
+        test_outlinks_skip_virtual_targets(backend)
+        test_inlinks_promotion_after_upsert(backend)
+        test_delete_node_demotes_inbound(backend)
+        test_delete_outgoing_links_cleared(backend)
+        test_clear(backend)
+        test_rebuild_links_idempotent(backend)
+        test_persistence_roundtrip(backend)
+        test_get_nodes_empty_inputs(backend)
+    print("\n所有测试通过!")

tests4/unittest/test_file_store.py

@@ -0,0 +1,330 @@
+"""Tests for LocalFileStore."""
+
+# pylint: disable=protected-access
+
+import asyncio
+import os
+import tempfile
+import warnings
+
+from reme4.components.file_store import LocalFileStore
+from reme4.schema import FileChunk, FileNode
+
+warnings.filterwarnings("ignore", category=DeprecationWarning, module="jieba")
+warnings.filterwarnings("ignore", category=DeprecationWarning, module="pkg_resources")
+
+
+class temp_chdir:
+    """Context manager to temporarily chdir into a path and restore on exit."""
+
+    def __init__(self, path):
+        self.path = path
+        self.old = None
+
+    def __enter__(self):
+        self.old = os.getcwd()
+        os.chdir(self.path)
+        return self
+
+    def __exit__(self, *exc):
+        os.chdir(self.old)
+
+
+async def make_store(store_name: str = "test_store", **kwargs) -> LocalFileStore:
+    """Build a started LocalFileStore with embedding disabled (no OpenAI dep)."""
+    store = LocalFileStore(store_name=store_name, embedding_model="", **kwargs)
+    await store.start()
+    return store
+
+
+def make_file(
+    path: str,
+    text: str,
+    chunk_count: int = 1,
+) -> tuple[FileNode, list[FileChunk]]:
+    """Build a (FileNode, [FileChunk]) tuple ready for upsert_file."""
+    chunks = [
+        FileChunk(id=f"{path}::chunk{i}", path=path, text=f"{text} part{i}", start_line=i, end_line=i + 1)
+        for i in range(chunk_count)
+    ]
+    node = FileNode(path=path, st_mtime=1.0, chunk_ids=[c.id for c in chunks])
+    return node, chunks
+
+
+def test_upsert_single_file():
+    """upsert_file with a single (node, chunks) tuple stores chunks and node."""
+
+    async def run():
+        with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
+            store = await make_store()
+
+            node, chunks = make_file("a.md", "hello world", chunk_count=2)
+            await store.upsert_file((node, chunks))
+
+            # Chunks landed in memory
+            assert len(store.file_chunks) == 2
+            assert {c.path for c in store.file_chunks.values()} == {"a.md"}
+            # Node landed in graph
+            nodes = await store.file_graph.get_nodes(["a.md"])
+            assert len(nodes) == 1
+            assert sorted(nodes[0].chunk_ids) == sorted([c.id for c in chunks])
+
+            await store.close()
+            print("✓ test_upsert_single_file passed")
+
+    asyncio.run(run())
+
+
+def test_upsert_multiple_files():
+    """upsert_file accepts a list of tuples and indexes them all."""
+
+    async def run():
+        with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
+            store = await make_store()
+
+            files = [make_file("a.md", "alpha"), make_file("b.md", "beta")]
+            await store.upsert_file(files)
+
+            assert len(store.file_chunks) == 2
+            paths = {n.path for n in await store.file_graph.get_nodes()}
+            assert paths == {"a.md", "b.md"}
+
+            await store.close()
+            print("✓ test_upsert_multiple_files passed")
+
+    asyncio.run(run())
+
+
+def test_upsert_replaces_old_chunks():
+    """Re-upserting the same path points the node at the new chunk set."""
+
+    async def run():
+        with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
+            store = await make_store()
+
+            n1, c1 = make_file("a.md", "v1", chunk_count=2)
+            await store.upsert_file((n1, c1))
+
+            # Different chunks for the same path
+            n2 = FileNode(path="a.md", st_mtime=2.0)
+            c2 = [FileChunk(id="a.md::new", path="a.md", text="v2 only", start_line=0, end_line=1)]
+            n2.chunk_ids = [c.id for c in c2]
+            await store.upsert_file((n2, c2))
+
+            # The node now references the new chunk set, not the old one.
+            nodes = await store.file_graph.get_nodes(["a.md"])
+            assert nodes[0].chunk_ids == ["a.md::new"]
+            assert "a.md::new" in store.file_chunks
+
+            await store.close()
+            print("✓ test_upsert_replaces_old_chunks passed")
+
+    asyncio.run(run())
+
+
+def test_delete_by_path_single():
+    """delete_by_path drops chunks and the node entry."""
+
+    async def run():
+        with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
+            store = await make_store()
+
+            await store.upsert_file([make_file("a.md", "alpha"), make_file("b.md", "beta")])
+            await store.delete_by_path("a.md")
+
+            assert all(c.path != "a.md" for c in store.file_chunks.values())
+            assert {n.path for n in await store.file_graph.get_nodes()} == {"b.md"}
+
+            await store.close()
+            print("✓ test_delete_by_path_single passed")
+
+    asyncio.run(run())
+
+
+def test_delete_by_path_list():
+    """delete_by_path accepts a list of paths."""
+
+    async def run():
+        with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
+            store = await make_store()
+
+            await store.upsert_file(
+                [
+                    make_file("a.md", "alpha"),
+                    make_file("b.md", "beta"),
+                    make_file("c.md", "gamma"),
+                ]
+            )
+            await store.delete_by_path(["a.md", "b.md"])
+
+            assert {n.path for n in await store.file_graph.get_nodes()} == {"c.md"}
+            assert all(c.path == "c.md" for c in store.file_chunks.values())
+
+            await store.close()
+            print("✓ test_delete_by_path_list passed")
+
+    asyncio.run(run())
+
+
+def test_delete_by_path_missing_is_noop():
+    """Deleting a nonexistent path is a no-op."""
+
+    async def run():
+        with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
+            store = await make_store()
+
+            await store.upsert_file(make_file("a.md", "alpha"))
+            before = len(store.file_chunks)
+            await store.delete_by_path("ghost.md")
+            assert len(store.file_chunks) == before
+
+            await store.close()
+            print("✓ test_delete_by_path_missing_is_noop passed")
+
+    asyncio.run(run())
+
+
+def test_clear():
+    """clear() empties chunks and the file graph."""
+
+    async def run():
+        with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
+            store = await make_store()
+
+            await store.upsert_file([make_file("a.md", "alpha"), make_file("b.md", "beta")])
+            await store.clear()
+
+            assert store.file_chunks == {}
+            assert await store.file_graph.get_nodes() == []
+
+            await store.close()
+            print("✓ test_clear passed")
+
+    asyncio.run(run())
+
+
+def test_keyword_search():
+    """keyword_search returns matching chunks ranked by BM25 score."""
+
+    async def run():
+        with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
+            store = await make_store()
+
+            await store.upsert_file(
+                [
+                    make_file("a.md", "python programming language"),
+                    make_file("b.md", "java programming language"),
+                    make_file("c.md", "python data analysis"),
+                ]
+            )
+
+            results = await store.keyword_search("python", limit=5, search_filter={})
+            paths = {r.path for r in results}
+            assert "a.md" in paths or "c.md" in paths
+            # Each result should carry a keyword score.
+            for r in results:
+                assert r.scores.get("keyword", 0) > 0
+
+            await store.close()
+            print("✓ test_keyword_search passed")
+
+    asyncio.run(run())
+
+
+def test_keyword_search_empty_query():
+    """Empty/whitespace queries return no results."""
+
+    async def run():
+        with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
+            store = await make_store()
+            await store.upsert_file(make_file("a.md", "hello"))
+
+            assert await store.keyword_search("", limit=5, search_filter={}) == []
+            assert await store.keyword_search("   ", limit=5, search_filter={}) == []
+
+            await store.close()
+            print("✓ test_keyword_search_empty_query passed")
+
+    asyncio.run(run())
+
+
+def test_vector_search_disabled_returns_empty():
+    """Without an embedding model, vector_search returns []."""
+
+    async def run():
+        with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
+            store = await make_store()
+            await store.upsert_file(make_file("a.md", "hello"))
+
+            assert store.embedding_model is None
+            assert await store.vector_search("hello", limit=5, search_filter={}) == []
+
+            await store.close()
+            print("✓ test_vector_search_disabled_returns_empty passed")
+
+    asyncio.run(run())
+
+
+def test_persistence_roundtrip():
+    """close() dumps chunks; a fresh store loads them from disk."""
+
+    async def run():
+        with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
+            s1 = await make_store()
+            await s1.upsert_file([make_file("a.md", "alpha"), make_file("b.md", "beta")])
+            await s1.close()
+
+            s2 = await make_store()
+            assert {c.path for c in s2.file_chunks.values()} == {"a.md", "b.md"}
+            # Graph should also be persisted independently via its own dump.
+            assert {n.path for n in await s2.file_graph.get_nodes()} == {"a.md", "b.md"}
+            await s2.close()
+            print("✓ test_persistence_roundtrip passed")
+
+    asyncio.run(run())
+
+
+def test_rebuild_links_delegates_to_graph():
+    """rebuild_links() on the store delegates to the underlying file_graph."""
+
+    async def run():
+        with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
+            store = await make_store()
+
+            from reme4.schema import FileLink
+
+            node = FileNode(
+                path="a.md",
+                st_mtime=1.0,
+                links=[FileLink(source_path="a.md", target_path="b.md")],
+            )
+            chunks = [FileChunk(id="a::1", path="a.md", text="x", start_line=0, end_line=1)]
+            node.chunk_ids = [c.id for c in chunks]
+            await store.upsert_file((node, chunks))
+            await store.upsert_file(make_file("b.md", "beta"))
+
+            await store.rebuild_links()
+            inlinks = await store.get_inlinks("b.md")
+            assert {lnk.source_path for lnk in inlinks} == {"a.md"}
+
+            await store.close()
+            print("✓ test_rebuild_links_delegates_to_graph passed")
+
+    asyncio.run(run())
+
+
+if __name__ == "__main__":
+    print("\n=== LocalFileStore Tests ===")
+    test_upsert_single_file()
+    test_upsert_multiple_files()
+    test_upsert_replaces_old_chunks()
+    test_delete_by_path_single()
+    test_delete_by_path_list()
+    test_delete_by_path_missing_is_noop()
+    test_clear()
+    test_keyword_search()
+    test_keyword_search_empty_query()
+    test_vector_search_disabled_returns_empty()
+    test_persistence_roundtrip()
+    test_rebuild_links_delegates_to_graph()
+    print("\n所有测试通过!")

tests4/unittest/test_file_watcher.py

@@ -0,0 +1,349 @@
+"""Tests for LiteFileWatcher (excluding the awatch main loop)."""
+
+# pylint: disable=protected-access
+
+import asyncio
+import os
+import tempfile
+import warnings
+from pathlib import Path
+
+from watchfiles import Change
+
+from reme4.components.file_parser import DefaultFileParser
+from reme4.components.file_store import LocalFileStore
+from reme4.components.file_watcher import LiteFileWatcher
+
+warnings.filterwarnings("ignore", category=DeprecationWarning, module="jieba")
+warnings.filterwarnings("ignore", category=DeprecationWarning, module="pkg_resources")
+
+
+class temp_chdir:
+    """Context manager to temporarily chdir into a path and restore on exit."""
+
+    def __init__(self, path):
+        self.path = path
+        self.old = None
+
+    def __enter__(self):
+        self.old = os.getcwd()
+        os.chdir(self.path)
+        return self
+
+    def __exit__(self, *exc):
+        os.chdir(self.old)
+
+
+async def make_watcher(watch_paths: list[str] | str = "vault", **kwargs) -> LiteFileWatcher:
+    """Build a LiteFileWatcher with real (started) file_store/file_parser, but no background loop.
+
+    We replace the bind() Dependency placeholders with concrete instances and start them
+    manually, so tests can call update_store / on_* directly without the background task.
+    """
+    watcher = LiteFileWatcher(watch_paths=watch_paths, **kwargs)
+    fs = LocalFileStore(store_name="test_store", embedding_model="")
+    parser = DefaultFileParser()
+    await fs.start()
+    await parser.start()
+    watcher.file_store = fs
+    watcher.file_parser = parser
+    return watcher
+
+
+async def teardown_watcher(watcher: LiteFileWatcher) -> None:
+    """Close the manually-started subcomponents."""
+    await watcher.file_parser.close()
+    await watcher.file_store.close()
+
+
+def write_file(path: Path, content: str = "x") -> Path:
+    """Create or overwrite a file and return the path."""
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(content, encoding="utf-8")
+    return path
+
+
+def test_watch_filter_default_md():
+    """Default suffix_filters=['md'] passes .md files and rejects others."""
+
+    async def run():
+        with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
+            (Path(tmpdir) / "vault").mkdir()
+            watcher = await make_watcher()
+
+            assert watcher.watch_filter(Change.added, "/x/foo.md")
+            assert not watcher.watch_filter(Change.added, "/x/foo.txt")
+            assert not watcher.watch_filter(Change.added, "/x/foo")
+
+            print("✓ test_watch_filter_default_md passed")
+
+    asyncio.run(run())
+
+
+def test_watch_filter_custom_suffix():
+    """Custom suffix_filters override the default."""
+
+    async def run():
+        with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
+            (Path(tmpdir) / "vault").mkdir()
+            watcher = await make_watcher(suffix_filters=["txt", ".rst"])
+
+            assert watcher.watch_filter(Change.added, "/x/foo.txt")
+            assert watcher.watch_filter(Change.added, "/x/foo.rst")
+            assert not watcher.watch_filter(Change.added, "/x/foo.md")
+
+            print("✓ test_watch_filter_custom_suffix passed")
+
+    asyncio.run(run())
+
+
+def test_watch_filter_no_filter_passes_all():
+    """When suffix_filters is empty (set after init), watch_filter passes everything."""
+
+    async def run():
+        with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
+            (Path(tmpdir) / "vault").mkdir()
+            watcher = await make_watcher()
+            # Constructor coerces [] / None → ["md"]; clear it directly to exercise the no-filter branch.
+            watcher.suffix_filters = []
+
+            assert watcher.watch_filter(Change.added, "/x/foo")
+            assert watcher.watch_filter(Change.added, "/x/foo.md")
+
+            print("✓ test_watch_filter_no_filter_passes_all passed")
+
+    asyncio.run(run())
+
+
+def test_relative_and_absolute_path_helpers():
+    """_get_relative_path strips working_path; _get_absolute_path resolves against it."""
+
+    async def run():
+        with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
+            (Path(tmpdir) / "vault").mkdir()
+            watcher = await make_watcher()
+
+            # Use watcher.working_path to match the same realpath form (macOS /var ↔ /private/var).
+            abs_in = (watcher.working_path / "vault" / "a.md").absolute()
+            assert watcher._get_relative_path(abs_in) == "vault/a.md"
+
+            # Path outside working_path → returns absolute
+            outside = Path("/opt/elsewhere/x.md").absolute()
+            assert watcher._get_relative_path(outside) == str(outside)
+
+            # _get_absolute_path: relative resolves under working_path
+            assert watcher._get_absolute_path("vault/a.md") == watcher.working_path / "vault/a.md"
+            # absolute stays absolute
+            assert watcher._get_absolute_path(str(abs_in)) == abs_in
+
+            print("✓ test_relative_and_absolute_path_helpers passed")
+
+    asyncio.run(run())
+
+
+def test_scan_existing_files_finds_md_recursive():
+    """scan_existing_files returns md files under watch_paths, recursively."""
+
+    async def run():
+        with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
+            vault = Path(tmpdir) / "vault"
+            write_file(vault / "a.md", "alpha")
+            write_file(vault / "sub" / "b.md", "beta")
+            write_file(vault / "ignore.txt", "skip")  # filtered by suffix
+            watcher = await make_watcher()
+
+            files = await watcher.scan_existing_files()
+            rels = set(files.keys())
+            assert "vault/a.md" in rels
+            assert "vault/sub/b.md" in rels
+            assert "vault/ignore.txt" not in rels
+
+            print("✓ test_scan_existing_files_finds_md_recursive passed")
+
+    asyncio.run(run())
+
+
+def test_scan_existing_files_non_recursive():
+    """recursive=False only scans direct children."""
+
+    async def run():
+        with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
+            vault = Path(tmpdir) / "vault"
+            write_file(vault / "a.md", "alpha")
+            write_file(vault / "sub" / "b.md", "beta")
+            watcher = await make_watcher(recursive=False)
+
+            files = await watcher.scan_existing_files()
+            rels = set(files.keys())
+            assert "vault/a.md" in rels
+            assert "vault/sub/b.md" not in rels
+
+            print("✓ test_scan_existing_files_non_recursive passed")
+
+    asyncio.run(run())
+
+
+def test_on_added_indexes_files():
+    """on_added parses files and writes them into the file_store."""
+
+    async def run():
+        with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
+            vault = Path(tmpdir) / "vault"
+            write_file(vault / "a.md", "hello")
+            watcher = await make_watcher()
+            try:
+                await watcher.on_added(["vault/a.md"])
+                nodes = await watcher.file_store.file_graph.get_nodes()
+                assert {n.path for n in nodes} == {"vault/a.md"}
+            finally:
+                await teardown_watcher(watcher)
+            print("✓ test_on_added_indexes_files passed")
+
+    asyncio.run(run())
+
+
+def test_on_modified_replaces_node():
+    """on_modified re-parses and updates the node entry for an existing path."""
+
+    async def run():
+        with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
+            vault = Path(tmpdir) / "vault"
+            file_path = write_file(vault / "a.md", "v1")
+            watcher = await make_watcher()
+            try:
+                await watcher.on_added(["vault/a.md"])
+                node_before = (await watcher.file_store.file_graph.get_nodes(["vault/a.md"]))[0]
+                # Bump mtime + content
+                file_path.write_text("v2 different content", encoding="utf-8")
+                os.utime(file_path, (node_before.st_mtime + 10, node_before.st_mtime + 10))
+
+                await watcher.on_modified(["vault/a.md"])
+                node_after = (await watcher.file_store.file_graph.get_nodes(["vault/a.md"]))[0]
+                assert node_after.st_mtime > node_before.st_mtime
+            finally:
+                await teardown_watcher(watcher)
+            print("✓ test_on_modified_replaces_node passed")
+
+    asyncio.run(run())
+
+
+def test_on_deleted_removes_node():
+    """on_deleted removes the node from the store regardless of file presence."""
+
+    async def run():
+        with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
+            vault = Path(tmpdir) / "vault"
+            write_file(vault / "a.md", "hello")
+            watcher = await make_watcher()
+            try:
+                await watcher.on_added(["vault/a.md"])
+                assert {n.path for n in await watcher.file_store.file_graph.get_nodes()} == {"vault/a.md"}
+
+                await watcher.on_deleted(["vault/a.md"])
+                assert await watcher.file_store.file_graph.get_nodes() == []
+            finally:
+                await teardown_watcher(watcher)
+            print("✓ test_on_deleted_removes_node passed")
+
+    asyncio.run(run())
+
+
+def test_update_store_initial_add():
+    """First update_store run on a fresh store reports all files as added."""
+
+    async def run():
+        with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
+            vault = Path(tmpdir) / "vault"
+            write_file(vault / "a.md", "alpha")
+            write_file(vault / "b.md", "beta")
+            watcher = await make_watcher()
+            try:
+                counts = await watcher.update_store(dump=False)
+                assert counts == {"added": 2, "modified": 0, "deleted": 0}
+                paths = {n.path for n in await watcher.file_store.file_graph.get_nodes()}
+                assert paths == {"vault/a.md", "vault/b.md"}
+            finally:
+                await teardown_watcher(watcher)
+            print("✓ test_update_store_initial_add passed")
+
+    asyncio.run(run())
+
+
+def test_update_store_detects_modify_and_delete():
+    """update_store distinguishes modified vs deleted vs added on a second pass."""
+
+    async def run():
+        with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
+            vault = Path(tmpdir) / "vault"
+            a = write_file(vault / "a.md", "alpha")
+            b = write_file(vault / "b.md", "beta")
+            watcher = await make_watcher()
+            try:
+                # Initial sync to seed the store.
+                await watcher.update_store(dump=False)
+
+                # Modify a (bump mtime), delete b, add c.
+                a.write_text("alpha-v2", encoding="utf-8")
+                os.utime(a, (9_999_999_999, 9_999_999_999))
+                b.unlink()
+                write_file(vault / "c.md", "gamma")
+
+                counts = await watcher.update_store(dump=False)
+                assert counts == {"added": 1, "modified": 1, "deleted": 1}
+                paths = {n.path for n in await watcher.file_store.file_graph.get_nodes()}
+                assert paths == {"vault/a.md", "vault/c.md"}
+            finally:
+                await teardown_watcher(watcher)
+            print("✓ test_update_store_detects_modify_and_delete passed")
+
+    asyncio.run(run())
+
+
+def test_update_store_no_changes():
+    """A second sync over an unchanged tree reports zero counts."""
+
+    async def run():
+        with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
+            vault = Path(tmpdir) / "vault"
+            write_file(vault / "a.md", "alpha")
+            watcher = await make_watcher()
+            try:
+                await watcher.update_store(dump=False)
+                counts = await watcher.update_store(dump=False)
+                assert counts == {"added": 0, "modified": 0, "deleted": 0}
+            finally:
+                await teardown_watcher(watcher)
+            print("✓ test_update_store_no_changes passed")
+
+    asyncio.run(run())
+
+
+def test_missing_watch_path_filtered():
+    """watch_paths entries that don't exist are dropped from self.watch_paths."""
+
+    async def run():
+        with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
+            (Path(tmpdir) / "vault").mkdir()
+            watcher = await make_watcher(watch_paths=["vault", "ghost"])
+            assert [p.name for p in watcher.watch_paths] == ["vault"]
+            print("✓ test_missing_watch_path_filtered passed")
+
+    asyncio.run(run())
+
+
+if __name__ == "__main__":
+    print("\n=== LiteFileWatcher Tests ===")
+    test_watch_filter_default_md()
+    test_watch_filter_custom_suffix()
+    test_watch_filter_no_filter_passes_all()
+    test_relative_and_absolute_path_helpers()
+    test_scan_existing_files_finds_md_recursive()
+    test_scan_existing_files_non_recursive()
+    test_on_added_indexes_files()
+    test_on_modified_replaces_node()
+    test_on_deleted_removes_node()
+    test_update_store_initial_add()
+    test_update_store_detects_modify_and_delete()
+    test_update_store_no_changes()
+    test_missing_watch_path_filtered()
+    print("\n所有测试通过!")