From 71e42dbad022cc8f02fbaa60b37bfd3b2b2133ab Mon Sep 17 00:00:00 2001 From: jinliyl <6469360+jinliyl@users.noreply.github.com> Date: Fri, 22 May 2026 10:26:36 +0800 Subject: [PATCH] refactor(reme4): replace file_watcher component with background steps pipeline (#251) * up * up * up * up * up * up * up * up * up * up * up * feat(config): add daily_dir configuration and background job logging - Added daily_dir setting with default value 'memory' to config - Implemented logging for background job startup events - Enhanced component start logic to handle background backend type - Updated default YAML configuration structure * refactor(file_parser): replace _get_relative_path with to_vault_relative method - Remove redundant working_dir property from base file parser - Add to_vault_relative method to base component for path resolution - Update bare_file_parser to use new to_vault_relative method - Update default_file_parser to use new to_vault_relative method - Update linked_file_parser to use new to_vault_relative method - Make working_path absolute in base_component and steps - Simplify index_changes step by removing redundant base variable - Consolidate path relative logic in single shared method * docs(reme4): update report with detailed architecture sections - Add comprehensive Markdown kernel section covering Obsidian compatibility - Include detailed explanation of YAML front matter and wikilink formats - Document smart slicing mechanism using Markdown AST instead of fixed tokens - Explain graph indexing with bidirectional links and multiple backends - Restructure sections with proper numbering from 4 to 7 - Move Markdown kernel section to appear before self-evolution features - Add detailed explanations of auto-memory, auto-dream, and auto-link processes - Document three-way hybrid search with RRF fusion and progressive expansion - Include engineering value explanations for keyword indexing in Chinese context --- docs4/reme4_report.md | 186 +++++----- reme4/application.py | 55 ++- reme4/components/__init__.py | 2 - reme4/components/base_component.py | 10 +- .../components/file_graph/base_file_graph.py | 22 ++ .../components/file_graph/local_file_graph.py | 7 +- reme4/components/file_graph/nx_file_graph.py | 8 - .../file_parser/bare_file_parser.py | 2 +- .../file_parser/base_file_parser.py | 12 +- .../file_parser/default_file_parser.py | 99 +++-- .../file_parser/linked_file_parser.py | 4 +- .../components/file_store/base_file_store.py | 102 ++--- .../components/file_store/local_file_store.py | 103 ++++-- reme4/components/file_watcher/__init__.py | 9 - .../file_watcher/base_file_watcher.py | 118 ------ .../file_watcher/lite_file_watcher.py | 129 ------- reme4/components/job/__init__.py | 7 +- reme4/components/job/background_job.py | 74 ++++ reme4/components/job/base_job.py | 8 +- reme4/components/runtime_context.py | 2 + reme4/components/service/base_service.py | 4 +- reme4/config/default.yaml | 70 +++- reme4/enumeration/component_enum.py | 2 - reme4/schema/application_config.py | 2 +- reme4/steps/__init__.py | 2 + reme4/steps/background/__init__.py | 11 + reme4/steps/background/index_changes.py | 81 ++++ reme4/steps/background/update_store.py | 66 ++++ reme4/steps/background/watch_changes.py | 67 ++++ reme4/steps/base_step.py | 40 +- reme4/steps/common/health_check.py | 12 - reme4/steps/crud/_file_io.py | 2 +- tests4/unittest/test_background_steps.py | 295 +++++++++++++++ tests4/unittest/test_common_steps.py | 124 ------- tests4/unittest/test_default_file_parser.py | 88 +++++ tests4/unittest/test_file_store.py | 49 ++- tests4/unittest/test_file_watcher.py | 349 ------------------ 37 files changed, 1148 insertions(+), 1075 deletions(-) delete mode 100644 reme4/components/file_watcher/__init__.py delete mode 100644 reme4/components/file_watcher/base_file_watcher.py delete mode 100644 reme4/components/file_watcher/lite_file_watcher.py create mode 100644 reme4/components/job/background_job.py create mode 100644 reme4/steps/background/__init__.py create mode 100644 reme4/steps/background/index_changes.py create mode 100644 reme4/steps/background/update_store.py create mode 100644 reme4/steps/background/watch_changes.py create mode 100644 tests4/unittest/test_background_steps.py delete mode 100644 tests4/unittest/test_file_watcher.py diff --git a/docs4/reme4_report.md b/docs4/reme4_report.md index 07ff4147..b9983961 100644 --- a/docs4/reme4_report.md +++ b/docs4/reme4_report.md @@ -11,10 +11,10 @@ > **ReMe 是一个把本地 Markdown 自进化成知识图谱的个人记忆引擎。** +- **记忆分层（形态的物化）** → 记忆按"原始 → 加工"两层组织：`resource/`（原始素材）、`daily/`（日记事件）是只增不删的流水帐；`digest/` 是加工层，下分 `personal/`（个性化）、`knowledge/`（主题知识）、`procedural/`（Agent 任务经验）、`proactive/`（主动洞察）四个固定子目录，写入策略和检索权重各有差异。 - **本地 Markdown（载体）** → 所有记忆都是 Obsidian 兼容的 .md 文件——YAML front matter、四种 wikilink（`[[X]]` / `[[X#anchor]]` / `[[X|alias]]` / `![[X]]`）、Dataview 风格 `predicate:: [[X]]` 语义关系全部沿用社区约定。用户可读、可备份、可迁移，对抗黑盒。 - **自进化（机制）** → 不需要用户手工整理，Agent 在后台让笔记自己长出结构。这一点把 ReMe 同时与"手动建图的 Obsidian"和"扁平存储的 Mem0"拉开。【event/trace】 - **知识图谱（结果）** → 自进化的产出物不是一堆扁平笔记，而是一张可被**多模 + 渐进式检索**消费的图：向量 + 关键词（中文 BM25）+ 图谱三路 RRF 融合，返回时通过 1-hop 邻居 meta 让 Agent"先看目录、再决定要不要展开正文"，不像传统 RAG 那样一次性把 top-K 切片塞进上下文。 -- **记忆分层（形态的物化）** → 记忆按"原始 → 加工"两层组织：`resource/`（原始素材）、`daily/`（日记事件）是只增不删的流水帐；`digest/` 是加工层，下分 `personal/`（个性化）、`knowledge/`（主题知识）、`procedural/`（Agent 任务经验）、`proactive/`（主动洞察）四个固定子目录，写入策略和检索权重各有差异。 - **被集成而非内置（分发形态）** → ReMe 不做独立 Agent 产品，而是作为**能力**被任意 Harness 调用：SDK 深度集成（qwenpaw / AgentScope）、MCP Tool（Claude Code / Cursor / Cherry Studio）、CLI + skill.md 三条路径并行，记忆跟着用户走，不绑定任何上层框架。 支撑这一切的是**自研轻量索引内核**——纯 Python(后期可以使用rust重写索引内核) + 文件持久化，无 sqlite / chroma @@ -146,9 +146,85 @@ ReMe 不把对话一股脑塞进数据库，而是按"原始 → 加工"分两 这一点对 Leader 视角尤其重要：用户对自己数据的掌控感，是所有"个人记忆"产品的信任基础。 ---- +## 四、Markdown 内核：把文件当数据库 -## 四、记忆的自进化（核心差异化） +### 4.1 Obsidian 兼容的 Markdown 格式 + +ReMe 没有发明新格式，而是完全复用 Obsidian 生态的约定： + +- **YAML front matter**：标题、标签、描述、自定义字段。 + + ```markdown + --- + title: 光伏产业链研究 + description: 从硅料到组件的全链条梳理 + tags: [新能源, 光伏, 产业链] + parent: 新能源 + author: 张三 + updated: 2026-05-19 + --- + + # 正文从这里开始 + ``` + + `title` / `description` / `tags` 是约定字段（参见 `reme4/schema/file_front_matter.py`），其余键值对作为 extras + 全部保留，可被检索和图索引消费。 + +- **4 种 wikilink 写法**： + - `[[X]]`：标准链接 + - `[[X#anchor]]`：链接到文件中的章节 + - `[[X|alias]]`：自定义显示文本 + - `![[X]]`：嵌入引用 +- **Dataview 风格语义关系**：`predicate:: [[X]]`，例如 `parent:: [[新能源]]`、`founder:: [[张三]]`，把 link 升级为带类型的" + 边"。 +- 标准 `[text](xxx.md)` 链接也会被识别为图边。 + +**意义**：用户的知识库可以直接用 Obsidian 打开做可视化浏览，可以用 Obsidian 插件做扩展。ReMe 不是替代 Obsidian，而是**给 +Obsidian 加上一个会自己写笔记的 Agent**。 + +``` + +### 4.2 比 RAG 更聪明的切片 + +传统 RAG 用固定 token 长度 + overlap 切片，经常切坏文档结构。ReMe 用 Markdown AST 切片： + +- 解析为章节嵌套树（按 H1/H2/H3 分层）。 +- 按章节边界递归切分，保留语义完整性。 +- **每个 chunk 自带完整的标题骨架（TOC）**：检索回来的片段一眼就能看出"这段在哪个章节、什么主题下"。 + +``` +某 chunk 实际内容长这样： +───────────────────── +# 光伏产业链 +## 上游：硅料 +### 多晶硅工艺 +[chunk 正文] +## 中游：硅片 +## 下游：组件 +───────────────────── +``````` + +Agent 拿到这个 chunk，立刻知道层级位置，不会断章取义。 + +### 4.3 Graph 索引：双向链接 + +定位句里"自进化成**知识图谱**"的物理形态，就落在这一节——每个文件参与两套索引： + +- **正向（outlinks）**：A → B（A 引用了 B） +- **反向（inlinks）**：B ← {A, C, D}（谁引用了 B） + +**反向链接**是知识库可用性的关键 —— 让你站在任意一个概念上，看到"还有哪些地方提到过我"。 + +ReMe 提供三种 graph backend，按规模和需求切换： + +- **本地 dict + JSONL**：轻量、零依赖、适合个人规模。 +- **NetworkX + pickle**：方便做图算法分析。 +- **Neo4j**：企业规模、Cypher 查询、可视化丰富。 + +切换只需配置一行。 + + +## 五、记忆的自进化（核心差异化） > 这是 ReMe 最重要的能力，也是和市面所有「记忆即数据库」产品的根本分野。 > @@ -157,7 +233,7 @@ ReMe 不把对话一股脑塞进数据库，而是按"原始 → 加工"分两 定位句中"自进化成知识图谱"的具体路径，由下面三件套共同承担：**auto-memory** 在前线把对话拆成事件，**auto-dream** 在空闲时把事件归档成主题，**auto-link** 把这一切用 wikilink 串成图。三者协作，daily 流水最终被织成一张越用越密的个人知识图谱。 -### 4.1 Auto-Memory：实时拆事件 +### 5.1 Auto-Memory：实时拆事件 主对话进行时，ReMe 在后台把上下文按"事件"自动拆分： @@ -169,7 +245,7 @@ ReMe 不把对话一股脑塞进数据库，而是按"原始 → 加工"分两 ：不需要手动整理。打开当天主索引，事件已经分章节列好，每条都能跳转到独立笔记。这就像有一个秘书在你说话的同时帮你做" 会议纪要的分章节"。 -### 4.2 Auto-Dream：空闲整理 +### 5.2 Auto-Dream：空闲整理 借鉴人在睡眠中"记忆巩固"的机制： @@ -181,7 +257,7 @@ ReMe 不把对话一股脑塞进数据库，而是按"原始 → 加工"分两 这是 ReMe 区别于"对话历史搜索"的关键 —— **它会自己整理**。 -### 4.3 Auto-Link：自动建图 +### 5.3 Auto-Link：自动建图 后台任务自动从正文里识别实体、候选链接，把隐式关系写回 wikilink： @@ -191,7 +267,7 @@ ReMe 不把对话一股脑塞进数据库，而是按"原始 → 加工"分两 **用户体验**：知识库随时间自然"越长越密"。浏览时可以从任意一处跳转到相关全部上下文，类似于在自己的脑子里"联想"。 -### 4.4 三者协同：从对话到知识图谱的自然演化 +### 5.4 三者协同：从对话到知识图谱的自然演化 ``` [实时] [离线] [持续] @@ -206,11 +282,10 @@ ReMe 不把对话一股脑塞进数据库，而是按"原始 → 加工"分两 整个过程**不需要用户操心**。用户只需要正常和 Agent 对话，三个月后回头看，就有了一张按主题组织、互相关联、可视化浏览的个人知识图谱——这就是一句话定位里"自进化成知识图谱"的物理产物。 ---- -## 五、检索体验：多模检索 + 渐进式展开 +## 六、检索体验：多模检索 + 渐进式展开 -### 5.1 三路融合的混合检索 +### 6.1 三路融合的混合检索 ReMe 同时跑三种检索通路，结果通过 RRF（Reciprocal Rank Fusion）排序融合： @@ -226,7 +301,7 @@ ReMe 同时跑三种检索通路，结果通过 RRF（Reciprocal Rank Fusion） 三路融合让检索像"三个人各自查一遍再开会确认"，结果鲁棒得多。 -### 5.2 渐进式展开 +### 6.2 渐进式展开 传统 RAG 是一次性把 top-K 切片塞进上下文，token 利用率低，而且经常带进不相关的噪音。ReMe 的检索（ `reme4/steps/common/search.py`）是**分跳**的，且每一跳的"信息密度"刻意不同： @@ -281,7 +356,7 @@ outlink + 10 个 inlink 的 meta，每条只占一行，**整张二跳目录的 **工程价值**：上下文窗口永远只装最相关的部分，token 成本可控；Agent 也能更精确地解释"我为什么知道这个"——因为它能引用 predicate=upstream、anchor=#应用 这种带语义的边。 -### 5.3 关键词索引的工程价值 +### 6.3 关键词索引的工程价值 很多人忽视：**做中文知识库，关键词检索比向量更重要**。 @@ -293,83 +368,6 @@ ReMe 自研增量 BM25 倒排索引，配合 jieba 中文分词： --- -## 六、Markdown 内核：把文件当数据库 - -### 6.1 Obsidian 兼容的 Markdown 格式 - -ReMe 没有发明新格式，而是完全复用 Obsidian 生态的约定： - -- **YAML front matter**：标题、标签、描述、自定义字段。 - - ```markdown - --- - title: 光伏产业链研究 - description: 从硅料到组件的全链条梳理 - tags: [新能源, 光伏, 产业链] - parent: 新能源 - author: 张三 - updated: 2026-05-19 - --- - - # 正文从这里开始 - ``` - - `title` / `description` / `tags` 是约定字段（参见 `reme4/schema/file_front_matter.py`），其余键值对作为 extras - 全部保留，可被检索和图索引消费。 - -- **4 种 wikilink 写法**： - - `[[X]]`：标准链接 - - `[[X#anchor]]`：链接到文件中的章节 - - `[[X|alias]]`：自定义显示文本 - - `![[X]]`：嵌入引用 -- **Dataview 风格语义关系**：`predicate:: [[X]]`，例如 `parent:: [[新能源]]`、`founder:: [[张三]]`，把 link 升级为带类型的" - 边"。 -- 标准 `[text](xxx.md)` 链接也会被识别为图边。 - -**意义**：用户的知识库可以直接用 Obsidian 打开做可视化浏览，可以用 Obsidian 插件做扩展。ReMe 不是替代 Obsidian，而是**给 -Obsidian 加上一个会自己写笔记的 Agent**。 - -````### 6.2 比 RAG 更聪明的切片 - -传统 RAG 用固定 token 长度 + overlap 切片，经常切坏文档结构。ReMe 用 Markdown AST 切片： - -- 解析为章节嵌套树（按 H1/H2/H3 分层）。 -- 按章节边界递归切分，保留语义完整性。 -- **每个 chunk 自带完整的标题骨架（TOC）**：检索回来的片段一眼就能看出"这段在哪个章节、什么主题下"。 - -``` -某 chunk 实际内容长这样： -───────────────────── -# 光伏产业链 -## 上游：硅料 -### 多晶硅工艺 -[chunk 正文] -## 中游：硅片 -## 下游：组件 -───────────────────── -``````` - -Agent 拿到这个 chunk，立刻知道层级位置，不会断章取义。 - -### 6.3 Graph 索引：双向链接 - -定位句里"自进化成**知识图谱**"的物理形态，就落在这一节——每个文件参与两套索引： - -- **正向（outlinks）**：A → B（A 引用了 B） -- **反向（inlinks）**：B ← {A, C, D}（谁引用了 B） - -**反向链接**是知识库可用性的关键 —— 让你站在任意一个概念上，看到"还有哪些地方提到过我"。 - -ReMe 提供三种 graph backend，按规模和需求切换： - -- **本地 dict + JSONL**：轻量、零依赖、适合个人规模。 -- **NetworkX + pickle**：方便做图算法分析。 -- **Neo4j**：企业规模、Cypher 查询、可视化丰富。 - -切换只需配置一行。 - ---- - ## 七、工程架构：可扩展、可替换、可演进 ### 7.1 Component 框架 @@ -428,11 +426,11 @@ jobs: ### 8.1 三种集成路径 -| 路径 | 适用对象 | 体验 | -|--------------------|-----------------------------------------------------|--------------------------------------------------------------------| -| **SDK 集成** | qwenpaw / AgentScope 等深度合作框架 | 直接调用 `AgentscopeTools`，无感拥有 auto-memory / auto-dream / auto-search | -| **MCP Tool** | 任何支持 MCP 的客户端（Claude Code / Cursor / Cherry Studio） | 配 skill.md，开箱即用 | -| **CLI + skill.md** | 通用方案，兜底所有 Harness | 一条命令调用，shell 友好 | +| 路径 | 适用对象 | 体验 | +|-------------------------|-----------------------------------------------------|--------------------------------------------------------------------| +| **SDK 集成** | qwenpaw / AgentScope 等深度合作框架 | 直接调用 `AgentscopeTools`，无感拥有 auto-memory / auto-dream / auto-search | +| **MCP Tool + skill.md** | 任何支持 MCP 的客户端（Claude Code / Cursor / Cherry Studio） | 配 skill.md，开箱即用 | +| **CLI + skill.md** | 通用方案，兜底所有 Harness | 一条命令调用，shell 友好 | 三条路径的设计哲学是：**不强迫任何 Agent 框架做 ReMe-specific 的改造**。 diff --git a/reme4/application.py b/reme4/application.py index a06dadbd..76adeccd 100644 --- a/reme4/application.py +++ b/reme4/application.py @@ -16,21 +16,18 @@ class Application(BaseComponent): def __init__(self, **kwargs) -> None: self.context = ApplicationContext(**kwargs) + self._started_components: list[BaseComponent] = [] working_path = Path(self.config.working_dir).absolute() working_path.mkdir(parents=True, exist_ok=True) (working_path / self.config.metadata_dir).mkdir(parents=True, exist_ok=True) (working_path / self.config.daily_dir).mkdir(parents=True, exist_ok=True) - (working_path / self.config.knowledge_dir).mkdir(parents=True, exist_ok=True) + (working_path / self.config.digest_dir).mkdir(parents=True, exist_ok=True) if self.config.enable_logo: print_logo(self.config) - logger = get_logger( - log_to_console=self.config.log_to_console, - log_to_file=self.config.log_to_file, - force_init=True, - ) + logger = get_logger(log_to_console=self.config.log_to_console, log_to_file=self.config.log_to_file) logger.info(f"Initializing {self.config.app_name} Application") super().__init__() @@ -113,37 +110,30 @@ class Application(BaseComponent): return ordered async def _start(self) -> None: - """Start components in topological order, then jobs.""" - start_order = self._topological_order() - order_str = " -> ".join(f"{c.component_type.value}:{c.name}" for c in start_order) - self.logger.info(f"Component start order: {order_str}") + """Start components, then regular jobs, then background jobs; record order for reverse close.""" + components = self._topological_order() + jobs = list(self.context.jobs.values()) + sequence = ( + components + [j for j in jobs if j.backend != "background"] + [j for j in jobs if j.backend == "background"] + ) - for component in start_order: + for c in sequence: try: - await component.start() + if c.backend == "background": + self.logger.info(f"Starting background job: {c.name}") + await c.start() + self._started_components.append(c) except Exception as e: - self.logger.exception(f"Failed to start {component.component_type.value}:{component.name}: {e}") - - for name, job in self.context.jobs.items(): - try: - await job.start() - except Exception as e: - self.logger.exception(f"Failed to start job '{name}': {e}") + self.logger.exception(f"Failed to start {c.component_type.value}:{c.name}: {e}") async def _close(self) -> None: - """Close all jobs, then components in reverse.""" - for name, job in self.context.jobs.items(): + """Close in reverse order of successful start.""" + for c in reversed(self._started_components): try: - await job.close() + await c.close() except Exception as e: - self.logger.exception(f"Failed to close job '{name}': {e}") - - for components in self.context.components.values(): - for component in components.values(): - try: - await component.close() - except Exception as e: - self.logger.exception(f"Failed to close {component.component_type.value}:{component.name}: {e}") + self.logger.exception(f"Failed to close {c.component_type.value}:{c.name}: {e}") + self._started_components.clear() async def run_job(self, name: str, /, **kwargs) -> Response: """Execute a registered job by name.""" @@ -169,6 +159,7 @@ class Application(BaseComponent): def run_app(self): """Start the service and serve the application.""" - if self.context.service is None: - raise RuntimeError("Service not configured") + from .components.service import BaseService + + assert isinstance(self.context.service, BaseService) self.context.service.run_app(app=self) diff --git a/reme4/components/__init__.py b/reme4/components/__init__.py index 148eeeaa..336da644 100644 --- a/reme4/components/__init__.py +++ b/reme4/components/__init__.py @@ -8,7 +8,6 @@ from . import embedding from . import file_graph from . import file_parser from . import file_store -from . import file_watcher from . import job from . import keyword_index from . import service @@ -35,7 +34,6 @@ __all__ = [ "file_graph", "file_parser", "file_store", - "file_watcher", "job", "keyword_index", "service", diff --git a/reme4/components/base_component.py b/reme4/components/base_component.py index 4c614d52..fa38dc96 100644 --- a/reme4/components/base_component.py +++ b/reme4/components/base_component.py @@ -125,7 +125,7 @@ class BaseComponent(ABC): """Resolved working directory from app context or cwd.""" if self.app_context is None: return Path.cwd() - return Path(self.app_context.app_config.working_dir) + return Path(self.app_context.app_config.working_dir).absolute() @property def working_metadata_path(self) -> Path: @@ -134,6 +134,14 @@ class BaseComponent(ABC): return Path.cwd() / "metadata" return self.working_path / self.app_context.app_config.metadata_dir + def to_vault_relative(self, path: str | Path) -> str: + """Return path relative to working_path; absolute path string if outside.""" + abs_path = Path(path).absolute() + try: + return str(abs_path.relative_to(self.working_path)) + except ValueError: + return str(abs_path) + # ----- Lifecycle ----------------------------------------------------- async def _start(self) -> None: diff --git a/reme4/components/file_graph/base_file_graph.py b/reme4/components/file_graph/base_file_graph.py index 333d75db..a5ef4110 100644 --- a/reme4/components/file_graph/base_file_graph.py +++ b/reme4/components/file_graph/base_file_graph.py @@ -20,6 +20,28 @@ class BaseFileGraph(BaseComponent): self.graph_path: Path = self.working_metadata_path / self.component_type.value self.graph_path.mkdir(parents=True, exist_ok=True) + # -- Lifecycle --------------------------------------------------------- + + async def _start(self) -> None: + await super()._start() + await self.load() + + async def _close(self) -> None: + await self.dump() + await super()._close() + + async def load(self) -> None: + """Load persisted state. No-op for backends without local files. + + Called at the end of ``_start()`` after base resources are ready + but before subclass-specific resources are initialised. Backends + that need their own resources for loading should override + ``_start()`` instead of this hook. + """ + + async def dump(self) -> None: + """Persist state. No-op for backends without local files.""" + # -- Node CRUD --------------------------------------------------------- @abstractmethod diff --git a/reme4/components/file_graph/local_file_graph.py b/reme4/components/file_graph/local_file_graph.py index e68ef46f..0966ebfc 100644 --- a/reme4/components/file_graph/local_file_graph.py +++ b/reme4/components/file_graph/local_file_graph.py @@ -21,14 +21,9 @@ class LocalFileGraph(BaseFileGraph): # -- Lifecycle --------------------------------------------------------- async def _start(self) -> None: - await super()._start() - await self.load() + await super()._start() # base calls load() await self.rebuild_links() - async def _close(self) -> None: - await self.dump() - await super()._close() - async def load(self) -> None: """Load nodes from JSONL file into memory; keep current state on failure.""" if not self._graph_file.exists(): diff --git a/reme4/components/file_graph/nx_file_graph.py b/reme4/components/file_graph/nx_file_graph.py index c082ad04..0a007c45 100644 --- a/reme4/components/file_graph/nx_file_graph.py +++ b/reme4/components/file_graph/nx_file_graph.py @@ -26,14 +26,6 @@ class NxFileGraph(BaseFileGraph): # -- Lifecycle --------------------------------------------------------- - async def _start(self) -> None: - await super()._start() - await self.load() - - async def _close(self) -> None: - await self.dump() - await super()._close() - async def load(self) -> None: """Load graph from pickle file; keep current graph on failure.""" if not self._graph_file.exists(): diff --git a/reme4/components/file_parser/bare_file_parser.py b/reme4/components/file_parser/bare_file_parser.py index e0ecef05..55b12e5c 100644 --- a/reme4/components/file_parser/bare_file_parser.py +++ b/reme4/components/file_parser/bare_file_parser.py @@ -19,4 +19,4 @@ class BareFileParser(BaseFileParser): async def parse(self, path: str | Path) -> tuple[FileNode, list[FileChunk]]: file_path = Path(path) stat = file_path.stat() - return FileNode(path=self._get_relative_path(path), st_mtime=stat.st_mtime, links=[], chunk_ids=[]), [] + return FileNode(path=self.to_vault_relative(path), st_mtime=stat.st_mtime, links=[], chunk_ids=[]), [] diff --git a/reme4/components/file_parser/base_file_parser.py b/reme4/components/file_parser/base_file_parser.py index 30595385..4d1621f4 100644 --- a/reme4/components/file_parser/base_file_parser.py +++ b/reme4/components/file_parser/base_file_parser.py @@ -13,17 +13,9 @@ class BaseFileParser(BaseComponent): component_type = ComponentEnum.FILE_PARSER - def __init__(self, **kwargs): + def __init__(self, supported_extensions: list[str] | None = None, **kwargs): super().__init__(**kwargs) - self.working_dir = self.app_context.app_config.working_dir if self.app_context else "" - - def _get_relative_path(self, path: str | Path) -> str: - """Return path relative to working_dir, or absolute path if outside.""" - file_path = Path(path).absolute() - try: - return str(file_path.relative_to(Path(self.working_dir).absolute())) - except ValueError: - return str(file_path) + self.supported_extensions: list[str] = supported_extensions or [] @abstractmethod async def parse(self, path: str | Path) -> tuple[FileNode, list[FileChunk]]: diff --git a/reme4/components/file_parser/default_file_parser.py b/reme4/components/file_parser/default_file_parser.py index 69a7c694..dfb68440 100644 --- a/reme4/components/file_parser/default_file_parser.py +++ b/reme4/components/file_parser/default_file_parser.py @@ -11,17 +11,18 @@ from .base_file_parser import BaseFileParser from ..component_registry import R from ...schema import FileChunk, FileFrontMatter, FileLink, FileNode -# Single-pass wikilink + optional Dataview predicate. -# Covers: [[X]] / [[X#h]] / [[X|alias]] / pred:: [[X]] / [pred:: [[X]]] -# - predicate group: optional leading '[' (Dataview inline-bracket form), an identifier, +# Single-pass wikilink + optional dataview predicate. +# Covers: [[X]] / ![[X]] / [[X#h]] / [[X|alias]] / pred:: [[X]] / [pred:: [[X]]] +# - predicate group: optional leading '[' (dataview inline-bracket form), an identifier, # then '::' — the whole prefix is non-capturing-optional so bare wikilinks still match. -# - target / anchor: target stops before '#', '|', '[', ']'; anchor stops before '|', '[', ']'. +# - optional '!' prefix matches the embed form (![[X]]). +# - target / anchor / alias all forbid '\n' so a wikilink cannot span lines. # - alias '|...': consumed but not captured (we don't need display text). _LINK_RE = re.compile( r"(?:\[?\s*(?P[A-Za-z][\w-]*)\s*::\s*)?" - r"\[\[\s*(?P[^\[\]|#]+?)" - r"(?:#(?P[^\[\]|]+?))?" - r"\s*(?:\|[^\[\]]*?)?\s*\]\]", + r"!?\[\[\s*(?P[^\[\]|#\n]+?)" + r"(?:#(?P[^\[\]|\n]+?))?" + r"\s*(?:\|[^\[\]\n]*?)?\s*]]", ) @@ -37,7 +38,7 @@ class DefaultFileParser(BaseFileParser): @staticmethod def parse_links(content: str, source_path: str) -> list[FileLink]: - """Extract wikilinks with optional Dataview predicate as outgoing FileLinks.""" + """Extract wikilinks with optional dataview predicate as outgoing FileLinks.""" links: list[FileLink] = [] for m in _LINK_RE.finditer(content): target = m["target"].strip() @@ -72,7 +73,7 @@ class DefaultFileParser(BaseFileParser): async def parse(self, path: str | Path) -> tuple[FileNode, list[FileChunk]]: file_path = Path(path) stat = file_path.stat() - rel_path = self._get_relative_path(path) + rel_path = self.to_vault_relative(path) async with aiofiles.open(file_path, encoding=self.encoding) as f: text = await f.read() @@ -80,12 +81,18 @@ class DefaultFileParser(BaseFileParser): if not text: return FileNode(path=rel_path, st_mtime=stat.st_mtime), [] - front_matter, content = self._parse_front_matter(text) - if not content: - return FileNode(path=rel_path, st_mtime=stat.st_mtime, front_matter=front_matter), [] + is_markdown = file_path.suffix.lower() == ".md" + if is_markdown: + front_matter, content = self._parse_front_matter(text) + if not content: + return FileNode(path=rel_path, st_mtime=stat.st_mtime, front_matter=front_matter), [] + links = self.parse_links(content, rel_path) + else: + front_matter = FileFrontMatter() + content = text + links = [] - links = self.parse_links(content, rel_path) - chunks = self._chunk_content(content, rel_path) + chunks = self._chunk_content(content, rel_path, parse_links=is_markdown) chunk_ids = [c.id for c in chunks] return ( FileNode( @@ -98,16 +105,59 @@ class DefaultFileParser(BaseFileParser): chunks, ) - def _chunk_content(self, content: str, rel_path: str) -> list[FileChunk]: - """Split content into overlapping byte-range chunks with line numbers.""" + def _link_byte_spans(self, content: str) -> list[tuple[int, int]]: + """Return [start, end) byte spans of every wikilink in content.""" + spans: list[tuple[int, int]] = [] + last_char, last_byte = 0, 0 + for m in _LINK_RE.finditer(content): + last_byte += len(content[last_char : m.start()].encode(self.encoding)) + match_bytes = len(m.group(0).encode(self.encoding)) + spans.append((last_byte, last_byte + match_bytes)) + last_byte += match_bytes + last_char = m.end() + return spans + + @staticmethod + def _span_containing( + pos: int, + spans: list[tuple[int, int]], + starts: list[int], + ) -> tuple[int, int] | None: + """Return the span strictly containing pos (s < pos < e), or None.""" + idx = bisect_right(starts, pos) - 1 + if idx < 0: + return None + s, e = spans[idx] + return (s, e) if s < pos < e else None + + def _chunk_content(self, content: str, rel_path: str, parse_links: bool = True) -> list[FileChunk]: + """Split content into overlapping byte-range chunks, avoiding cuts inside wikilinks. + + When ``parse_links`` is False, skip wikilink span computation and boundary checks + — used for non-markdown files where wikilink semantics don't apply. + """ content_bytes = content.encode(self.encoding) + n = len(content_bytes) newline_positions = [i for i, b in enumerate(content_bytes) if b == ord("\n")] + if parse_links: + link_spans = self._link_byte_spans(content) + link_starts = [s for s, _ in link_spans] + else: + link_spans: list[tuple[int, int]] = [] + link_starts: list[int] = [] + # Refuse to retreat past half of chunk_byte_size; falls back to hard cut + # for pathologically long links so we always make forward progress. + min_chunk = self.chunk_byte_size // 2 chunks: list[FileChunk] = [] - step = self.chunk_byte_size - self.overlap_byte_size start = 0 - while start < len(content_bytes): - end = min(start + self.chunk_byte_size, len(content_bytes)) + while start < n: + end = min(start + self.chunk_byte_size, n) + if end < n: + span = self._span_containing(end, link_spans, link_starts) + if span is not None and span[0] - start >= min_chunk: + end = span[0] + chunk_text = content_bytes[start:end].decode(self.encoding, errors="ignore") start_line = bisect_right(newline_positions, start - 1) + 1 end_line = bisect_right(newline_positions, end - 1) + 1 @@ -116,8 +166,15 @@ class DefaultFileParser(BaseFileParser): chunks.append( FileChunk(path=rel_path, start_line=start_line, end_line=end_line, text=chunk_text).set_hash_id(), ) - if end >= len(content_bytes): + if end >= n: break - start += step + + next_start = end - self.overlap_byte_size + span = self._span_containing(next_start, link_spans, link_starts) + if span is not None: + next_start = span[1] + if next_start <= start: + next_start = end + start = next_start return chunks diff --git a/reme4/components/file_parser/linked_file_parser.py b/reme4/components/file_parser/linked_file_parser.py index 2c825dfa..6bac9c3f 100644 --- a/reme4/components/file_parser/linked_file_parser.py +++ b/reme4/components/file_parser/linked_file_parser.py @@ -267,7 +267,7 @@ def _subtree_toc(n: MdNode) -> str: # -- Parser --------------------------------------------------------------- -@R.register("md") +@R.register("linked") class LinkedFileParser(BaseFileParser): """Markdown parser: frontmatter + wikilink edges + full-skeleton chunks.""" @@ -309,7 +309,7 @@ class LinkedFileParser(BaseFileParser): from mistletoe.block_token import Document file_path = Path(path) - rel_path = self._get_relative_path(path) + rel_path = self.to_vault_relative(path) post = frontmatter.loads(file_path.read_text(encoding=self.encoding)) chunks: list[FileChunk] = [] diff --git a/reme4/components/file_store/base_file_store.py b/reme4/components/file_store/base_file_store.py index 5cc6edd9..7237c60a 100644 --- a/reme4/components/file_store/base_file_store.py +++ b/reme4/components/file_store/base_file_store.py @@ -3,70 +3,56 @@ from abc import abstractmethod from ..base_component import BaseComponent -from ..embedding import BaseEmbeddingModel -from ..file_graph import BaseFileGraph -from ..keyword_index import BaseKeywordIndex from ...enumeration import ComponentEnum -from ...schema import FileChunk, FileNode, FileLink +from ...schema import FileChunk, FileLink, FileNode class BaseFileStore(BaseComponent): - """Abstract base for file store backends.""" + """Abstract base for file store backends. + + Defines the *semantic* contract a file store must offer: write (upsert / delete / clear), + retrieve (vector / keyword), and graph queries (nodes / links). Sub-component composition + (embedding model, keyword index, file graph) is each backend's implementation choice and + is not part of the base contract. + """ component_type = ComponentEnum.FILE_STORE - def __init__( - self, - store_name: str, - embedding_model: str = "default", - keyword_index: str = "default", - file_graph: str = "default", - store_version: str = "v1", - **kwargs, - ): + def __init__(self, store_name: str, store_version: str = "v1", **kwargs): super().__init__(**kwargs) - from ..embedding import OpenAIEmbeddingModel - from ..file_graph import LocalFileGraph - from ..keyword_index import BM25Index - self.store_name = store_name or self.name self.store_version = store_version - if not embedding_model and not keyword_index: - raise ValueError("At least one of embedding_model or keyword_index must be set.") - - self.embedding_model = self.bind(embedding_model, BaseEmbeddingModel, default_factory=OpenAIEmbeddingModel) - self.keyword_index = self.bind(keyword_index, BaseKeywordIndex, default_factory=BM25Index) - self.file_graph = self.bind(file_graph, BaseFileGraph, default_factory=LocalFileGraph) - self.store_path = self.working_metadata_path / self.component_type.value / store_name + self.store_path = self.working_metadata_path / self.component_type.value / self.store_name self.store_path.mkdir(parents=True, exist_ok=True) - async def _start(self) -> None: - """Probe embedding model; disable vector capability if it fails.""" - if self.embedding_model is None: - return - if not await self.embedding_model.health_check(): - self.logger.warning(f"{self.store_name}: embedding unhealthy, vector disabled") - self.embedding_model = None + # -- CRUD ------------------------------------------------------------ - def _disable_embedding(self, reason: str) -> None: - """Drop embedding after a runtime failure; keyword search still works.""" - if self.embedding_model is None: - return - self.logger.error(f"{self.store_name}: embedding disabled, {reason}") - self.embedding_model = None + @abstractmethod + async def upsert(self, files: list[tuple[FileNode, list[FileChunk]]]) -> None: + """Upsert files and their chunks into the store.""" - async def upsert_file( - self, - file: tuple[FileNode, list[FileChunk]] | list[tuple[FileNode, list[FileChunk]]], - ) -> None: - """Upsert a file and its chunks into the store.""" + @abstractmethod + async def delete(self, path: str | list[str]) -> None: + """Delete files by path from the store.""" - async def delete_by_path(self, path: str | list[str]) -> None: - """Delete files by their paths from the store.""" + @abstractmethod + async def get_nodes(self, paths: list[str] | None = None) -> list[FileNode]: + """Return file nodes; None = all nodes; missing paths are skipped.""" - async def clear(self): + @abstractmethod + async def get_outlinks(self, path: str) -> list[FileLink]: + """Return outgoing links for *path*.""" + + @abstractmethod + async def get_inlinks(self, path: str) -> list[FileLink]: + """Return incoming links for *path*.""" + + @abstractmethod + async def clear(self) -> None: """Clear the store of all files and chunks.""" + # -- Search ----------------------------------------------------------- + @abstractmethod async def vector_search(self, query: str, limit: int, search_filter: dict) -> list[FileChunk]: """Perform vector similarity search.""" @@ -74,27 +60,3 @@ class BaseFileStore(BaseComponent): @abstractmethod async def keyword_search(self, query: str, limit: int, search_filter: dict) -> list[FileChunk]: """Perform full-text keyword search.""" - - async def rebuild_links(self) -> None: - """Rebuild all edges from each node's link payload.""" - if not self.file_graph: - raise RuntimeError("file_graph is required for delete_by_path") - return await self.file_graph.rebuild_links() - - async def get_nodes(self, paths: list[str]) -> list[FileNode]: - """Return file nodes for the given paths (missing paths are skipped).""" - if not self.file_graph: - raise RuntimeError("file_graph is required for get_nodes") - return await self.file_graph.get_nodes(paths) - - async def get_outlinks(self, path: str) -> list[FileLink]: - """Return outgoing links for *path*.""" - if not self.file_graph: - raise RuntimeError("file_graph is required for delete_by_path") - return await self.file_graph.get_outlinks(path) - - async def get_inlinks(self, path: str) -> list[FileLink]: - """Return incoming links for *path*.""" - if not self.file_graph: - raise RuntimeError("file_graph is required for delete_by_path") - return await self.file_graph.get_inlinks(path) diff --git a/reme4/components/file_store/local_file_store.py b/reme4/components/file_store/local_file_store.py index 60b920d0..d3867038 100644 --- a/reme4/components/file_store/local_file_store.py +++ b/reme4/components/file_store/local_file_store.py @@ -5,16 +5,45 @@ import numpy as np from .base_file_store import BaseFileStore from ..component_registry import R -from ...schema import FileChunk, FileNode +from ..embedding import BaseEmbeddingModel +from ..file_graph import BaseFileGraph +from ..keyword_index import BaseKeywordIndex +from ...schema import FileChunk, FileLink, FileNode from ...utils import batch_cosine_similarity @R.register("local") class LocalFileStore(BaseFileStore): - """In-memory file store with deferred JSONL persistence.""" + """In-memory file store with deferred JSONL persistence. - def __init__(self, encoding: str = "utf-8", **kwargs): + Composes three subcomponents: ``embedding_model`` for vector retrieval, + ``keyword_index`` for full-text retrieval, and ``file_graph`` for node / link + storage. ``file_graph`` is mandatory; at least one of embedding / keyword + must be present. + """ + + def __init__( + self, + embedding_model: str = "default", + keyword_index: str = "default", + file_graph: str = "default", + encoding: str = "utf-8", + **kwargs, + ): super().__init__(**kwargs) + from ..embedding import OpenAIEmbeddingModel + from ..file_graph import LocalFileGraph + from ..keyword_index import BM25Index + + if not embedding_model and not keyword_index: + raise ValueError("At least one of embedding_model or keyword_index must be set.") + if not file_graph: + raise ValueError("file_graph is required for LocalFileStore.") + + self.embedding_model = self.bind(embedding_model, BaseEmbeddingModel, default_factory=OpenAIEmbeddingModel) + self.keyword_index = self.bind(keyword_index, BaseKeywordIndex, default_factory=BM25Index) + self.file_graph = self.bind(file_graph, BaseFileGraph, default_factory=LocalFileGraph) + self.encoding = encoding self.file_chunks: dict[str, FileChunk] = {} self.chunks_path = self.store_path / f"file_chunks_{self.store_version}.jsonl" @@ -23,6 +52,9 @@ class LocalFileStore(BaseFileStore): async def _start(self) -> None: await super()._start() + if self.embedding_model is not None and not await self.embedding_model.health_check(): + self.logger.warning(f"{self.store_name}: embedding unhealthy, vector disabled") + self.embedding_model = None await self.load() async def _close(self) -> None: @@ -30,6 +62,13 @@ class LocalFileStore(BaseFileStore): self.file_chunks.clear() await super()._close() + def _disable_embedding(self, reason: str) -> None: + """Drop embedding after a runtime failure; keyword search still works.""" + if self.embedding_model is None: + return + self.logger.error(f"{self.store_name}: embedding disabled, {reason}") + self.embedding_model = None + async def load(self) -> None: """Load chunks from JSONL file into memory.""" if not self.chunks_path.exists(): @@ -47,6 +86,7 @@ class LocalFileStore(BaseFileStore): async def dump(self) -> None: """Persist chunks to JSONL via atomic rename, then cascade to keyword_index and file_graph.""" + assert self.file_graph is not None try: tmp = self.chunks_path.with_suffix(".tmp") async with aiofiles.open(tmp, "w", encoding=self.encoding) as f: @@ -57,28 +97,23 @@ class LocalFileStore(BaseFileStore): self.logger.exception(f"Failed to write {self.chunks_path}: {e}") if self.keyword_index: await self.keyword_index.dump() - if self.file_graph: - await self.file_graph.dump() + await self.file_graph.dump() - # Base class interface + # CRUD - async def upsert_file( - self, - file: tuple[FileNode, list[FileChunk]] | list[tuple[FileNode, list[FileChunk]]], - ) -> None: - if not self.file_graph: - raise RuntimeError("file_graph is required for upsert_file") - if isinstance(file, tuple): - file = [file] + async def upsert(self, files: list[tuple[FileNode, list[FileChunk]]]) -> None: + if not files: + return + assert self.file_graph is not None - old_map = {n.path: n for n in await self.file_graph.get_nodes([node.path for node, _ in file])} + old_map = {n.path: n for n in await self.file_graph.get_nodes([node.path for node, _ in files])} new_nodes: list[FileNode] = [] needs_embed: list[FileChunk] = [] keyword_docs: dict[str, str] = {} - for node, chunks in file: + for node, chunks in files: old_node: FileNode | None = old_map.get(node.path) - cached = {} + cached: dict = {} if old_node and self.embedding_model: for cid in old_node.chunk_ids: old = self.file_chunks.pop(cid, None) @@ -107,24 +142,33 @@ class LocalFileStore(BaseFileStore): if self.keyword_index and keyword_docs: await self.keyword_index.add_docs(keyword_docs) - async def delete_by_path(self, path: str | list[str]) -> None: - if not self.file_graph: - raise RuntimeError("file_graph is required for delete_by_path") - if isinstance(path, str): - path = [path] - nodes = await self.file_graph.get_nodes(path) + async def delete(self, path: str | list[str]) -> None: + assert self.file_graph is not None + paths = [path] if isinstance(path, str) else path + nodes: list[FileNode] = await self.file_graph.get_nodes(paths) if not nodes: return deleted_chunk_ids = [cid for n in nodes for cid in n.chunk_ids] for cid in deleted_chunk_ids: self.file_chunks.pop(cid, None) - await self.file_graph.delete_nodes([n.path for n in nodes]) + await self.file_graph.delete_nodes([str(n.path) for n in nodes]) if self.keyword_index and deleted_chunk_ids: await self.keyword_index.delete_docs(deleted_chunk_ids) + async def get_nodes(self, paths: list[str] | None = None) -> list[FileNode]: + assert self.file_graph is not None + return await self.file_graph.get_nodes(paths) + + async def get_outlinks(self, path: str) -> list[FileLink]: + assert self.file_graph is not None + return await self.file_graph.get_outlinks(path) + + async def get_inlinks(self, path: str) -> list[FileLink]: + assert self.file_graph is not None + return await self.file_graph.get_inlinks(path) + async def clear(self) -> None: - if not self.file_graph: - raise RuntimeError("file_graph is required for clear") + assert self.file_graph is not None self.file_chunks.clear() self.chunks_path.unlink(missing_ok=True) if self.keyword_index: @@ -175,3 +219,10 @@ class LocalFileStore(BaseFileStore): results.append(chunk.model_copy(update={"scores": {"keyword": score, "score": score}})) return results + + # Extensions + + async def rebuild_links(self) -> None: + """Rebuild graph links via the underlying file graph.""" + assert self.file_graph is not None + return await self.file_graph.rebuild_links() diff --git a/reme4/components/file_watcher/__init__.py b/reme4/components/file_watcher/__init__.py deleted file mode 100644 index 0b11d995..00000000 --- a/reme4/components/file_watcher/__init__.py +++ /dev/null @@ -1,9 +0,0 @@ -"""File watcher implementations for monitoring file system changes.""" - -from .base_file_watcher import BaseFileWatcher -from .lite_file_watcher import LiteFileWatcher - -__all__ = [ - "BaseFileWatcher", - "LiteFileWatcher", -] diff --git a/reme4/components/file_watcher/base_file_watcher.py b/reme4/components/file_watcher/base_file_watcher.py deleted file mode 100644 index 3c6fbc9e..00000000 --- a/reme4/components/file_watcher/base_file_watcher.py +++ /dev/null @@ -1,118 +0,0 @@ -"""Abstract base for file watchers.""" - -import asyncio -from abc import abstractmethod -from pathlib import Path - -from watchfiles import Change - -from ..base_component import BaseComponent -from ..file_parser import BaseFileParser -from ..file_store import BaseFileStore -from ...enumeration import ComponentEnum - - -class BaseFileWatcher(BaseComponent): - """Abstract base for file watchers. Subclasses implement watch_loop and event handlers.""" - - component_type = ComponentEnum.FILE_WATCHER - - def __init__( - self, - watch_paths: list[str] | str, - suffix_filters: list[str] | None = None, - recursive: bool = True, - force_polling: bool = True, - debounce: int = 2000, - poll_delay_ms: int = 2000, - file_store: str = "default", - file_parser: str = "default", - **kwargs, - ): - super().__init__(**kwargs) - from ..file_parser import DefaultFileParser - from ..file_store import LocalFileStore - - watch_paths = [watch_paths] if isinstance(watch_paths, str) else watch_paths - base = self.working_path - self.watch_paths: list[Path] = [base / x for x in watch_paths if (base / x).exists()] - self.suffix_filters: list[str] = suffix_filters or ["md"] - self.recursive: bool = recursive - self.force_polling: bool = force_polling - self.debounce: int = debounce - self.poll_delay_ms: int = poll_delay_ms - self.file_store = self.bind(file_store, BaseFileStore, default_factory=LocalFileStore) - self.file_parser = self.bind(file_parser, BaseFileParser, default_factory=DefaultFileParser) - self._stop_event: asyncio.Event = asyncio.Event() - self._background_task: asyncio.Task | None = None - self._retry_interval: float = 10 - - async def _start(self): - self._stop_event = asyncio.Event() - self._background_task = asyncio.create_task(self._background_run()) - self.logger.info(f"Started watching: {[str(p) for p in self.watch_paths]}") - - async def _background_run(self): - """Sync store then enter watch loop.""" - await self.update_store() - await self.watch_loop() - - async def _close(self): - self._stop_event.set() - if self._background_task: - await self._background_task - self.logger.info("Stopped watching") - - def watch_filter(self, _change: Change, path: str) -> bool: - """Return True if the file suffix matches the filter list.""" - if not self.suffix_filters: - return True - return any(path.endswith("." + s.strip(".")) for s in self.suffix_filters) - - def _get_relative_path(self, path: str | Path) -> str: - """Return path relative to working_dir, or absolute path if outside.""" - file_path = Path(path).absolute() - try: - return str(file_path.relative_to(self.working_path.absolute())) - except ValueError: - return str(file_path) - - def _get_absolute_path(self, path: str | Path) -> Path: - """Return absolute path; relative paths are resolved against working_dir.""" - p = Path(path) - return p if p.is_absolute() else self.working_path / p - - async def scan_existing_files(self) -> dict[str, Path]: - """Collect watchable files under watch_paths as {relative_path: absolute_path}.""" - files: dict[str, Path] = {} - for path in self.watch_paths: - if not path.exists(): - continue - candidates = [path] if path.is_file() else (path.rglob("*") if self.recursive else path.iterdir()) - for p in candidates: - if p.is_file() and self.watch_filter(Change.added, str(p)): - files[self._get_relative_path(p)] = p.absolute() - return files - - @abstractmethod - async def watch_loop(self): - """Watch for file changes and dispatch events.""" - - @abstractmethod - async def update_store(self, dump: bool = True) -> dict[str, int]: - """Sync the store with watch_paths; dump store if any changes and dump=True. - - Returns counts {"added": int, "modified": int, "deleted": int}. - """ - - @abstractmethod - async def on_added(self, path: str | list[str]): - """Handle file added event (relative paths).""" - - @abstractmethod - async def on_modified(self, path: str | list[str]): - """Handle file modified event (relative paths).""" - - @abstractmethod - async def on_deleted(self, path: str | list[str]): - """Handle file deleted event (relative paths).""" diff --git a/reme4/components/file_watcher/lite_file_watcher.py b/reme4/components/file_watcher/lite_file_watcher.py deleted file mode 100644 index da48b2b3..00000000 --- a/reme4/components/file_watcher/lite_file_watcher.py +++ /dev/null @@ -1,129 +0,0 @@ -"""Polling-based file watcher using watchfiles.""" - -import asyncio - -from watchfiles import Change, awatch - -from .base_file_watcher import BaseFileWatcher -from ..component_registry import R -from ...schema import FileChunk, FileNode - - -@R.register("lite") -class LiteFileWatcher(BaseFileWatcher): - """Polling-based file watcher using watchfiles awatch.""" - - async def _interruptible_sleep(self): - """Sleep until stop or timeout, whichever comes first.""" - try: - await asyncio.wait_for(self._stop_event.wait(), timeout=self._retry_interval) - except asyncio.TimeoutError: - pass - - async def watch_loop(self): - if not self.watch_paths: - self.logger.warning("No watch paths specified") - return - - while not self._stop_event.is_set(): - valid_paths = [p for p in self.watch_paths if p.exists()] - if not valid_paths: - self.logger.warning(f"No valid paths, retrying in {self._retry_interval}s...") - await self._interruptible_sleep() - continue - - invalid = set(self.watch_paths) - set(valid_paths) - if invalid: - self.logger.warning(f"Skipping invalid paths: {[str(p) for p in invalid]}") - - try: - self.logger.info(f"Watching: {[str(p) for p in valid_paths]}") - async for changes in awatch( - *valid_paths, - watch_filter=self.watch_filter, - recursive=self.recursive, - force_polling=self.force_polling, - debounce=self.debounce, - poll_delay_ms=self.poll_delay_ms, - stop_event=self._stop_event, - ): - if self._stop_event.is_set(): - break - await self._dispatch_changes(changes) - except Exception: - self.logger.exception(f"Watch error, retrying in {self._retry_interval}s...") - if not self._stop_event.is_set(): - await self._interruptible_sleep() - - async def _dispatch_changes(self, changes: set[tuple[Change, str]]): - """Classify raw changes and dispatch to event handlers.""" - buckets: dict[Change, list[str]] = {Change.added: [], Change.modified: [], Change.deleted: []} - for c, p in changes: - if c in buckets: - buckets[c].append(self._get_relative_path(p)) - for change, handler, label in ( - (Change.added, self.on_added, "added"), - (Change.modified, self.on_modified, "modified"), - (Change.deleted, self.on_deleted, "deleted"), - ): - if buckets[change]: - self.logger.info(f"Detected {len(buckets[change])} {label} file(s)") - await handler(buckets[change]) - - async def update_store(self, dump: bool = True) -> dict[str, int]: - if self.file_store is None: - raise ValueError("file_store is not initialized!") - - existing: dict[str, float] = { - rel: abs_p.stat().st_mtime for rel, abs_p in (await self.scan_existing_files()).items() - } - indexed: dict[str, float] = {n.path: n.st_mtime for n in await self.file_store.file_graph.get_nodes()} - - to_delete = list(indexed.keys() - existing.keys()) - to_add = list(existing.keys() - indexed.keys()) - to_modify = [p for p in existing.keys() & indexed.keys() if existing[p] != indexed[p]] - - if to_modify: - self.logger.info(f"Updating {len(to_modify)} modified file(s)") - await self.on_modified(to_modify) - if to_delete: - self.logger.info(f"Removing {len(to_delete)} deleted file(s)") - await self.on_deleted(to_delete) - if to_add: - self.logger.info(f"Indexing {len(to_add)} new file(s)") - await self.on_added(to_add) - - changed = bool(to_add or to_modify or to_delete) - if not changed: - self.logger.info("Store is up to date") - if dump and changed: - await self.file_store.dump() - return {"added": len(to_add), "modified": len(to_modify), "deleted": len(to_delete)} - - async def _parse_and_upsert(self, paths: list[str], action: str): - """Parse files and upsert into store. Shared by on_added / on_modified.""" - if self.file_parser is None or self.file_store is None: - raise RuntimeError("file_parser or file_store is not initialized!") - - parsed: list[tuple[FileNode, list[FileChunk]]] = [] - for rel in paths: - abs_path = self._get_absolute_path(rel) - if abs_path.is_file(): - self.logger.info(f"{action} file: {rel}") - parsed.append(await self.file_parser.parse(abs_path)) - if parsed: - await self.file_store.delete_by_path([n.path for n, _ in parsed]) - await self.file_store.upsert_file(parsed) - - async def on_added(self, path: str | list[str]): - await self._parse_and_upsert([path] if isinstance(path, str) else path, "Adding") - - async def on_modified(self, path: str | list[str]): - await self._parse_and_upsert([path] if isinstance(path, str) else path, "Updating") - - async def on_deleted(self, path: str | list[str]): - if self.file_store is None: - raise RuntimeError("file_store is not initialized!") - paths = [path] if isinstance(path, str) else path - self.logger.info(f"Deleting {len(paths)} file(s)") - await self.file_store.delete_by_path(paths) diff --git a/reme4/components/job/__init__.py b/reme4/components/job/__init__.py index 0387572d..af8c0255 100644 --- a/reme4/components/job/__init__.py +++ b/reme4/components/job/__init__.py @@ -1,6 +1,11 @@ """Job components for executing workflows.""" +from .background_job import BackgroundJob from .base_job import BaseJob from .stream_job import StreamJob -__all__ = ["BaseJob", "StreamJob"] +__all__ = [ + "BackgroundJob", + "BaseJob", + "StreamJob", +] diff --git a/reme4/components/job/background_job.py b/reme4/components/job/background_job.py new file mode 100644 index 00000000..a0a4efa3 --- /dev/null +++ b/reme4/components/job/background_job.py @@ -0,0 +1,74 @@ +"""Long-running background job with optional supervisor.""" + +import asyncio +import random + +from .base_job import BaseJob +from ..component_registry import R +from ..runtime_context import RuntimeContext +from ...schema import Response + + +@R.register("background") +class BackgroundJob(BaseJob): + """Long-running job started by Application._start; runs __call__ until close. + + Subclasses override __call__ (or use the default step-based body). If + supervisor=True (default) and __call__ raises, it is restarted with + exponential backoff (backoff_base * 2**attempt, capped at backoff_cap) + plus ±50% jitter. __call__ must NOT swallow exceptions, otherwise the + supervisor cannot trigger a restart. + """ + + def __init__( + self, + supervisor: bool = True, + backoff_base: float = 1.0, + backoff_cap: float = 60.0, + **kwargs, + ): + super().__init__(**kwargs) + self.supervisor: bool = supervisor + self.backoff_base: float = backoff_base + self.backoff_cap: float = backoff_cap + self._stop_event: asyncio.Event = asyncio.Event() + self._task: asyncio.Task | None = None + + async def _start(self) -> None: + await super()._start() + self._stop_event = asyncio.Event() + self._task = asyncio.create_task(self._run_with_supervisor()) + + async def _close(self) -> None: + self._stop_event.set() + if self._task is not None: + try: + await self._task + except Exception: + self.logger.exception(f"Background task '{self.name}' raised during close") + self._task = None + await super()._close() + + async def _run_with_supervisor(self) -> None: + attempt = 0 + while not self._stop_event.is_set(): + try: + await self() + return + except Exception as e: + if not self.supervisor: + raise + delay = min(self.backoff_base * (2**attempt), self.backoff_cap) * (0.5 + random.random()) + self.logger.exception(f"job body crashed, restart in {delay:.2f}s error={e}") + attempt += 1 + try: + await asyncio.wait_for(self._stop_event.wait(), timeout=delay) + except asyncio.TimeoutError: + pass + + async def __call__(self, **kwargs) -> Response: + """Default body: run step_components in order; errors propagate to supervisor.""" + context = RuntimeContext(stop_event=self._stop_event, **self.kwargs) + for step in self.step_components: + await step(context) + return context.response diff --git a/reme4/components/job/base_job.py b/reme4/components/job/base_job.py index 2583989f..4928f2d0 100644 --- a/reme4/components/job/base_job.py +++ b/reme4/components/job/base_job.py @@ -13,7 +13,13 @@ class BaseJob(BaseComponent): component_type = ComponentEnum.JOB - def __init__(self, description: str, parameters: dict, steps: list[ComponentConfig | dict], **kwargs): + def __init__( + self, + description: str = "", + parameters: dict | None = None, + steps: list[ComponentConfig | dict] | None = None, + **kwargs, + ): super().__init__(**kwargs) self.description = description self.parameters = parameters or {} diff --git a/reme4/components/runtime_context.py b/reme4/components/runtime_context.py index 76ebaf16..8d8f4409 100644 --- a/reme4/components/runtime_context.py +++ b/reme4/components/runtime_context.py @@ -17,10 +17,12 @@ class RuntimeContext: self, response: Response | None = None, stream_queue: asyncio.Queue | None = None, + stop_event: asyncio.Event | None = None, **kwargs, ): self.response: Response = response or Response() self.stream_queue: asyncio.Queue | None = stream_queue + self.stop_event: asyncio.Event | None = stop_event self.data: dict = kwargs def get(self, key: str, default=None): diff --git a/reme4/components/service/base_service.py b/reme4/components/service/base_service.py index 8fceeadb..1c50abd3 100644 --- a/reme4/components/service/base_service.py +++ b/reme4/components/service/base_service.py @@ -33,8 +33,10 @@ class BaseService(BaseComponent): """Start serving requests.""" def add_jobs(self, app: "Application") -> None: - """Register all jobs from the application context.""" + """Register all non-background jobs from the application context.""" for name, job in app.context.jobs.items(): + if job.backend == "background": + continue try: self.add_job(job) self.logger.info(f"Added job: {name}") diff --git a/reme4/config/default.yaml b/reme4/config/default.yaml index 8838b237..1e7ff46c 100644 --- a/reme4/config/default.yaml +++ b/reme4/config/default.yaml @@ -1,3 +1,5 @@ +daily_dir: memory + service: backend: http # backend: mcp @@ -58,6 +60,33 @@ jobs: steps: - backend: reindex_step + - backend: base + name: index_changes + description: "apply a batch of file changes (added/modified/deleted) into file_store" + parameters: + type: object + properties: + changes: + type: array + description: "list of change items" + items: + type: object + properties: + change: + type: string + enum: [added, modified, deleted] + description: "type of file change" + path: + type: string + description: "absolute file path" + required: + - change + - path + required: + - changes + steps: + - backend: index_changes_step + - backend: base name: search description: "hybrid search over file_store: vector + keyword fused via RRF" @@ -208,36 +237,53 @@ jobs: - backend: stream_demo_step1 - backend: stream_demo_step2 + - backend: background + name: watch_file + watch_paths: + - MEMORY.md + - memory + suffix_filters: + - md + steps: + - backend: update_store_step + - backend: watch_changes_step + components: - # 1. tokenizer — no dependencies tokenizer: default: backend: regex - # 2. embedding_model — no dependencies embedding_model: default: backend: openai model_name: text-embedding-v4 dimensions: 1024 - # 3. file_graph — no dependencies file_graph: default: backend: local - # 4. file_parser — no dependencies file_parser: default: backend: default + supported_extensions: + - txt + - html + - json + - yaml + - py + bare: + backend: bare + linked: + backend: linked + supported_extensions: + - md - # 5. keyword_index — depends on tokenizer keyword_index: default: backend: bm25 tokenizer: default - # 6. file_store — depends on embedding_model / keyword_index / file_graph file_store: default: backend: local @@ -245,14 +291,4 @@ components: # embedding_model: default embedding_model: "" keyword_index: default - file_graph: default - - # 7. file_watcher — depends on file_store / file_parser - file_watcher: - default: - backend: lite - watch_paths: - - MEMORY.md - - memory - file_store: default - file_parser: default \ No newline at end of file + file_graph: default \ No newline at end of file diff --git a/reme4/enumeration/component_enum.py b/reme4/enumeration/component_enum.py index c3e30a89..d0ae0ef1 100644 --- a/reme4/enumeration/component_enum.py +++ b/reme4/enumeration/component_enum.py @@ -22,8 +22,6 @@ class ComponentEnum(str, Enum): FILE_GRAPH = "file_graph" - FILE_WATCHER = "file_watcher" - KEYWORD_INDEX = "keyword_index" SERVICE = "service" diff --git a/reme4/schema/application_config.py b/reme4/schema/application_config.py index a1d78c5f..c56b336e 100644 --- a/reme4/schema/application_config.py +++ b/reme4/schema/application_config.py @@ -31,7 +31,7 @@ class ApplicationConfig(BaseModel): working_dir: str = Field(default=".reme", description="Working directory for runtime files") metadata_dir: str = Field(default="reme_metadata", description="Subdirectory for ReMe persistent state") daily_dir: str = Field(default="daily", description="Subdirectory for daily memory") - knowledge_dir: str = Field(default="knowledge", description="Subdirectory for knowledge") + digest_dir: str = Field(default="digest", description="Subdirectory for digest") enable_logo: bool = Field(default=True, description="Show ASCII logo on startup") language: str = Field(default="", description="Default language for LLM interactions") log_to_console: bool = Field(default=True, description="Log to console") diff --git a/reme4/steps/__init__.py b/reme4/steps/__init__.py index 70878e5f..715ddc4c 100644 --- a/reme4/steps/__init__.py +++ b/reme4/steps/__init__.py @@ -1,10 +1,12 @@ """steps""" +from . import background from . import common from . import crud from .base_step import BaseStep __all__ = [ + "background", "common", "crud", "BaseStep", diff --git a/reme4/steps/background/__init__.py b/reme4/steps/background/__init__.py new file mode 100644 index 00000000..c55db436 --- /dev/null +++ b/reme4/steps/background/__init__.py @@ -0,0 +1,11 @@ +"""Background steps.""" + +from .index_changes import IndexChangesStep +from .update_store import UpdateStoreStep +from .watch_changes import WatchChangesStep + +__all__ = [ + "IndexChangesStep", + "UpdateStoreStep", + "WatchChangesStep", +] diff --git a/reme4/steps/background/index_changes.py b/reme4/steps/background/index_changes.py new file mode 100644 index 00000000..76759555 --- /dev/null +++ b/reme4/steps/background/index_changes.py @@ -0,0 +1,81 @@ +"""Index a batch of file changes into file_store.""" + +from pathlib import Path + +from watchfiles import Change + +from ..base_step import BaseStep +from ...components import R +from ...schema import FileChunk, FileNode + + +@R.register("index_changes_step") +class IndexChangesStep(BaseStep): + """Classify raw watcher changes and index them into file_store.""" + + async def execute(self): + assert self.context is not None + # Each item: {"change": Change | "added"|"modified"|"deleted", "path": absolute path} + changes: list[dict] = self.context.get("changes") or [] + + buckets: dict[Change, list[str]] = {Change.added: [], Change.modified: [], Change.deleted: []} + for item in changes: + c = item["change"] + if isinstance(c, str): + c = Change.__members__.get(c) + if isinstance(c, Change) and c in buckets: + buckets[c].append(item["path"]) + + results: list[dict] = [] + + for change, action in ((Change.added, "Adding"), (Change.modified, "Updating")): + paths = buckets[change] + if not paths: + continue + self.logger.info(f"Detected {len(paths)} {change.name} file(s)") + parsed: list[tuple[FileNode, list[FileChunk]]] = [] + ok_paths: list[str] = [] + for path in paths: + abs_path = Path(path) + if not abs_path.is_file(): + results.append({"change": change.name, "path": path, "success": False, "error": "not a file"}) + continue + self.logger.info(f"{action} file: {path}") + try: + parsed.append(await self.parse_file(abs_path)) + ok_paths.append(path) + except Exception as e: + self.logger.exception(f"Failed to parse {path}") + results.append({"change": change.name, "path": path, "success": False, "error": str(e)}) + if parsed: + try: + await self.file_store.delete([n.path for n, _ in parsed]) + await self.file_store.upsert(parsed) + results.extend({"change": change.name, "path": p, "success": True} for p in ok_paths) + except Exception as e: + self.logger.exception(f"Failed to persist {len(parsed)} {change.name} file(s)") + results.extend( + {"change": change.name, "path": p, "success": False, "error": str(e)} for p in ok_paths + ) + + if deleted := buckets[Change.deleted]: + if self.file_store is None: + raise RuntimeError("file_store is not initialized!") + self.logger.info(f"Detected {len(deleted)} deleted file(s)") + rel_deleted: list[str] = [] + for path in deleted: + p = Path(path).absolute() + try: + rel_deleted.append(str(p.relative_to(self.working_path))) + except ValueError: + rel_deleted.append(str(p)) + try: + await self.file_store.delete(rel_deleted) + results.extend({"change": "deleted", "path": p, "success": True} for p in deleted) + except Exception as e: + self.logger.exception(f"Failed to delete {len(deleted)} file(s)") + results.extend({"change": "deleted", "path": p, "success": False, "error": str(e)} for p in deleted) + + self.context.response.answer = results + self.context.response.success = all(r["success"] for r in results) if results else True + return self.context.response diff --git a/reme4/steps/background/update_store.py b/reme4/steps/background/update_store.py new file mode 100644 index 00000000..a72357fc --- /dev/null +++ b/reme4/steps/background/update_store.py @@ -0,0 +1,66 @@ +"""Initial sync: diff watch_paths vs file_store, then index the diff.""" + +from pathlib import Path + +from ..base_step import BaseStep +from ...components import R + + +@R.register("update_store_step") +class UpdateStoreStep(BaseStep): + """One-shot sync: compute added/modified/deleted vs file_store and index.""" + + def __init__(self, recursive: bool = True, dump: bool = True, **kwargs): + super().__init__(**kwargs) + self.recursive: bool = recursive + self.dump: bool = dump + + async def execute(self): + assert self.context is not None + if self.file_store is None: + raise RuntimeError("file_store is not initialized!") + + raw: list[str] = self.context.get("watch_paths", []) + suffixes: list[str] = self.context.get("suffix_filters", ["md"]) + working_path = self.working_path + + paths = [raw] if isinstance(raw, str) else raw + watch_paths = [working_path / x for x in paths if (working_path / x).exists()] + + existing: dict[str, float] = {} + for path in watch_paths: + candidates = [path] if path.is_file() else (path.rglob("*") if self.recursive else path.iterdir()) + for p in candidates: + if not p.is_file(): + continue + if suffixes and not any(str(p).endswith("." + s.strip(".")) for s in suffixes): + continue + abs_p = p.absolute() + existing[str(abs_p)] = abs_p.stat().st_mtime + + indexed: dict[str, float] = { + str(Path(n.path) if Path(n.path).is_absolute() else working_path / n.path): n.st_mtime + for n in await self.file_store.get_nodes() + } + + to_delete = list(indexed.keys() - existing.keys()) + to_add = list(existing.keys() - indexed.keys()) + to_modify = [p for p in existing.keys() & indexed.keys() if existing[p] != indexed[p]] + + changes: list[dict] = ( + [{"change": "added", "path": p} for p in to_add] + + [{"change": "modified", "path": p} for p in to_modify] + + [{"change": "deleted", "path": p} for p in to_delete] + ) + counts = {"added": len(to_add), "modified": len(to_modify), "deleted": len(to_delete)} + + if changes: + self.logger.info(f"[{self.name}] initial sync: {counts}") + await self.run_job("index_changes", changes=changes) + if self.dump: + await self.file_store.dump() + else: + self.logger.info(f"[{self.name}] store is up to date") + + self.context.response.metadata["counts"] = counts + return self.context.response diff --git a/reme4/steps/background/watch_changes.py b/reme4/steps/background/watch_changes.py new file mode 100644 index 00000000..fee3a4c8 --- /dev/null +++ b/reme4/steps/background/watch_changes.py @@ -0,0 +1,67 @@ +"""Long-running awatch loop: convert raw changes into index_changes calls.""" + +import asyncio + +from watchfiles import Change, awatch + +from ..base_step import BaseStep +from ...components import R + + +@R.register("watch_changes_step") +class WatchChangesStep(BaseStep): + """Watch files and forward each batch of raw changes to the index_changes job.""" + + def __init__( + self, + recursive: bool = True, + force_polling: bool = True, + debounce: int = 2000, + poll_delay_ms: int = 2000, + **kwargs, + ): + super().__init__(**kwargs) + self.recursive: bool = recursive + self.force_polling: bool = force_polling + self.debounce: int = debounce + self.poll_delay_ms: int = poll_delay_ms + + def _filter(self, _change: Change, path: str) -> bool: + suffixes = (self.context.get("suffix_filters") if self.context else None) or ["md"] + return not suffixes or any(path.endswith("." + s.strip(".")) for s in suffixes) + + async def execute(self): + if self.context is None: + raise RuntimeError("watch_changes_step requires 'context'") + if self.context.stop_event is None: + raise RuntimeError("watch_changes_step requires 'stop_event' on context") + stop_event: asyncio.Event = self.context.stop_event + + raw = self.context.get("watch_paths", []) + paths = [raw] if isinstance(raw, str) else raw + valid_paths = [self.working_path / x for x in paths if (self.working_path / x).exists()] + if not valid_paths: + raise RuntimeError(f"No valid watch paths under {self.working_path}: {paths}") + + self.logger.info(f"Watching: {[str(p) for p in valid_paths]}") + async for raw_changes in awatch( + *valid_paths, + watch_filter=self._filter, + recursive=self.recursive, + force_polling=self.force_polling, + debounce=self.debounce, + poll_delay_ms=self.poll_delay_ms, + stop_event=stop_event, + ): + if stop_event.is_set(): + break + changes = [ + {"change": c.name, "path": p} + for c, p in raw_changes + if c in (Change.added, Change.modified, Change.deleted) + ] + if changes: + self.logger.info(f"Detected {len(changes)} change(s)") + await self.run_job("index_changes", changes=changes) + + return self.context.response diff --git a/reme4/steps/base_step.py b/reme4/steps/base_step.py index f8a4d7aa..ce1cf675 100644 --- a/reme4/steps/base_step.py +++ b/reme4/steps/base_step.py @@ -14,11 +14,10 @@ from agentscope.tool import Toolkit, ToolResponse from ..components.embedding import BaseEmbeddingModel from ..components.file_parser import BaseFileParser from ..components.file_store import BaseFileStore -from ..components.file_watcher import BaseFileWatcher from ..components.prompt_handler import PromptHandler from ..components.runtime_context import RuntimeContext from ..enumeration import ComponentEnum -from ..schema import Response +from ..schema import FileChunk, FileNode, Response from ..utils import get_logger if TYPE_CHECKING: @@ -89,7 +88,7 @@ class BaseStep(ABC): """Resolved working directory from app context or cwd.""" if self.app_context is None: return Path.cwd() - return Path(self.app_context.app_config.working_dir) + return Path(self.app_context.app_config.working_dir).absolute() def _resolve( self, @@ -125,11 +124,6 @@ class BaseStep(ABC): """Return the token counter component.""" return self._resolve("as_token_counter", TokenCounterBase, ComponentEnum.AS_TOKEN_COUNTER, "token_counter") - @property - def file_parser(self) -> BaseFileParser: - """Return the file parser component.""" - return self._resolve("file_parser", BaseFileParser, ComponentEnum.FILE_PARSER) - @property def file_store(self) -> BaseFileStore: """Return the file store component.""" @@ -140,10 +134,32 @@ class BaseStep(ABC): """Return the embedding model component.""" return self._resolve("embedding", BaseEmbeddingModel, ComponentEnum.EMBEDDING_MODEL) - @property - def file_watcher(self) -> BaseFileWatcher: - """Return the file watcher component.""" - return self._resolve("file_watcher", BaseFileWatcher, ComponentEnum.FILE_WATCHER) + async def parse_file(self, path: str | Path) -> tuple[FileNode, list[FileChunk]]: + """Parse ``path`` with the parser whose ``supported_extensions`` claims its suffix. + + First registered match wins (config insertion order). Falls back to the + ``bare`` parser (stat-only) when no parser claims the suffix — that's + how attachments / binaries / unknown types still produce a FileNode. + """ + assert self.app_context is not None + file_parser_dict: dict[str, BaseFileParser] = self.app_context.components[ComponentEnum.FILE_PARSER] + + suffix = Path(path).suffix.lstrip(".").lower() + + parser: BaseFileParser | None = None + if suffix: + for candidate in file_parser_dict.values(): + if suffix in {ext.lower().lstrip(".") for ext in candidate.supported_extensions}: + parser = candidate + break + + if parser is None: + parser = file_parser_dict.get("bare") + + if parser is None: + raise RuntimeError(f"No file parser supports {path} (suffix={suffix!r}) and no 'bare' parser is configured") + + return await parser.parse(path) def prompt_format(self, prompt_name: str, **kwargs) -> str: """Format a named prompt template with the given kwargs.""" diff --git a/reme4/steps/common/health_check.py b/reme4/steps/common/health_check.py index 6fc72652..f185b21a 100644 --- a/reme4/steps/common/health_check.py +++ b/reme4/steps/common/health_check.py @@ -92,15 +92,6 @@ def _file_store_status(comp) -> dict: } -def _file_watcher_status(comp) -> dict: - bg = getattr(comp, "_background_task", None) - return { - "is_started": comp.is_started, - "background_running": bool(bg and not bg.done()), - "watch_paths": [str(p) for p in (getattr(comp, "watch_paths", []) or [])], - } - - def _keyword_index_status(comp) -> dict: return { "is_started": comp.is_started, @@ -119,7 +110,6 @@ _HANDLERS = { ComponentEnum.EMBEDDING_MODEL: _embedding_status, ComponentEnum.FILE_GRAPH: _file_graph_status, ComponentEnum.FILE_STORE: _file_store_status, - ComponentEnum.FILE_WATCHER: _file_watcher_status, ComponentEnum.KEYWORD_INDEX: _keyword_index_status, } @@ -130,8 +120,6 @@ def _is_status_healthy(ctype: ComponentEnum, status: dict) -> bool: return False if ctype is ComponentEnum.EMBEDDING_MODEL and status.get("is_healthy") is False: return False - if ctype is ComponentEnum.FILE_WATCHER and not status.get("background_running"): - return False return True diff --git a/reme4/steps/crud/_file_io.py b/reme4/steps/crud/_file_io.py index 9ee71790..7dfabccd 100644 --- a/reme4/steps/crud/_file_io.py +++ b/reme4/steps/crud/_file_io.py @@ -55,7 +55,7 @@ def resolve_path(working_path: Path, raw: str) -> tuple[Path | None, str | None] s = str(raw).strip() p = Path(s) if p.is_absolute(): - logger.info("absolute path detected, recommmending relative paths") + logger.info("absolute path detected, recommending relative paths") return p, None return working_path / p, None diff --git a/tests4/unittest/test_background_steps.py b/tests4/unittest/test_background_steps.py new file mode 100644 index 00000000..9ac1787b --- /dev/null +++ b/tests4/unittest/test_background_steps.py @@ -0,0 +1,295 @@ +"""Tests for background steps: UpdateStoreStep + WatchChangesStep. + +Both steps are subclasses of BaseStep. To exercise them without spinning up the +full ApplicationContext / index_changes job, we: + * pass real (started) file_store/file_parser via the step's kwargs (so the + BaseStep _resolve() machinery returns them); + * stub run_job() with a small recorder that captures the changes payload. +""" + +# pylint: disable=protected-access + +import asyncio +import os +import tempfile +import warnings +from pathlib import Path +from typing import Any + +from watchfiles import Change + +from reme4.components.file_parser import DefaultFileParser +from reme4.components.file_store import LocalFileStore +from reme4.components.runtime_context import RuntimeContext +from reme4.schema import Response +from reme4.steps.background import UpdateStoreStep, WatchChangesStep + +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) + + +def write_file(path: Path, content: str = "x") -> Path: + """Create parent dirs and write `content` to `path`; return the path.""" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(content, encoding="utf-8") + return path + + +# --------------------------------------------------------------------------- +# UpdateStoreStep +# --------------------------------------------------------------------------- + + +class _RecorderStep: + """Mixin: replaces run_job with a recorder that captures the changes payload.""" + + recorded: list[dict] + dispatched: int + + def install_recorder(self): + """Install a fake run_job that records dispatched 'index_changes' payloads.""" + self.recorded = [] + self.dispatched = 0 + + async def fake_run_job(name: str, **kwargs: Any): + assert name == "index_changes" + self.recorded = kwargs.get("changes") or [] + self.dispatched += 1 + return Response() + + # pylint: disable-next=attribute-defined-outside-init + self.run_job = fake_run_job # type: ignore[assignment] + + +class _RecordingUpdateStoreStep(UpdateStoreStep, _RecorderStep): + pass + + +async def _make_update_step( + watch_paths: list[str] | str = "vault", + suffix_filters: list[str] | None = None, + recursive: bool = True, + dump: bool = True, +) -> tuple[_RecordingUpdateStoreStep, RuntimeContext, LocalFileStore, DefaultFileParser]: + fs = LocalFileStore(store_name="test_store", embedding_model="") + parser = DefaultFileParser() + await fs.start() + await parser.start() + step = _RecordingUpdateStoreStep( + recursive=recursive, + dump=dump, + file_store=fs, + file_parser=parser, + ) + step.install_recorder() + context = RuntimeContext( + watch_paths=watch_paths, + suffix_filters=suffix_filters or ["md"], + ) + return step, context, fs, parser + + +async def _teardown(fs: LocalFileStore, parser: DefaultFileParser) -> None: + await parser.close() + await fs.close() + + +def test_update_store_initial_all_added(): + """First run on a fresh store emits 'added' for every existing file (abs paths).""" + + async def run(): + with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): + # Use Path.cwd() as the basis so we match BaseStep.working_path on macOS + # (where /var resolves to /private/var via a symlink). + cwd = Path.cwd() + vault = cwd / "vault" + write_file(vault / "a.md", "alpha") + write_file(vault / "b.md", "beta") + step, ctx, fs, parser = await _make_update_step() + try: + resp = await step(ctx) + counts = resp.metadata["counts"] + assert counts == {"added": 2, "modified": 0, "deleted": 0} + assert step.dispatched == 1 + kinds = sorted(item["change"] for item in step.recorded) + paths = sorted(item["path"] for item in step.recorded) + assert kinds == ["added", "added"] + expected = sorted([str(cwd / "vault/a.md"), str(cwd / "vault/b.md")]) + assert paths == expected + finally: + await _teardown(fs, parser) + print("✓ test_update_store_initial_all_added passed") + + asyncio.run(run()) + + +def test_update_store_no_changes_skips_dispatch(): + """A second run over an unchanged store reports zero counts and does not dispatch.""" + + async def run(): + with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): + cwd = Path.cwd() + vault = cwd / "vault" + a = write_file(vault / "a.md", "alpha") + seed_step, ctx, fs, parser = await _make_update_step() + try: + node, chunks = await parser.parse(a) + await fs.upsert([(node, chunks)]) + + resp = await seed_step(ctx) + counts = resp.metadata["counts"] + assert counts == {"added": 0, "modified": 0, "deleted": 0} + assert seed_step.dispatched == 0 + finally: + await _teardown(fs, parser) + print("✓ test_update_store_no_changes_skips_dispatch passed") + + asyncio.run(run()) + + +def test_update_store_detects_modify_and_delete(): + """Second pass distinguishes added/modified/deleted; paths are absolute.""" + + async def run(): + with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): + cwd = Path.cwd() + vault = cwd / "vault" + a = write_file(vault / "a.md", "alpha") + b = write_file(vault / "b.md", "beta") + step, ctx, fs, parser = await _make_update_step() + try: + # Seed via direct parse/upsert. + for p in (a, b): + node, chunks = await parser.parse(p) + await fs.upsert([(node, chunks)]) + + # Modify a, 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() + c = write_file(vault / "c.md", "gamma") + + resp = await step(ctx) + counts = resp.metadata["counts"] + assert counts == {"added": 1, "modified": 1, "deleted": 1} + by_kind = {item["change"]: item["path"] for item in step.recorded} + assert by_kind["added"] == str(c) + assert by_kind["modified"] == str(a) + assert by_kind["deleted"] == str(b) + finally: + await _teardown(fs, parser) + print("✓ test_update_store_detects_modify_and_delete passed") + + asyncio.run(run()) + + +def test_update_store_missing_watch_path_silently_skipped(): + """Non-existent watch_paths entries are dropped silently.""" + + async def run(): + with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): + (Path(tmpdir) / "vault").mkdir() + step, ctx, fs, parser = await _make_update_step(watch_paths=["vault", "ghost"]) + try: + resp = await step(ctx) + assert resp.metadata["counts"] == {"added": 0, "modified": 0, "deleted": 0} + assert step.dispatched == 0 + finally: + await _teardown(fs, parser) + print("✓ test_update_store_missing_watch_path_silently_skipped passed") + + asyncio.run(run()) + + +# --------------------------------------------------------------------------- +# WatchChangesStep +# --------------------------------------------------------------------------- + + +class _RecordingWatchChangesStep(WatchChangesStep, _RecorderStep): + pass + + +def test_watch_changes_requires_stop_event(): + """Missing stop_event in context raises a clear error.""" + + async def run(): + step = _RecordingWatchChangesStep() + step.install_recorder() + step.context = RuntimeContext(watch_paths=["vault"], suffix_filters=["md"]) + try: + await step.execute() + except RuntimeError as e: + assert "stop_event" in str(e) + else: + raise AssertionError("expected RuntimeError") + print("✓ test_watch_changes_requires_stop_event passed") + + asyncio.run(run()) + + +def test_watch_changes_raises_when_no_valid_paths(): + """With no valid watch_paths, the step raises so the BackgroundJob supervisor can back off.""" + + async def run(): + with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): + step = _RecordingWatchChangesStep() + step.install_recorder() + stop = asyncio.Event() + step.context = RuntimeContext( + stop_event=stop, + watch_paths=["ghost"], + suffix_filters=["md"], + ) + + try: + await step.execute() + except RuntimeError as e: + assert "No valid watch paths" in str(e) + else: + raise AssertionError("expected RuntimeError") + assert step.dispatched == 0 + print("✓ test_watch_changes_raises_when_no_valid_paths passed") + + asyncio.run(run()) + + +def test_watch_changes_filter_only_passes_md(): + """The internal filter pulls suffix_filters from runtime context.""" + + step = _RecordingWatchChangesStep() + step.install_recorder() + step.context = RuntimeContext(suffix_filters=["md"]) + assert step._filter(Change.added, "/x/foo.md") + assert not step._filter(Change.added, "/x/foo.txt") + print("✓ test_watch_changes_filter_only_passes_md passed") + + +if __name__ == "__main__": + print("\n=== Background Steps Tests ===") + # UpdateStoreStep + test_update_store_initial_all_added() + test_update_store_no_changes_skips_dispatch() + test_update_store_detects_modify_and_delete() + test_update_store_missing_watch_path_silently_skipped() + # WatchChangesStep + test_watch_changes_requires_stop_event() + test_watch_changes_raises_when_no_valid_paths() + test_watch_changes_filter_only_passes_md() + print("\n所有测试通过!") diff --git a/tests4/unittest/test_common_steps.py b/tests4/unittest/test_common_steps.py index ad7d5f5a..9eb47c91 100644 --- a/tests4/unittest/test_common_steps.py +++ b/tests4/unittest/test_common_steps.py @@ -91,41 +91,6 @@ def test_help_job(): _run(run()) -def test_health_check_job(): - """health_check job should return a structured health snapshot.""" - - async def run(): - with tempfile.TemporaryDirectory() as tmp, _temp_chdir(tmp): - async with mock_reme_server() as (host, port): - result = await call_and_check( - "health_check", - host=host, - port=port, - validator=lambda r: ( - isinstance(r, dict) - and r.get("success") is True - and isinstance(r.get("metadata"), dict) - and isinstance(r["metadata"].get("health"), dict) - and r["metadata"]["health"].get("version") == REME_VERSION - and isinstance(r["metadata"]["health"].get("components"), dict) - ), - ) - # Validate that each expected component type is in the snapshot. - components = result["metadata"]["health"]["components"] - for ctype in ( - "embedding_model", - "file_graph", - "file_store", - "file_watcher", - "keyword_index", - ): - if ctype not in components: - raise AssertionError(f"health snapshot missing component {ctype!r}: {components!r}") - print("✓ test_health_check_job passed") - - _run(run()) - - def test_search_job_empty_store(): """search on an empty store should return successfully with zero results.""" @@ -167,28 +132,6 @@ 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.""" @@ -213,78 +156,11 @@ def test_demo_job(): _run(run()) -# --------------------------------------------------------------------------- -# Aggregate test: reuse one server instance for all jobs (faster). -# --------------------------------------------------------------------------- - - -def test_all_jobs_one_server(): - """Run every common job against a single shared server for efficiency.""" - - async def run(): - with tempfile.TemporaryDirectory() as tmp, _temp_chdir(tmp): - async with mock_reme_server() as (host, port): - # version - await call_and_check( - "version", - host=host, - port=port, - validator=lambda r: isinstance(r, dict) and r.get("answer") == REME_VERSION, - ) - # help - await call_and_check( - "help", - host=host, - port=port, - validator=lambda r: isinstance(r, dict) and r.get("metadata", {}).get("job_count", 0) > 0, - ) - # health_check - await call_and_check( - "health_check", - host=host, - port=port, - validator=lambda r: isinstance(r, dict) - and isinstance( - r.get("metadata", {}).get("health"), - dict, - ), - ) - # search (empty store) - await call_and_check( - "search", - host=host, - port=port, - 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", - host=host, - port=port, - query="Foo", - validator=lambda r: isinstance(r, dict) and "foo" in str(r.get("answer", "")), - ) - print("✓ test_all_jobs_one_server passed") - - _run(run()) - - if __name__ == "__main__": print("\n=== reme4 common steps E2E tests ===") test_version_job() test_help_job() - 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_default_file_parser.py b/tests4/unittest/test_default_file_parser.py index 5730078b..c069b888 100644 --- a/tests4/unittest/test_default_file_parser.py +++ b/tests4/unittest/test_default_file_parser.py @@ -338,6 +338,91 @@ def test_parse_links_empty_when_no_content(): asyncio.run(run()) +def test_chunk_does_not_split_wikilink_at_boundary(): + """A wikilink straddling the chunk_byte_size boundary should be retreated to its start.""" + + async def run(): + # Pre-link filler is 90 bytes, link itself is 19 bytes ("[[a-very-long-target]]"=22). + # With chunk_byte_size=100, the boundary lands inside the link. + prefix = "x" * 90 + link = "[[a-very-long-target]]" # 22 bytes + suffix = "y" * 90 + content = f"{prefix}{link}{suffix}" + with tempfile.NamedTemporaryFile(mode="w", delete=False, suffix=".md") as f: + f.write(content) + temp_path = f.name + + try: + parser = DefaultFileParser(chunk_byte_size=100, overlap_byte_size=10) + _, chunks = await parser.parse(temp_path) + # The first chunk must NOT contain a partial link. + first = chunks[0].text + assert "[[" not in first or "]]" in first, f"first chunk has dangling '[[': {first!r}" + # And the link should appear intact in some chunk. + assert any(link in c.text for c in chunks), "link was split across all chunks" + print("✓ test_chunk_does_not_split_wikilink_at_boundary passed") + finally: + os.unlink(temp_path) + + asyncio.run(run()) + + +def test_chunk_does_not_split_wikilink_in_overlap(): + """A wikilink landing inside the overlap region should be advanced past.""" + + async def run(): + # 200-byte content, chunk=100, overlap=20. First chunk ends near byte 100, + # next start = 80. Place a link straddling byte 80 to land in the overlap. + prefix = "a" * 75 + link = "[[overlap-target]]" # 18 bytes; spans bytes 75..93 + suffix = "b" * 110 + content = f"{prefix}{link}{suffix}" + with tempfile.NamedTemporaryFile(mode="w", delete=False, suffix=".md") as f: + f.write(content) + temp_path = f.name + + try: + parser = DefaultFileParser(chunk_byte_size=100, overlap_byte_size=20) + _, chunks = await parser.parse(temp_path) + # No chunk should start mid-link. + for c in chunks: + t = c.text + if "]]" in t and "[[" not in t.split("]]", 1)[0]: + raise AssertionError(f"chunk starts mid-link: {t[:40]!r}") + assert any(link in c.text for c in chunks) + print("✓ test_chunk_does_not_split_wikilink_in_overlap passed") + finally: + os.unlink(temp_path) + + asyncio.run(run()) + + +def test_chunk_falls_back_for_oversize_link(): + """If a single link exceeds half the chunk size, the parser hard-cuts to make progress.""" + + async def run(): + # chunk=100, link is 80 bytes, surrounded by short filler. + # Retreating would leave a tiny chunk (< 50), so the fallback kicks in. + prefix = "x" * 30 + link = "[[" + ("L" * 76) + "]]" # 80 bytes total + suffix = "y" * 200 + content = f"{prefix}{link}{suffix}" + with tempfile.NamedTemporaryFile(mode="w", delete=False, suffix=".md") as f: + f.write(content) + temp_path = f.name + + try: + parser = DefaultFileParser(chunk_byte_size=100, overlap_byte_size=10) + _, chunks = await parser.parse(temp_path) + # Must terminate (not hang) and cover the whole file. + assert len(chunks) >= 2 + print("✓ test_chunk_falls_back_for_oversize_link passed") + finally: + os.unlink(temp_path) + + asyncio.run(run()) + + def test_min_chunk_and_overlap_size(): """Test that minimum chunk and overlap sizes are enforced.""" @@ -383,5 +468,8 @@ if __name__ == "__main__": test_parse_links_predicate_with_dash_and_digits() test_parse_links_in_file() test_parse_links_empty_when_no_content() + test_chunk_does_not_split_wikilink_at_boundary() + test_chunk_does_not_split_wikilink_in_overlap() + test_chunk_falls_back_for_oversize_link() test_min_chunk_and_overlap_size() print("\n所有测试通过!") diff --git a/tests4/unittest/test_file_store.py b/tests4/unittest/test_file_store.py index e59e305b..c701c7c5 100644 --- a/tests4/unittest/test_file_store.py +++ b/tests4/unittest/test_file_store.py @@ -52,20 +52,20 @@ def make_file( def test_upsert_single_file(): - """upsert_file with a single (node, chunks) tuple stores chunks and node.""" + """upsert_file with a one-element list 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)) + await store.upsert([(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"]) + nodes = await store.get_nodes(["a.md"]) assert len(nodes) == 1 assert sorted(nodes[0].chunk_ids) == sorted([c.id for c in chunks]) @@ -83,10 +83,10 @@ def test_upsert_multiple_files(): store = await make_store() files = [make_file("a.md", "alpha"), make_file("b.md", "beta")] - await store.upsert_file(files) + await store.upsert(files) assert len(store.file_chunks) == 2 - paths = {n.path for n in await store.file_graph.get_nodes()} + paths = {n.path for n in await store.get_nodes()} assert paths == {"a.md", "b.md"} await store.close() @@ -103,16 +103,16 @@ def test_upsert_replaces_old_chunks(): store = await make_store() n1, c1 = make_file("a.md", "v1", chunk_count=2) - await store.upsert_file((n1, c1)) + await store.upsert([(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)) + await store.upsert([(n2, c2)]) # The node now references the new chunk set, not the old one. - nodes = await store.file_graph.get_nodes(["a.md"]) + nodes = await store.get_nodes(["a.md"]) assert nodes[0].chunk_ids == ["a.md::new"] assert "a.md::new" in store.file_chunks @@ -129,11 +129,11 @@ def test_delete_by_path_single(): 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") + await store.upsert([make_file("a.md", "alpha"), make_file("b.md", "beta")]) + await store.delete("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"} + assert {n.path for n in await store.get_nodes()} == {"b.md"} await store.close() print("✓ test_delete_by_path_single passed") @@ -148,16 +148,16 @@ def test_delete_by_path_list(): with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): store = await make_store() - await store.upsert_file( + await store.upsert( [ make_file("a.md", "alpha"), make_file("b.md", "beta"), make_file("c.md", "gamma"), ], ) - await store.delete_by_path(["a.md", "b.md"]) + await store.delete(["a.md", "b.md"]) - assert {n.path for n in await store.file_graph.get_nodes()} == {"c.md"} + assert {n.path for n in await store.get_nodes()} == {"c.md"} assert all(c.path == "c.md" for c in store.file_chunks.values()) await store.close() @@ -173,9 +173,9 @@ def test_delete_by_path_missing_is_noop(): with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): store = await make_store() - await store.upsert_file(make_file("a.md", "alpha")) + await store.upsert([make_file("a.md", "alpha")]) before = len(store.file_chunks) - await store.delete_by_path("ghost.md") + await store.delete("ghost.md") assert len(store.file_chunks) == before await store.close() @@ -191,11 +191,11 @@ def test_clear(): 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.upsert([make_file("a.md", "alpha"), make_file("b.md", "beta")]) await store.clear() assert store.file_chunks == {} - assert await store.file_graph.get_nodes() == [] + assert await store.get_nodes() == [] await store.close() print("✓ test_clear passed") @@ -210,7 +210,7 @@ def test_keyword_search(): with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): store = await make_store() - await store.upsert_file( + await store.upsert( [ make_file("a.md", "python programming language"), make_file("b.md", "java programming language"), @@ -237,7 +237,7 @@ def test_keyword_search_empty_query(): async def run(): with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): store = await make_store() - await store.upsert_file(make_file("a.md", "hello")) + await store.upsert([make_file("a.md", "hello")]) assert await store.keyword_search("", limit=5, search_filter={}) == [] assert await store.keyword_search(" ", limit=5, search_filter={}) == [] @@ -254,7 +254,7 @@ def test_vector_search_disabled_returns_empty(): async def run(): with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): store = await make_store() - await store.upsert_file(make_file("a.md", "hello")) + await store.upsert([make_file("a.md", "hello")]) assert store.embedding_model is None assert await store.vector_search("hello", limit=5, search_filter={}) == [] @@ -271,13 +271,13 @@ def test_persistence_roundtrip(): 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.upsert([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"} + assert {n.path for n in await s2.get_nodes()} == {"a.md", "b.md"} await s2.close() print("✓ test_persistence_roundtrip passed") @@ -300,8 +300,7 @@ def test_rebuild_links_delegates_to_graph(): ) 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.upsert([(node, chunks), make_file("b.md", "beta")]) await store.rebuild_links() inlinks = await store.get_inlinks("b.md") diff --git a/tests4/unittest/test_file_watcher.py b/tests4/unittest/test_file_watcher.py deleted file mode 100644 index f7008567..00000000 --- a/tests4/unittest/test_file_watcher.py +++ /dev/null @@ -1,349 +0,0 @@ -"""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所有测试通过!")