ReMe/docs4/watch_loop_step_refactor_plan.md

Watch Loop Step 重构计划

背景

当前 index_update_loop、resource_watch_loop、digest_watch_loop 都是同一种范式：

1. 启动时扫描已有文件变化。
2. 用一组 step 处理扫描出来的变化。
3. 进入持续监听。
4. 持续监听到变化后，再用同一组 step 处理变化。

也就是说，初始化扫描和持续监听只是变化来源不同，后续更新逻辑应该共享。

现在的问题是这个范式没有被显式建模：

• index_update_loop 初始化和监听都走 update_index_step，基本一致。
• resource_watch_loop 初始化和监听都走 update_catalog_step + foreach_dispatch_step，基本一致。
• digest_watch_loop 初始化走 update_catalog_step，但监听阶段只走 log_changes_step，导致 live changes 不更新 file_catalog。
• watch_changes_step 同时负责监听和 dispatch 下游逻辑，职责偏重。
• update_index_step 和 update_catalog_step 内部有较多重复的变化分桶、结果收集、删除、持久化逻辑。

目标

重构后希望形成统一模型：

change producer:
  init_changes_step     # 初始化，一次性产生 changes 并 dispatch
  watch_changes_step    # 持续监听，持续产生 changes

change handlers:
  update_index_step
  update_catalog_step
  foreach_dispatch_step
  log_changes_step
  channel_notify_step   # 后续可选

核心原则：

• producer 只负责产生 context["changes"]。
• handler 只负责消费 context["changes"]。
• 初始化和持续监听都通过 BaseStep.dispatch_steps(...) 调用 handler。
• init_changes_step 和 watch_changes_step 显式配置同一组 dispatch_steps，让范式直接可见。

配置设计

不新增 job 级 change_steps。每个 producer step 自己声明 dispatch_steps，初始化 producer 和持续监听 producer 配同一组 handler。

配置精简约定：

• init_changes_step.recursive 和 watch_changes_step.recursive 默认就是 true，配置中不再显式写。
• update_index_step / update_catalog_step 的 persist 语义统一为默认 true，配置中不再显式写。
• 只有当某个 loop 需要关闭递归或关闭持久化时，才显式写 recursive: false / persist: false。

index_update_loop

index_update_loop:
  backend: background
  watch_dirs: [daily_dir, digest_dir, resource_dir]
  watch_suffixes: [md, jsonl]
  steps:
    - backend: init_changes_step
      store: file_store
      dispatch_steps: [update_index_step]
    - backend: watch_changes_step
      dispatch_steps: [update_index_step]

resource_watch_loop

resource_watch_loop:
  backend: background
  watch_dirs: [resource_dir]
  watch_suffixes: [md, txt, json, jsonl, csv, yaml, html]
  dispatch_job: auto_resource
  steps:
    - backend: init_changes_step
      store: file_catalog
      dispatch_steps: [update_catalog_step, foreach_dispatch_step]
    - backend: watch_changes_step
      dispatch_steps: [update_catalog_step, foreach_dispatch_step]

digest_watch_loop

digest_watch_loop:
  backend: background
  watch_dirs: [daily_dir, digest_dir]
  watch_suffixes: [md]
  steps:
    - backend: init_changes_step
      store: file_catalog
      dispatch_steps: [update_catalog_step, log_changes_step]
    - backend: watch_changes_step
      dispatch_steps: [update_catalog_step, log_changes_step]

这样三个 loop 都统一成：

startup:
  scan changes
  dispatch handlers

runtime:
  watch changes
  dispatch same handlers

Step 拆分

1. BaseStep dispatch 能力

把 dispatch 能力沉到 BaseStep，所有 producer step 共享：

• normalize_dispatch_steps(dispatch_step, dispatch_steps)
• dispatch_steps(dispatch_steps, **kwargs)

这样 init_changes_step 和 watch_changes_step 都不需要各自实现 registry 查询、step 实例化和 context 透传。

dispatch_steps 支持两种形式：

dispatch_steps: [update_catalog_step, log_changes_step]

也支持给单个 handler 传参数：

dispatch_steps:
  - backend: update_catalog_step
  - backend: some_step
    option: value

2. init_changes_step

新增 init_changes_step，替代现有两个初始化扫描 step：

• scan_store_changes_step
• scan_catalog_changes_step

参数：

store: file_catalog | file_store
dispatch_steps: [...]

职责：

• 根据 watch_dirs / watch_suffixes 收集磁盘文件。
• 根据 store 和目标状态源比较。
• 生成统一格式的 context["changes"]。
• 如果有变化，调用 BaseStep.dispatch_steps(...) 执行 handler。

输出格式：

[
    {"change": "added", "path": "/abs/path/to/file.md"},
    {"change": "modified", "path": "/abs/path/to/file.md"},
    {"change": "deleted", "path": "/abs/path/to/file.md"},
]

3. watch_changes_step

保留监听职责，弱化业务 dispatch 职责。

职责：

• 根据 watch_dirs / watch_suffixes 建立文件监听。
• 对每个 debounced batch 生成同样格式的 changes。
• 调用 BaseStep.dispatch_steps(...) 执行 handler。

4. update_index_step / update_catalog_step

第二阶段再精简。

它们现在重复逻辑包括：

• 解析 added / modified / deleted。
• 判断文件是否存在。
• 收集 per-path result。
• 删除旧记录。
• upsert 新记录。
• persist。
• 写 response。

建议抽内部基类，例如：

class ChangeApplyStep(BaseStep):
    async def parse_added_or_modified(self, path): ...
    async def upsert_items(self, items): ...
    async def delete_paths(self, rel_paths): ...
    async def dump_target(self): ...

然后：

• UpdateCatalogStep 只实现 stat -> FileNode，写 file_catalog。
• UpdateIndexStep 只实现 chunk_file -> FileNode + chunks，写 file_store。

这一步可以在 init_changes_step 落地后做，降低一次性改动风险。

实施顺序

当前落地状态：

• Phase 1 已完成：BaseStep.dispatch_steps(...)、init_changes_step、默认持久化、默认配置精简已落地。
• Phase 2 已完成：digest_watch_loop 的初始化和监听都执行 update_catalog_step + log_changes_step。
• 兼容旧 backend 不保留：scan_store_changes_step / scan_catalog_changes_step 已删除。
• reindex 已改为 clear_store_step + init_changes_step(store=file_store, dispatch_steps=[update_index_step])。
• Phase 3 已完成一层：update_catalog_step 和 update_index_step 已合并到 update_changes.py， 并抽出 ChangeApplyStep 复用 added/modified/deleted、upsert/delete/persist 模板逻辑。

Phase 1：统一范式

1. 把 dispatch 能力沉到 BaseStep。
2. 新增 init_changes_step。
3. 将 update_index_step 和 update_catalog_step 的 persist 默认值统一为 true。
4. 修改 default.yaml 里的三个 loop：
   • 初始化阶段统一使用 init_changes_step。
   • init_changes_step 和 watch_changes_step 配置相同的 dispatch_steps。
5. watch_changes_step 保留 dispatch_step 到 dispatch_steps 的轻量兼容。
6. 验证三个 loop 的启动扫描和 live watch 都会执行同一条 handler pipeline。

Phase 2：修正 digest_watch_loop 语义

digest_watch_loop 的 live changes 应该更新 file_catalog，因此初始化和监听阶段都应配置同一组 dispatch_steps：

dispatch_steps: [update_catalog_step, log_changes_step]

如果后续要通知 channel，可以追加：

  - backend: channel_notify_step

Phase 3：精简 update 类 step

1. 抽 ChangeApplyStep 基类或 helper 函数。
2. 让 update_index_step 和 update_catalog_step 只保留各自差异逻辑。
3. 保持外部行为不变：
   • 输入仍然是 context["changes"]。
   • 输出仍然写 response.answer 和 response.success。
   • persist 语义不变。

验证点

最少需要覆盖这些场景：

• index_update_loop 启动扫描新增文件，会更新 file_store。
• index_update_loop live 新增/修改/删除文件，会更新 file_store。
• resource_watch_loop 启动扫描新增文件，会更新 file_catalog 并触发 auto_resource。
• resource_watch_loop live 新增文件，会更新 file_catalog 并触发 auto_resource。
• digest_watch_loop 启动扫描新增/修改/删除文件，会更新 file_catalog。
• digest_watch_loop live 新增/修改/删除文件，也会更新 file_catalog。
• changes 为空时，init_changes_step 不应执行 handler，也不应报错。

预期收益

• 三个 background loop 的结构统一。
• 初始化扫描和持续监听的处理逻辑完全复用。
• digest_watch_loop 不再出现启动和 live 语义不一致。
• watch_changes_step 和 init_changes_step 共享 BaseStep dispatch 能力。
• 后续新增日志、channel notification、auto dream 等 handler 时，初始化和监听两处使用同一组 dispatch_steps。