ReMe/tests4/unit/test_background_steps.py
jinliyl 83831ec90c
feat(core): enhance reme4 (#281)
### 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` 更新。
**
2026-06-19 01:35:31 +08:00

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所有测试通过!")