mirror of
https://github.com/agentscope-ai/ReMe.git
synced 2026-09-06 08:16:00 +00:00
### 1. Agent Wrapper(统一 Agent 后端抽象) - **`base_agent_wrapper.py`**:`reply()` 返回值从 `tuple[str, Any]` 改为 `dict`(含 `session_id` / `last_message` / `result` / 可选 `structured_output`);`reply_stream()` 改为产出统一的 `StreamChunk`。废弃 `add_tools()`,改为 `add_job_tools(names: list[str])`(按名解析 BaseJob)与 `add_skills()`;新增 `_resolve_job_tools()`、`_merged_kwargs()`、`_chunk()` 辅助方法及 `project_path` / `project_skills_root` 属性。 - **`as_agent_wrapper.py`(AgentScope 后端)**: - 会话持久化重写:`session_path` 落地到 `<vault>/<session_dir>/agentscope/`,`_load_state` 支持 `resume` / `session_id` / `fork_session`,并做 UUID 校验(`_validate_session_id`);`_cleanup_expired_sessions` 按天数清理过期会话。 - 新增内置工具集(`BypassAnalysisBash` + Edit/Glob/Grep/Read/Write),`BypassAnalysisBash` 绕过 AgentScope 自带 Bash 静态分析以让 permission_mode 生效;`_resolve_skills()` 把配置的 skill 暴露给后端,`_load_tool_env()` 注入项目 `.env`。 - `_event_to_chunk()` 把 20+ 种 AgentScope 事件(Reply/Text/Thinking/Data/ToolCall/ToolResult/ModelCall/ExceedMaxIters)归一化为 `StreamChunk`。 - **`cc_agent_wrapper.py`(Claude Code SDK 后端,+551 行)**: - 新增 `_CcFileSessionStore`:基于 vault 的文件型会话存储,实现 append(按 uuid 去重)/ load / list / delete / list_subkeys,并对路径做 `_safe_parts` + `resolve()` 防越界校验。 - `_build_options()`:统一构建 `ClaudeAgentOptions`,处理 skills、disallowed_tools(默认禁 `WebSearch`)、`.env` 注入、Claude Code 的 API 凭据解析(`_claude_code_api_env`,多级 base_url/api_key 回退)、`CLAUDE_CONFIG_DIR` 设置、skill 目录软链接(`_ensure_claude_skill_dir`)。 - `_raw_event_to_chunk()` / `_message_content_to_chunks()`:把 Anthropic 流式事件(message_start/delta/stop、content_block_*)与 SDK 消息块(AssistantMessage/UserMessage/ResultMessage/RateLimitEvent)转换为统一 `StreamChunk`;跟踪 block_id/block_type/tool_call_name 做关联;处理尾部 `"success"` 误报异常的吞掉逻辑。 ### 2. 统一流式协议(StreamChunk / ChunkEnum) - **`stream_chunk.py`**:`StreamChunk` 扩展为承载 AS + CC 双后端完整信息的统一结构,新增 `session_id` / `block_id` / `tool_call_id` / `tool_call_name` / `media_type` / `input_tokens` / `output_tokens` 等字段,纯文本流仍保持轻量。 - **`chunk_enum.py`**:补全生命周期标记 `REPLY_START` / `REPLY_END`,并文档化两套后端事件 → ChunkEnum 的映射。 ### 3. Index 模块重构(变化批次化 + dispatch) - 新增 `_change_batch.py`:`coalesce_changes()` 把同路径多次事件折叠为最终状态(结合 path 存在性判定),`bucket_changes()` 按 watchfiles.Change 分桶。 - 新增 `init_changes.py`(`InitChangesStep`):一次性扫描,对比 file_store / file_catalog 已索引节点计算 added/modified/deleted,写入 `context["changes"]` 后 dispatch。 - 新增 `update_changes.py`:抽象基类 `ChangeApplyStep` 统一 added/modified/deleted 处理与错误收集;`UpdateCatalogStep`(写 file_catalog)、`UpdateIndexStep`(写 file_store,含按后缀解析 chunker)。 - **`watch_changes.py`**:改用 `dispatch_step_specs`(基类提供的 `dispatch_steps()`),每批先 `coalesce_changes` 再 dispatch;默认参数调整(debounce 5000ms / step 1000ms / poll 5000ms)并暴露常量。 - 删除旧步骤:`clear_and_scan` / `foreach_dispatch` / `scan_changes` / `update_catalog`(旧) / `update_index`(旧);`clear_store.py` 取代 clear_and_scan。 ### 4. Evolve / Dream 模块(拆分为多步 pipeline) - 删除旧的单体 `auto_dream.py` / `dream.py` / `dream.yaml`,新增 `dream/` 子包,按 5 个步骤组织: - **`extract.py`**:扫描当日 day-index + daily 笔记,对比 file_catalog 找出 changed/deleted,调用 LLM 全局抽取 `units`(procedure/personal/wiki 三桶)与 `topics`,路径与桶做清洗/路由。 - **`integrate.py`**:逐个 unit 调用 LLM 写入 digest,结构化输出 `IntegrateOutcome`(CREATE/CORROBORATE/REFINE/CORRECT),失败 unit/路径收集回写。 - **`topics.py`**:写 `daily/<date>/interests.yaml`,结合当天已有 + 近 N 天做去重(`normalize_topic`),可走 LLM 或纯规则去重两条路径。 - **`proactive.py`**:读取当日 `interests.yaml`,作为主动推荐话题的入口。 - **`finish.py`**:把变更路径落盘到 dream file_catalog(checkpoint),渲染最终汇总摘要。 - 新增 `schema.py`(`DreamState` 等跨步骤共享状态与结构化输出模型)与 `utils.py`(状态存取、扫描打包、YAML 读写、结构化回复解析等公共函数)。 - `evolve/__init__.py` 导出全部新 step。 ### 5. auto_memory / auto_resource(适配新 Agent API) - **`auto_memory.py`**:会话路径迁移到 `<session_dir>/dialog/<session_id>.jsonl`;改用 `job_tools`;新增 `source_conversation` frontmatter 反向链接(`_session_link`);执行后刷新 day 索引(`refresh_day_index`),并对 session_id 做合法性校验。 - **`auto_resource.py`**:资源改用「同名 daily note」方案(`_compute_note_stem` 取文件 stem);批量处理 `changes: list[dict]`(`_handle_change` 逐项处理,返回逐项结果摘要);agent 会话 id 用稳定的 `uuid5`;同样刷新 day 索引。 ### 6. BaseStep 基类增强 - 新增 `dispatch_steps` / `dispatch_step_specs` 机制:`_resolve_dispatch_step()` 支持字符串或 dict 形式的 step spec,`dispatch_steps()` 复用当前 context 调用下游 step。 - 新增 `config_value()`:按 key 取 app config,缺失时回退 `ApplicationConfig` 默认值。 - 小幅清理:`language` 初始化、`copy()`、`Ref.__init__` 签名精简。 ### 7. Components 改动 - **`file_store/local_file_store.py`**:持久化改用 zstd 压缩(`.jsonl.zst`,通过新 `utils/jsonl_zst.py`);upsert 时先删除旧 chunk 的 keyword 文档;embedding 复用改为 `(text, embedding)` 键控,要求文本一致才复用;新增 `_matches_search_filter()` 对 vector/keyword 搜索做 path/path_prefix/metadata 的统一后过滤。 - **`keyword_index/bm25_index.py`**:索引文件名加入组件名 + tokenizer 指纹(sha256 前 12 位),快照/恢复时校验指纹防配置漂移;空索引 dump 时删除文件,加载失败抛错而非静默。 - **`file_chunker/markdown_file_chunker.py`**:弃用 `python-frontmatter`,改用内置 YAML 解析(非法 YAML 不阻断正文索引),并修正因 frontmatter 占用行号导致的 AST 行号偏移(`line_offset`)。 - **`cron_job.py`**:大幅简化(-187 行),由原来「dispatch 外部 job/step + 多种调度模式」改为「在自身 steps 上跑 cron 表达式」;`Application` 启动顺序随之调整为 base > stream > background > cron。 - 其余小调整:service(base/http/mcp)、file_graph、file_catalog、as_llm、as_embedding、tokenizer、prompt_handler、base_component 的签名/接口微调。 ### 8. Application 生命周期 - `_start()` 启动顺序明确为 components → base → stream → background → cron,启动失败会触发 `_close()` 回滚并 re-raise(不再吞异常)。 - 启动时创建 `session_dir` 目录;新增 `update_component()`(按类型/名就地更新已存在组件,不存在则报错)。 ### 9. File IO / 路径安全 - **`_path.py`**:`resolve_path` 增加 vault 越界防护(`is_relative_to` 校验),禁止 `.` / `..` 路径分量,支持 `allow_empty`。 - **`read.py`**:大文件(超过 `MAX_FILE_READ_BYTES`)走按行读取 `read_file_lines_safe`,避免一次性载入内存。 - **`_file_io.py` / `_daily_index.py` / `_path.py`** 等支持函数补齐(如 `refresh_day_index`、`read_file_lines_safe`)。 - **`env_utils.py`**:新增 `parse_env_file()`,`load_env()` 返回加载到的键值、支持 `override`、对无路径调用做幂等缓存。 ### 10. Config - `ApplicationConfig` 新增 `session_dir`(默认 `reme_session`)。 - `config_parser.py`:环境变量展开后做类型转换(`_convert_value`)、dot-notation 与 key=value 参数校验更严格、配置文件路径支持相对 `_CONFIG_DIR` 查找、根非 dict 报错。 - `default.yaml`:作业编排改用 `init_changes_step` + `dispatch_steps`(index/resource/digest 三个 watch loop 与 reindex);新增 `auto_dream`(4 步)、`proactive` 作业,移除旧 `dream`;file_catalog 增配 `resource` / `digest` / `dream` 实例;LLM 默认值与 Claude Code 凭据配置调整(tool_result_limit 50000、thinking_enable=false 等)。 ### 11. 其它 - 新增 `steps/common/add.py`(`AddStep` 算术 demo)、`channel/__init__.py` 与 common `__init__` 导出整理。 - 新增 4 篇文档:`docs4/auto_dream_logic_and_step_refactor.md`、`docs4/watch_loop_step_refactor_plan.md`、`docs4/todo.md`,以及 `reme_design.md` 更新。 **
718 lines
28 KiB
Python
718 lines
28 KiB
Python
"""Tests for background steps: scan/watch/dispatch steps.
|
|
|
|
Both scan steps are subclasses of BaseStep. To exercise them without spinning up
|
|
the full ApplicationContext, we pass real (started) file_store/file_chunker via
|
|
the step's kwargs (so the BaseStep _resolve() machinery returns them).
|
|
|
|
InitChangesStep writes its result into ``context["changes"]`` for a
|
|
downstream ``update_index_step`` to consume; tests assert against that key.
|
|
"""
|
|
|
|
# pylint: disable=protected-access
|
|
|
|
import asyncio
|
|
import os
|
|
import tempfile
|
|
import warnings
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock
|
|
|
|
from watchfiles import Change
|
|
|
|
from reme4.components.file_chunker import DefaultFileChunker
|
|
from reme4.components.file_catalog import LocalFileCatalog
|
|
from reme4.components.file_store import LocalFileStore
|
|
from reme4.components.runtime_context import RuntimeContext
|
|
from reme4.enumeration import ComponentEnum
|
|
from reme4.steps.evolve.auto_resource import AutoResourceStep, _compute_note_stem
|
|
from reme4.steps.index import (
|
|
DEFAULT_LOW_POWER_POLL_MS,
|
|
DEFAULT_WATCH_DEBOUNCE_MS,
|
|
DEFAULT_WATCH_STEP_MS,
|
|
ClearStoreStep,
|
|
InitChangesStep,
|
|
LogChangesStep,
|
|
UpdateCatalogStep,
|
|
WatchChangesStep,
|
|
)
|
|
from reme4.steps.index._change_batch import bucket_changes
|
|
from reme4.steps.index._watch_rules import WatchRule, build_watch_rules, collect_existing, match_file
|
|
|
|
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
|
|
|
|
|
|
def _make_app_context(vault_path: Path, daily_dir="daily", digest_dir="digest", resource_dir="resource"):
|
|
"""Create a mock app_context with app_config pointing to the given vault."""
|
|
ctx = MagicMock()
|
|
ctx.app_config.vault_dir = str(vault_path)
|
|
ctx.app_config.daily_dir = daily_dir
|
|
ctx.app_config.digest_dir = digest_dir
|
|
ctx.app_config.resource_dir = resource_dir
|
|
return ctx
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _watch_rules module tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_build_watch_rules_basic():
|
|
"""Build rules from watch_dirs and watch_suffixes."""
|
|
app_config = MagicMock()
|
|
app_config.daily_dir = "daily"
|
|
app_config.digest_dir = "digest"
|
|
app_config.resource_dir = "resource"
|
|
vault = Path("/fake/vault")
|
|
|
|
rules = build_watch_rules(app_config, vault, watch_dirs=["daily_dir", "digest_dir"], watch_suffixes=["md"])
|
|
assert len(rules) == 2
|
|
assert rules[0].path == vault / "daily"
|
|
assert rules[0].suffixes == ["md"]
|
|
assert rules[1].path == vault / "digest"
|
|
print("✓ test_build_watch_rules_basic passed")
|
|
|
|
|
|
def test_build_watch_rules_multiple_suffixes():
|
|
"""Multiple suffixes are forwarded to each rule."""
|
|
app_config = MagicMock()
|
|
app_config.daily_dir = "daily"
|
|
app_config.resource_dir = "resource"
|
|
vault = Path("/fake/vault")
|
|
|
|
rules = build_watch_rules(
|
|
app_config,
|
|
vault,
|
|
watch_dirs=["daily_dir", "resource_dir"],
|
|
watch_suffixes=["md", "jsonl"],
|
|
)
|
|
assert len(rules) == 2
|
|
assert rules[0].suffixes == ["md", "jsonl"]
|
|
assert rules[1].suffixes == ["md", "jsonl"]
|
|
print("✓ test_build_watch_rules_multiple_suffixes passed")
|
|
|
|
|
|
def test_build_watch_rules_fallback_literal():
|
|
"""Unknown field names are used as literal directory names."""
|
|
app_config = MagicMock(spec=[]) # no attributes
|
|
vault = Path("/fake/vault")
|
|
rules = build_watch_rules(app_config, vault, watch_dirs=["custom_dir"], watch_suffixes=["txt"])
|
|
assert rules[0].path == vault / "custom_dir"
|
|
print("✓ test_build_watch_rules_fallback_literal passed")
|
|
|
|
|
|
def test_match_file_suffix():
|
|
"""match_file accepts files matching suffix under rule path."""
|
|
rules = [WatchRule(path=Path("/vault/daily"), suffixes=["md"])]
|
|
assert match_file("/vault/daily/2026-01-01.md", rules)
|
|
assert match_file("/vault/daily/sub/note.md", rules)
|
|
assert not match_file("/vault/daily/file.txt", rules)
|
|
assert not match_file("/vault/other/file.md", rules)
|
|
print("✓ test_match_file_suffix passed")
|
|
|
|
|
|
def test_match_file_no_suffix_filter():
|
|
"""Empty suffixes list means all files match."""
|
|
rules = [WatchRule(path=Path("/vault/resource"), suffixes=[])]
|
|
assert match_file("/vault/resource/anything.xyz", rules)
|
|
assert match_file("/vault/resource/sub/deep.pdf", rules)
|
|
assert not match_file("/vault/other/file.md", rules)
|
|
print("✓ test_match_file_no_suffix_filter passed")
|
|
|
|
|
|
def test_collect_existing_filters():
|
|
"""collect_existing applies suffix rules correctly."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
vault = Path(tmpdir)
|
|
daily = vault / "daily"
|
|
resource = vault / "resource"
|
|
write_file(daily / "note.md")
|
|
write_file(daily / "ignore.txt")
|
|
write_file(resource / "data.json")
|
|
write_file(resource / "binary.png")
|
|
|
|
rules = [
|
|
WatchRule(path=daily, suffixes=["md"]),
|
|
WatchRule(path=resource, suffixes=["json"]),
|
|
]
|
|
result = collect_existing(rules, recursive=True)
|
|
paths = set(result.keys())
|
|
assert str((daily / "note.md").absolute()) in paths
|
|
assert str((daily / "ignore.txt").absolute()) not in paths
|
|
assert str((resource / "data.json").absolute()) in paths
|
|
assert str((resource / "binary.png").absolute()) not in paths
|
|
print("✓ test_collect_existing_filters passed")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# InitChangesStep
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_clear_and_scan_defaults_include_jsonl():
|
|
"""Full reindex should include jsonl files when no explicit suffix filter is passed."""
|
|
|
|
async def run():
|
|
with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
|
|
cwd = Path.cwd()
|
|
write_file(cwd / "daily" / "note.md", "alpha")
|
|
write_file(cwd / "resource" / "events.jsonl", '{"a": 1}\n')
|
|
write_file(cwd / "resource" / "ignore.txt", "skip")
|
|
|
|
fs = LocalFileStore(name="test_store", embedding_store="")
|
|
await fs.start()
|
|
try:
|
|
clear_step = ClearStoreStep(file_store=fs, app_context=_make_app_context(cwd))
|
|
scan_step = InitChangesStep(store="file_store", file_store=fs, app_context=_make_app_context(cwd))
|
|
ctx = RuntimeContext(watch_dirs=["daily_dir", "resource_dir"], watch_suffixes=["md", "jsonl"])
|
|
await clear_step(ctx)
|
|
resp = await scan_step(ctx)
|
|
paths = {Path(item["path"]).name for item in ctx["changes"]}
|
|
assert resp.metadata["counts"] == {"added": 2, "modified": 0, "deleted": 0}
|
|
assert paths == {"note.md", "events.jsonl"}
|
|
finally:
|
|
await fs.close()
|
|
print("✓ test_clear_and_scan_defaults_include_jsonl passed")
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
async def _make_scan_step(vault_path: Path, watch_dirs=None, watch_suffixes=None, recursive=True):
|
|
fs = LocalFileStore(name="test_store", embedding_store="")
|
|
chunker = DefaultFileChunker()
|
|
await fs.start()
|
|
await chunker.start()
|
|
app_ctx = _make_app_context(vault_path)
|
|
step = InitChangesStep(
|
|
store="file_store",
|
|
recursive=recursive,
|
|
file_store=fs,
|
|
file_chunker=chunker,
|
|
app_context=app_ctx,
|
|
)
|
|
context = RuntimeContext(
|
|
watch_dirs=watch_dirs or ["daily_dir", "digest_dir"],
|
|
watch_suffixes=watch_suffixes or ["md"],
|
|
)
|
|
return step, context, fs, chunker
|
|
|
|
|
|
async def _teardown(fs: LocalFileStore, chunker: DefaultFileChunker) -> None:
|
|
await chunker.close()
|
|
await fs.close()
|
|
|
|
|
|
def test_scan_changes_initial_all_added():
|
|
"""First run on a fresh store emits 'added' for every existing file."""
|
|
|
|
async def run():
|
|
with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
|
|
cwd = Path.cwd()
|
|
write_file(cwd / "daily" / "a.md", "alpha")
|
|
write_file(cwd / "daily" / "b.md", "beta")
|
|
(cwd / "digest").mkdir(parents=True, exist_ok=True)
|
|
|
|
step, ctx, fs, chunker = await _make_scan_step(cwd)
|
|
try:
|
|
resp = await step(ctx)
|
|
counts = resp.metadata["counts"]
|
|
assert counts == {"added": 2, "modified": 0, "deleted": 0}
|
|
assert len(ctx["changes"]) == 2
|
|
finally:
|
|
await _teardown(fs, chunker)
|
|
print("✓ test_scan_changes_initial_all_added passed")
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_scan_changes_no_changes():
|
|
"""Second run over an unchanged store reports zero counts."""
|
|
|
|
async def run():
|
|
with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
|
|
cwd = Path.cwd()
|
|
a = write_file(cwd / "daily" / "a.md", "alpha")
|
|
(cwd / "digest").mkdir(parents=True, exist_ok=True)
|
|
|
|
step, ctx, fs, chunker = await _make_scan_step(cwd)
|
|
try:
|
|
node, chunks = await chunker.chunk(a)
|
|
await fs.upsert([(node, chunks)])
|
|
resp = await step(ctx)
|
|
assert resp.metadata["counts"] == {"added": 0, "modified": 0, "deleted": 0}
|
|
assert ctx["changes"] == []
|
|
finally:
|
|
await _teardown(fs, chunker)
|
|
print("✓ test_scan_changes_no_changes passed")
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_scan_changes_detect_modify_delete():
|
|
"""Second pass distinguishes added/modified/deleted."""
|
|
|
|
async def run():
|
|
with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
|
|
cwd = Path.cwd()
|
|
a = write_file(cwd / "daily" / "a.md", "alpha")
|
|
b = write_file(cwd / "daily" / "b.md", "beta")
|
|
(cwd / "digest").mkdir(parents=True, exist_ok=True)
|
|
|
|
step, ctx, fs, chunker = await _make_scan_step(cwd)
|
|
try:
|
|
for p in (a, b):
|
|
node, chunks = await chunker.chunk(p)
|
|
await fs.upsert([(node, chunks)])
|
|
a.write_text("alpha-v2", encoding="utf-8")
|
|
os.utime(a, (9_999_999_999, 9_999_999_999))
|
|
b.unlink()
|
|
write_file(cwd / "daily" / "c.md", "gamma")
|
|
|
|
resp = await step(ctx)
|
|
counts = resp.metadata["counts"]
|
|
assert counts == {"added": 1, "modified": 1, "deleted": 1}
|
|
finally:
|
|
await _teardown(fs, chunker)
|
|
print("✓ test_scan_changes_detect_modify_delete passed")
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_scan_changes_missing_dir_skipped():
|
|
"""Non-existent watch_dirs entries are dropped silently."""
|
|
|
|
async def run():
|
|
with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
|
|
cwd = Path.cwd()
|
|
(cwd / "daily").mkdir()
|
|
# digest dir missing
|
|
step, ctx, fs, chunker = await _make_scan_step(cwd)
|
|
try:
|
|
resp = await step(ctx)
|
|
assert resp.metadata["counts"] == {"added": 0, "modified": 0, "deleted": 0}
|
|
finally:
|
|
await _teardown(fs, chunker)
|
|
print("✓ test_scan_changes_missing_dir_skipped passed")
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_scan_changes_resource_dir():
|
|
"""Scanning resource_dir with multiple suffixes works."""
|
|
|
|
async def run():
|
|
with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
|
|
cwd = Path.cwd()
|
|
resource = cwd / "resource"
|
|
write_file(resource / "data.json", "{}")
|
|
write_file(resource / "note.md", "# Note")
|
|
write_file(resource / "image.png", "binary")
|
|
|
|
step, ctx, fs, chunker = await _make_scan_step(
|
|
cwd,
|
|
watch_dirs=["resource_dir"],
|
|
watch_suffixes=["md", "json"],
|
|
)
|
|
try:
|
|
resp = await step(ctx)
|
|
assert resp.metadata["counts"]["added"] == 2
|
|
finally:
|
|
await _teardown(fs, chunker)
|
|
print("✓ test_scan_changes_resource_dir passed")
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_init_changes_named_file_catalog_monitor():
|
|
"""monitor_type/monitor_name selects the requested file_catalog component."""
|
|
|
|
async def run():
|
|
with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
|
|
cwd = Path.cwd()
|
|
write_file(cwd / "resource" / "2026-01-01" / "a.md", "alpha")
|
|
|
|
catalog = LocalFileCatalog(name="resource")
|
|
await catalog.start()
|
|
try:
|
|
app_ctx = _make_app_context(cwd)
|
|
app_ctx.components = {ComponentEnum.FILE_CATALOG: {"resource": catalog}}
|
|
step = InitChangesStep(monitor_type="file_catalog", monitor_name="resource", app_context=app_ctx)
|
|
ctx = RuntimeContext(watch_dirs=["resource_dir"], watch_suffixes=["md"])
|
|
resp = await step(ctx)
|
|
|
|
assert resp.metadata["counts"] == {"added": 1, "modified": 0, "deleted": 0}
|
|
assert ctx["changes"][0]["change"] == "added"
|
|
finally:
|
|
await catalog.close()
|
|
print("✓ test_init_changes_named_file_catalog_monitor passed")
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_bucket_changes_coalesces_by_final_file_state():
|
|
"""A delete+add replacement batch for an existing file becomes one modified event."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
p = write_file(Path(tmpdir) / "daily" / "a.md", "alpha")
|
|
buckets = bucket_changes(
|
|
[
|
|
{"change": "deleted", "path": str(p)},
|
|
{"change": "added", "path": str(p)},
|
|
],
|
|
)
|
|
|
|
assert buckets[Change.modified] == [str(p)]
|
|
assert buckets[Change.added] == []
|
|
assert buckets[Change.deleted] == []
|
|
print("✓ test_bucket_changes_coalesces_by_final_file_state passed")
|
|
|
|
|
|
def test_update_catalog_relative_path_uses_vault():
|
|
"""update_catalog_step resolves vault-relative change paths against vault_path."""
|
|
|
|
async def run():
|
|
with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
|
|
cwd = Path.cwd()
|
|
write_file(cwd / "daily" / "a.md", "alpha")
|
|
|
|
catalog = LocalFileCatalog(name="test_catalog")
|
|
await catalog.start()
|
|
try:
|
|
step = UpdateCatalogStep(file_catalog=catalog, app_context=_make_app_context(cwd))
|
|
ctx = RuntimeContext(changes=[{"change": "added", "path": "daily/a.md"}])
|
|
resp = await step(ctx)
|
|
|
|
assert resp.success is True
|
|
nodes = await catalog.get_nodes()
|
|
assert [n.path for n in nodes] == ["daily/a.md"]
|
|
finally:
|
|
await catalog.close()
|
|
print("✓ test_update_catalog_relative_path_uses_vault passed")
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_index_update_loop_init_dispatch_updates_store_across_batches():
|
|
"""index_update_loop init scan dispatches to update_index_step and preserves final store state."""
|
|
|
|
async def run():
|
|
with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
|
|
cwd = Path.cwd()
|
|
daily_a = write_file(cwd / "daily" / "a.md", "alpha\n[[digest/report.md]]\n")
|
|
write_file(cwd / "digest" / "report.md", "# Report\nbeta\n")
|
|
write_file(cwd / "daily" / "ignore.txt", "skip")
|
|
|
|
fs = LocalFileStore(name="default", embedding_store="")
|
|
chunker = DefaultFileChunker()
|
|
await fs.start()
|
|
await chunker.start()
|
|
try:
|
|
app_ctx = _make_app_context(cwd)
|
|
app_ctx.components = {
|
|
ComponentEnum.FILE_STORE: {"default": fs},
|
|
ComponentEnum.FILE_CHUNKER: {"default": chunker},
|
|
}
|
|
ctx = RuntimeContext(watch_dirs=["daily_dir", "digest_dir"], watch_suffixes=["md"])
|
|
init_step = InitChangesStep(
|
|
monitor_type="file_store",
|
|
monitor_name="default",
|
|
dispatch_steps=["update_index_step"],
|
|
app_context=app_ctx,
|
|
)
|
|
|
|
first = await init_step(ctx)
|
|
assert first.metadata["counts"] == {"added": 2, "modified": 0, "deleted": 0}
|
|
nodes = {n.path: n for n in await fs.get_nodes()}
|
|
assert set(nodes) == {"daily/a.md", "digest/report.md"}
|
|
assert all(nodes[p].chunk_ids for p in nodes)
|
|
|
|
daily_a.write_text("alpha v2\n[[digest/report.md]]\n", encoding="utf-8")
|
|
os.utime(daily_a, (9_999_999_999, 9_999_999_999))
|
|
(cwd / "digest" / "report.md").unlink()
|
|
write_file(cwd / "daily" / "c.md", "gamma\n")
|
|
|
|
second = await init_step(ctx)
|
|
assert second.metadata["counts"] == {"added": 1, "modified": 1, "deleted": 1}
|
|
assert {(c["change"], Path(c["path"]).name) for c in ctx["changes"]} == {
|
|
("modified", "a.md"),
|
|
("deleted", "report.md"),
|
|
("added", "c.md"),
|
|
}
|
|
|
|
nodes = {n.path: n for n in await fs.get_nodes()}
|
|
assert set(nodes) == {"daily/a.md", "daily/c.md"}
|
|
assert all(nodes[p].chunk_ids for p in nodes)
|
|
assert all(chunk.path in nodes for chunk in fs.file_chunks.values())
|
|
finally:
|
|
await chunker.close()
|
|
await fs.close()
|
|
print("✓ test_index_update_loop_init_dispatch_updates_store_across_batches passed")
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_digest_watch_loop_init_dispatch_updates_named_catalog_and_logs():
|
|
"""digest_watch_loop style config updates the digest catalog without touching resource/default catalogs."""
|
|
|
|
async def run():
|
|
with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
|
|
cwd = Path.cwd()
|
|
daily = write_file(cwd / "daily" / "2026-01-01.md", "day one")
|
|
digest = write_file(cwd / "digest" / "week.md", "weekly")
|
|
write_file(cwd / "resource" / "asset.md", "not watched by digest loop")
|
|
|
|
digest_catalog = LocalFileCatalog(name="digest")
|
|
resource_catalog = LocalFileCatalog(name="resource")
|
|
await digest_catalog.start()
|
|
await resource_catalog.start()
|
|
try:
|
|
app_ctx = _make_app_context(cwd)
|
|
app_ctx.components = {
|
|
ComponentEnum.FILE_CATALOG: {
|
|
"digest": digest_catalog,
|
|
"resource": resource_catalog,
|
|
},
|
|
}
|
|
ctx = RuntimeContext(watch_dirs=["daily_dir", "digest_dir"], watch_suffixes=["md"])
|
|
init_step = InitChangesStep(
|
|
monitor_type="file_catalog",
|
|
monitor_name="digest",
|
|
dispatch_steps=[
|
|
{"backend": "update_catalog_step", "file_catalog": "digest"},
|
|
{"backend": "log_changes_step"},
|
|
],
|
|
app_context=app_ctx,
|
|
)
|
|
|
|
first = await init_step(ctx)
|
|
assert first.metadata["counts"] == {"added": 2, "modified": 0, "deleted": 0}
|
|
assert {n.path for n in await digest_catalog.get_nodes()} == {
|
|
"daily/2026-01-01.md",
|
|
"digest/week.md",
|
|
}
|
|
assert await resource_catalog.get_nodes() == []
|
|
|
|
daily.write_text("day two", encoding="utf-8")
|
|
os.utime(daily, (9_999_999_999, 9_999_999_999))
|
|
digest.unlink()
|
|
write_file(cwd / "daily" / "2026-01-02.md", "next day")
|
|
|
|
second = await init_step(ctx)
|
|
assert second.metadata["counts"] == {"added": 1, "modified": 1, "deleted": 1}
|
|
assert {(c["change"], Path(c["path"]).name) for c in ctx["changes"]} == {
|
|
("modified", "2026-01-01.md"),
|
|
("deleted", "week.md"),
|
|
("added", "2026-01-02.md"),
|
|
}
|
|
assert {n.path for n in await digest_catalog.get_nodes()} == {
|
|
"daily/2026-01-01.md",
|
|
"daily/2026-01-02.md",
|
|
}
|
|
assert await resource_catalog.get_nodes() == []
|
|
finally:
|
|
await resource_catalog.close()
|
|
await digest_catalog.close()
|
|
print("✓ test_digest_watch_loop_init_dispatch_updates_named_catalog_and_logs passed")
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# WatchChangesStep
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_watch_changes_default_low_power_timing():
|
|
"""Default watcher timing favors lower resource use."""
|
|
step = WatchChangesStep()
|
|
|
|
assert step.debounce == DEFAULT_WATCH_DEBOUNCE_MS
|
|
assert step.step == DEFAULT_WATCH_STEP_MS
|
|
assert step.poll_delay_ms == DEFAULT_LOW_POWER_POLL_MS
|
|
|
|
custom = WatchChangesStep(debounce=1000, step=250, poll_delay_ms=3000)
|
|
assert custom.debounce == 1000
|
|
assert custom.step == 250
|
|
assert custom.poll_delay_ms == 3000
|
|
|
|
print("✓ test_watch_changes_default_low_power_timing passed")
|
|
|
|
|
|
def test_watch_changes_requires_stop_event():
|
|
"""Missing stop_event in context raises a clear error."""
|
|
|
|
async def run():
|
|
with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
|
|
cwd = Path.cwd()
|
|
(cwd / "daily").mkdir()
|
|
app_ctx = _make_app_context(cwd)
|
|
step = WatchChangesStep(app_context=app_ctx)
|
|
step.context = RuntimeContext(watch_dirs=["daily_dir"], watch_suffixes=["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_no_valid_paths():
|
|
"""With no valid watch_paths, the step raises."""
|
|
|
|
async def run():
|
|
with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
|
|
cwd = Path.cwd()
|
|
app_ctx = _make_app_context(cwd)
|
|
step = WatchChangesStep(app_context=app_ctx)
|
|
stop = asyncio.Event()
|
|
step.context = RuntimeContext(stop_event=stop, watch_dirs=["daily_dir"], watch_suffixes=["md"])
|
|
try:
|
|
await step.execute()
|
|
except RuntimeError as e:
|
|
assert "No valid watch paths" in str(e)
|
|
else:
|
|
raise AssertionError("expected RuntimeError")
|
|
print("✓ test_watch_changes_raises_no_valid_paths passed")
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_watch_changes_filter_matches_rules():
|
|
"""The internal filter uses watch rules from context."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
vault = Path(tmpdir)
|
|
(vault / "daily").mkdir()
|
|
(vault / "digest").mkdir()
|
|
(vault / "resource").mkdir()
|
|
app_ctx = _make_app_context(vault)
|
|
|
|
step = WatchChangesStep(app_context=app_ctx)
|
|
step.context = RuntimeContext(watch_dirs=["daily_dir", "digest_dir"], watch_suffixes=["md"])
|
|
step._rules = step._get_watch_rules()
|
|
|
|
assert step._filter(Change.added, str(vault / "daily/foo.md"))
|
|
assert step._filter(Change.added, str(vault / "digest/bar.md"))
|
|
assert not step._filter(Change.added, str(vault / "daily/foo.txt"))
|
|
assert not step._filter(Change.added, str(vault / "resource/file.md"))
|
|
|
|
print("✓ test_watch_changes_filter_matches_rules passed")
|
|
|
|
|
|
def test_watch_changes_dispatch_steps_list():
|
|
"""dispatch_steps config is stored by BaseStep."""
|
|
step = WatchChangesStep(dispatch_steps=["update_catalog_step", "auto_resource_step"])
|
|
assert step.dispatch_step_specs == ["update_catalog_step", "auto_resource_step"]
|
|
|
|
print("✓ test_watch_changes_dispatch_steps_list passed")
|
|
|
|
|
|
def test_auto_resource_batch_deleted_changes():
|
|
"""AutoResourceStep accepts a batch of change dicts from dispatch_steps."""
|
|
|
|
async def run():
|
|
with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
|
|
cwd = Path.cwd()
|
|
app_ctx = _make_app_context(cwd)
|
|
fs = LocalFileStore(name="test_store", embedding_store="")
|
|
await fs.start()
|
|
try:
|
|
filename = "file.md"
|
|
note_stem = _compute_note_stem(filename)
|
|
note_path = cwd / "daily" / "2026-01-01" / f"{note_stem}.md"
|
|
write_file(note_path, "---\nname: test\n---\nbody\n")
|
|
|
|
step = AutoResourceStep(app_context=app_ctx, file_store=fs)
|
|
ctx = RuntimeContext(
|
|
changes=[
|
|
{"change": "deleted", "path": str(cwd / "resource" / "2026-01-01" / filename)},
|
|
],
|
|
)
|
|
resp = await step(ctx)
|
|
|
|
assert resp.success is True
|
|
assert resp.answer == "Processed 1/1 resource change(s)"
|
|
assert resp.metadata["processed"] == 1
|
|
assert resp.metadata["results"][0]["path"] == "resource/2026-01-01/file.md"
|
|
assert not note_path.exists()
|
|
finally:
|
|
await fs.close()
|
|
print("✓ test_auto_resource_batch_deleted_changes passed")
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
# LogChangesStep
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_log_changes_step():
|
|
"""LogChangesStep logs and reports count."""
|
|
|
|
async def run():
|
|
step = LogChangesStep()
|
|
changes = [
|
|
{"change": "added", "path": "/vault/daily/note.md"},
|
|
{"change": "deleted", "path": "/vault/daily/old.md"},
|
|
]
|
|
ctx = RuntimeContext(changes=changes)
|
|
resp = await step(ctx)
|
|
assert resp.success is True
|
|
assert resp.metadata["count"] == 2
|
|
print("✓ test_log_changes_step passed")
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print("\n=== Background Steps Tests ===")
|
|
# _watch_rules
|
|
test_build_watch_rules_basic()
|
|
test_build_watch_rules_multiple_suffixes()
|
|
test_build_watch_rules_fallback_literal()
|
|
test_match_file_suffix()
|
|
test_match_file_no_suffix_filter()
|
|
test_collect_existing_filters()
|
|
# InitChangesStep
|
|
test_clear_and_scan_defaults_include_jsonl()
|
|
test_scan_changes_initial_all_added()
|
|
test_scan_changes_no_changes()
|
|
test_scan_changes_detect_modify_delete()
|
|
test_scan_changes_missing_dir_skipped()
|
|
test_scan_changes_resource_dir()
|
|
test_index_update_loop_init_dispatch_updates_store_across_batches()
|
|
test_digest_watch_loop_init_dispatch_updates_named_catalog_and_logs()
|
|
# WatchChangesStep
|
|
test_watch_changes_default_low_power_timing()
|
|
test_watch_changes_requires_stop_event()
|
|
test_watch_changes_raises_no_valid_paths()
|
|
test_watch_changes_filter_matches_rules()
|
|
test_watch_changes_dispatch_steps_list()
|
|
test_auto_resource_batch_deleted_changes()
|
|
# LogChangesStep
|
|
test_log_changes_step()
|
|
print("\n所有测试通过!")
|