From 72e243b7d6529ffa86ff30ac357f4fb024c12cc7 Mon Sep 17 00:00:00 2001 From: hwan <3373484735@qq.com> Date: Fri, 16 Oct 2026 12:13:48 +0800 Subject: [PATCH] fix(auto-dream): checkpoint scanned source versions (#618) * fix(auto-dream): checkpoint scanned source versions * test(auto-dream): exercise failed day-index checkpoint guard * test(auto-dream): exercise valid correction receipts --------- Co-authored-by: jinli.yl --- docs/en/auto_dream.md | 6 +- docs/zh/auto_dream.md | 5 +- reme/steps/evolve/dream/finish.py | 8 +- tests/unit/test_auto_dream.py | 7 +- tests/unit/test_dream_checkpoint_versions.py | 205 +++++++++++++++++++ 5 files changed, 225 insertions(+), 6 deletions(-) create mode 100644 tests/unit/test_dream_checkpoint_versions.py diff --git a/docs/en/auto_dream.md b/docs/en/auto_dream.md index e75fabb5..6cc6ce92 100644 --- a/docs/en/auto_dream.md +++ b/docs/en/auto_dream.md @@ -125,7 +125,7 @@ paths enter `failed_paths`. The Finish stage does not checkpoint failed paths, e `dream_finish_step` completes the run: 1. Write successfully processed changed paths to `file_catalog: dream`. -2. Also write every refreshed day-index page in the scan window to the catalog. +2. Also write refreshed day-index pages observed by Extract in the scan window to the catalog. 3. Persist the dream catalog if there were upserts or deletions. 4. Return a summary containing counts for scanned, changed, integrated, checkpoints, and related values. @@ -135,6 +135,10 @@ pipeline; see [Proactive](./proactive.md). Failed paths are not checkpointed. The next `auto_dream` run therefore continues to treat them as changed inputs until integration succeeds. +Checkpoints retain the input mtimes observed by Extract, including scanned day-index pages. If a source is edited while +model work runs, its newer version remains changed for a later run, even after restart. The later run must include the +source date in its scan window. Finish does not checkpoint files absent from that scan. + ### 4. Auto Tag After Finish, both `auto_dream` and `dream_cron` run `auto_tag_step` on Markdown digest files actually created or modified diff --git a/docs/zh/auto_dream.md b/docs/zh/auto_dream.md index 990540d7..ad543711 100644 --- a/docs/zh/auto_dream.md +++ b/docs/zh/auto_dream.md @@ -117,7 +117,7 @@ Finish 阶段不会 checkpoint 失败路径，保证下次还能重试。 `dream_finish_step` 负责收尾： 1. 将成功处理的 changed paths 写入 `file_catalog: dream`。 -2. 将扫描窗口内每个已刷新的 day-index 页也写入 catalog。 +2. 将扫描窗口内由 Extract 扫描到的已刷新 day-index 页也写入 catalog。 3. 如果有 upsert 或 delete，持久化 dream catalog。 4. 返回包含 scanned、changed、integrated、checkpoint 等计数的摘要。 @@ -126,6 +126,9 @@ Auto Dream 不读取或写入 proactive 状态和 `interests.yaml`。这些文 失败路径不会被 checkpoint。这样下一次 `auto_dream` 仍会把它们视作 changed input，直到整合成功。 +Checkpoint 保留 Extract 扫描到的输入 mtime，day-index 页也遵循同一规则。如果来源在模型处理期间被编辑，新版本在后续调用中 +仍会被视为变化，重启后也能继续处理。后续调用的扫描窗口须包含来源日期。Finish 不会 checkpoint 本轮未扫描到的文件。 + ### 4. Auto Tag Finish 后，`auto_dream` 和 `dream_cron` 都会通过 `auto_tag_step` 为本轮整合实际新增或修改的 Markdown digest 文件打标， diff --git a/reme/steps/evolve/dream/finish.py b/reme/steps/evolve/dream/finish.py index d1245d82..243a449a 100644 --- a/reme/steps/evolve/dream/finish.py +++ b/reme/steps/evolve/dream/finish.py @@ -33,7 +33,7 @@ class DreamFinishStep(BaseStep): f"checkpoint={len(checkpoint)} day_indexes={len(day_index_paths)} " f"deleted={len(state.deleted_paths)} persist={self.persist}", ) - upserts = self._nodes(workspace, upsert_paths) + upserts = self._nodes(workspace, upsert_paths, state.existing) if upserts: self.logger.info(f"[{self.name}] catalog upsert start nodes={len(upserts)}") await self.file_catalog.upsert(upserts) @@ -59,11 +59,13 @@ class DreamFinishStep(BaseStep): return self.context.response @staticmethod - def _nodes(workspace: Path, paths: list[str]) -> list[FileNode]: + def _nodes(workspace: Path, paths: list[str], observed: dict[str, float]) -> list[FileNode]: + """Checkpoint scanned versions, leaving later edits eligible for another run.""" out: list[FileNode] = [] for rel in paths: try: - out.append(FileNode(path=rel, st_mtime=(workspace / rel).stat().st_mtime)) + if rel in observed and (workspace / rel).is_file(): + out.append(FileNode(path=rel, st_mtime=observed[rel])) except OSError: continue return out diff --git a/tests/unit/test_auto_dream.py b/tests/unit/test_auto_dream.py index 1bafb889..8662c1e8 100644 --- a/tests/unit/test_auto_dream.py +++ b/tests/unit/test_auto_dream.py @@ -712,6 +712,9 @@ def test_finish_does_not_checkpoint_failed_changed_paths(): workspace=str(workspace), daily_dir="daily", changed_paths=[ok.relative_to(workspace).as_posix(), failed.relative_to(workspace).as_posix()], + existing={ + path.relative_to(workspace).as_posix(): path.stat().st_mtime for path in (ok, failed, day_index) + }, failed_paths=[failed.relative_to(workspace).as_posix()], modified_paths=["digest/procedure/example.md"], integrate_results=[ @@ -753,6 +756,7 @@ def test_finish_does_not_readd_a_failed_day_index(tmp_path): workspace=str(tmp_path), daily_dir="daily", changed_paths=[rel_path], + existing={rel_path: day_index.stat().st_mtime}, failed_paths=[rel_path], errors=["extract unavailable"], ) @@ -773,13 +777,14 @@ def test_finish_keeps_skipped_agent_output_successful(tmp_path): async def run(): rel_path = "daily/2026-05-28/source.md" - _touch(tmp_path / rel_path) + source = _touch(tmp_path / rel_path) state = DreamState( date="2026-05-28", dates=["2026-05-28"], workspace=str(tmp_path), daily_dir="daily", changed_paths=[rel_path], + existing={rel_path: source.stat().st_mtime}, skipped_units=[{"name": "unit", "reason": "unusable receipt"}], warnings=["unit 'unit' skipped: unusable agent receipt"], ) diff --git a/tests/unit/test_dream_checkpoint_versions.py b/tests/unit/test_dream_checkpoint_versions.py new file mode 100644 index 00000000..38afe875 --- /dev/null +++ b/tests/unit/test_dream_checkpoint_versions.py @@ -0,0 +1,205 @@ +"""Auto Dream checkpoints must acknowledge the versions observed by Extract.""" + +import asyncio +import copy +import json +import os +from unittest.mock import AsyncMock, Mock + +import frontmatter +import pytest +from agentscope.model import ChatModelBase + +from reme.application import Application +from reme.components.agent_wrapper import BaseAgentWrapper +from reme.components.application_context import ApplicationContext +from reme.components.file_catalog import LocalFileCatalog +from reme.components.runtime_context import RuntimeContext +from reme.config import resolve_app_config +from reme.schema import DreamState +from reme.steps.evolve.dream.finish import DreamFinishStep + +DAY = "2026-05-28" +SOURCE = f"daily/{DAY}/deployment.md" +TARGET = "digest/wiki/deployment.md" +OLD = "Deploy to the staging cluster." +NEW = "Deploy to the production cluster." + + +def _app(workspace): + """Build the default dream/write jobs with real local storage and no model providers.""" + defaults = resolve_app_config(config="default", log_config=False) + types = ("tokenizer", "keyword_index", "file_graph", "tag_index", "file_store", "file_catalog") + return Application( + workspace_dir=str(workspace), + enable_logo=False, + log_to_file=False, + log_to_console=False, + service={"backend": "http", "web_enabled": False, "mcp_enabled": False}, + components={name: copy.deepcopy(defaults["components"][name]) for name in types}, + jobs={name: copy.deepcopy(defaults["jobs"][name]) for name in ("auto_dream", "write")}, + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("edit_phase", ["extract", "integrate", None]) +async def test_dream_reprocesses_concurrent_source_edits_after_restart(tmp_path, edit_phase): + """A real catalog restart retries later edits while unchanged sources stay deduplicated.""" + source = tmp_path / SOURCE + source.parent.mkdir(parents=True) + source.write_text(frontmatter.dumps(frontmatter.Post(OLD, name="deployment", description="Deployment policy"))) + scanned_mtime = source.stat().st_mtime + entered, release = asyncio.Event(), asyncio.Event() + calls = [] + + def agent_for(app, blocked=False): + async def reply(message, **kwargs): + if "list_tags" in kwargs["job_tools"]: + return {"result": "No tags added by the controlled provider"} + phase = "extract" if kwargs["job_tools"] == ["read"] else "integrate" + prompt = str(message) + calls.append((phase, prompt)) + fact = NEW if NEW in prompt else OLD + if blocked and phase == (edit_phase or "integrate"): + assert OLD in prompt and NEW not in prompt + entered.set() + await release.wait() + if phase == "extract": + unit = {"name": "deployment", "bucket": "wiki", "summary": fact, "paths": [SOURCE]} + return {"result": json.dumps({"units": [unit]})} + action = "CORRECT" if (tmp_path / TARGET).exists() else "CREATE" + response = await app.run_job("write", path=TARGET, content=fact) + assert response.success, response.answer + return {"result": json.dumps({"action": action, "target_path": TARGET})} + + agent = AsyncMock(spec=BaseAgentWrapper) + agent.reply.side_effect = reply + return agent + + app = _app(tmp_path) + await app.start() + task = asyncio.create_task( + app.run_job( + "auto_dream", + date=DAY, + scan_days=1, + agent_wrapper=agent_for(app, True), + as_llm=Mock(spec=ChatModelBase), + ), + ) + try: + await asyncio.wait_for(entered.wait(), timeout=5) + if edit_phase is not None: + response = await app.run_job( + "write", + path=SOURCE, + content=frontmatter.dumps(frontmatter.Post(NEW, name="deployment", description="Deployment policy")), + ) + assert response.success, response.answer + os.utime(source, (scanned_mtime + 2, scanned_mtime + 2)) + release.set() + first = await asyncio.wait_for(task, timeout=5) + assert first.success, first.answer + assert first.metadata["dream"]["warnings"] == [] + nodes = await app.context.components["file_catalog"]["dream"].get_nodes([SOURCE]) + first_checkpoint = nodes[0].st_mtime + if edit_phase == "integrate": + assert OLD in (tmp_path / TARGET).read_text() + finally: + release.set() + if not task.done(): + task.cancel() + await asyncio.gather(task, return_exceptions=True) + await app.close() + + before_retry = len(calls) + app = _app(tmp_path) + await app.start() + try: + second = await app.run_job( + "auto_dream", + date=DAY, + scan_days=1, + agent_wrapper=agent_for(app), + as_llm=Mock(spec=ChatModelBase), + ) + assert second.success, second.answer + assert second.metadata["dream"]["warnings"] == [] + assert first_checkpoint == scanned_mtime + if edit_phase is None: + assert second.metadata["dream"]["changed_paths"] == [] + assert len(calls) == before_retry + else: + assert second.metadata["dream"]["integrate_results"][0]["action"] == "CORRECT" + assert SOURCE in second.metadata["dream"]["changed_paths"] + assert NEW in (tmp_path / TARGET).read_text() + assert NEW in source.read_text() + assert len(calls) == before_retry + 2 + nodes = await app.context.components["file_catalog"]["dream"].get_nodes([SOURCE]) + assert nodes[0].st_mtime == source.stat().st_mtime + finally: + await app.close() + + +@pytest.mark.asyncio +async def test_finish_keeps_scanned_versions_for_changed_sources_and_day_indexes(tmp_path): + """Supplemental indexes obey the same snapshot and failed-source rules as notes.""" + good, failed, index, unscanned = "daily/good.md", "daily/failed.md", f"daily/{DAY}.md", "daily/2026-05-27.md" + observed = {} + for rel in (good, failed, index): + path = tmp_path / rel + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text("scanned input") + observed[rel] = path.stat().st_mtime + path.write_text("later input") + os.utime(path, (observed[rel] + 2, observed[rel] + 2)) + (tmp_path / unscanned).write_text("created after the scan") + state = DreamState( + date=DAY, + dates=["2026-05-27", DAY], + workspace=str(tmp_path), + daily_dir="daily", + changed_paths=[good, failed], + failed_paths=[failed], + existing=observed, + ) + context = ApplicationContext(workspace_dir=str(tmp_path)) + catalog = LocalFileCatalog(name="dream", app_context=context) + await catalog.start() + try: + response = await DreamFinishStep(app_context=context)( + RuntimeContext(dream=state.model_dump(), file_catalog=catalog), + ) + assert response.success + assert response.metadata["dream"]["checkpoint_paths"] == [good] + assert {node.path: node.st_mtime for node in await catalog.get_nodes()} == { + good: observed[good], + index: observed[index], + } + finally: + await catalog.close() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("replacement", [False, True]) +async def test_finish_handles_a_removed_or_replaced_source(tmp_path, replacement): + """A replacement stays eligible; a missing source cannot create a fresh checkpoint.""" + source = tmp_path / SOURCE + source.parent.mkdir(parents=True) + source.write_text(OLD) + observed = source.stat().st_mtime + source.unlink() + if replacement: + source.write_text(NEW) + os.utime(source, (observed + 2, observed + 2)) + state = DreamState(date=DAY, workspace=str(tmp_path), changed_paths=[SOURCE], existing={SOURCE: observed}) + context = ApplicationContext(workspace_dir=str(tmp_path)) + catalog = LocalFileCatalog(name="dream", app_context=context) + await catalog.start() + try: + await DreamFinishStep(app_context=context)(RuntimeContext(dream=state.model_dump(), file_catalog=catalog)) + assert {node.path: node.st_mtime for node in await catalog.get_nodes()} == ( + {SOURCE: observed} if replacement else {} + ) + finally: + await catalog.close()