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 <jinli.yl@alibaba-inc.com>

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

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 文件打标，

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

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"],
         )

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()