diff --git a/docs4/reme_design.md b/docs4/reme_design.md index c382bfa1..c513dcc4 100644 --- a/docs4/reme_design.md +++ b/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 diff --git a/reme4/components/client/http_client.py b/reme4/components/client/http_client.py index 1ccb7620..efec1ff7 100644 --- a/reme4/components/client/http_client.py +++ b/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.""" diff --git a/reme4/components/embedding/base_embedding_model.py b/reme4/components/embedding/base_embedding_model.py index 02e2c877..0d03ec61 100644 --- a/reme4/components/embedding/base_embedding_model.py +++ b/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") diff --git a/reme4/components/file_graph/local_file_graph.py b/reme4/components/file_graph/local_file_graph.py index 2d6f58bb..e68ef46f 100644 --- a/reme4/components/file_graph/local_file_graph.py +++ b/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 ------------------------------------------------------- diff --git a/reme4/components/file_graph/nx_file_graph.py b/reme4/components/file_graph/nx_file_graph.py index 485212c5..c082ad04 100644 --- a/reme4/components/file_graph/nx_file_graph.py +++ b/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 ------------------------------------------------------- diff --git a/reme4/components/file_store/local_file_store.py b/reme4/components/file_store/local_file_store.py index c02dfa5f..60b920d0 100644 --- a/reme4/components/file_store/local_file_store.py +++ b/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() diff --git a/reme4/components/keyword_index/base_keyword_index.py b/reme4/components/keyword_index/base_keyword_index.py index c577e878..0be21c58 100644 --- a/reme4/components/keyword_index/base_keyword_index.py +++ b/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: diff --git a/reme4/components/keyword_index/bm25_index.py b/reme4/components/keyword_index/bm25_index.py index 3c2591ac..81899828 100644 --- a/reme4/components/keyword_index/bm25_index.py +++ b/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.""" diff --git a/reme4/config/default.yaml b/reme4/config/default.yaml index 5e753124..1a501912 100644 --- a/reme4/config/default.yaml +++ b/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 diff --git a/reme4/steps/common/health_check.py b/reme4/steps/common/health_check.py index fa7d55c0..6fc72652 100644 --- a/reme4/steps/common/health_check.py +++ b/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), } diff --git a/reme4/steps/common/search.py b/reme4/steps/common/search.py index deded425..595d665e 100644 --- a/reme4/steps/common/search.py +++ b/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), diff --git a/reme4/utils/common_utils.py b/reme4/utils/common_utils.py index 53996187..7df0dbef 100644 --- a/reme4/utils/common_utils.py +++ b/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) diff --git a/tests4/unittest/test_common_steps.py b/tests4/unittest/test_common_steps.py index a2f60559..ad7d5f5a 100644 --- a/tests4/unittest/test_common_steps.py +++ b/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所有测试通过!") diff --git a/tests4/unittest/test_file_graph.py b/tests4/unittest/test_file_graph.py new file mode 100644 index 00000000..a029b02a --- /dev/null +++ b/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所有测试通过!") diff --git a/tests4/unittest/test_file_store.py b/tests4/unittest/test_file_store.py new file mode 100644 index 00000000..cc59efd1 --- /dev/null +++ b/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所有测试通过!") diff --git a/tests4/unittest/test_file_watcher.py b/tests4/unittest/test_file_watcher.py new file mode 100644 index 00000000..f7008567 --- /dev/null +++ b/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所有测试通过!")