diff --git a/reme/schema/dream.py b/reme/schema/dream.py index 361417d9..15628576 100644 --- a/reme/schema/dream.py +++ b/reme/schema/dream.py @@ -84,6 +84,10 @@ class DreamState(BaseModel): skipped_units: list[dict] = Field(default_factory=list) nodes_created: list[str] = Field(default_factory=list) nodes_updated: list[str] = Field(default_factory=list) + modified_paths: list[str] = Field( + default_factory=list, + description="Durable digest or interests files detected as created or changed during this run.", + ) failed_units: list[dict] = Field(default_factory=list) failed_paths: list[str] = Field(default_factory=list) interests_path: str = "" diff --git a/reme/steps/evolve/dream/finish.py b/reme/steps/evolve/dream/finish.py index cb4f699a..7ea79995 100644 --- a/reme/steps/evolve/dream/finish.py +++ b/reme/steps/evolve/dream/finish.py @@ -51,6 +51,7 @@ class DreamFinishStep(BaseStep): store_state(self, state) self.context.response.success = not state.errors self.context.response.answer = state.summary + self.context.response.metadata["modified"] = bool(state.modified_paths) self.logger.info( f"[{self.name}] finish success={self.context.response.success} " f"checkpointed={len(state.checkpoint_paths)} failed_units={len(state.failed_units)} " diff --git a/reme/steps/evolve/dream/integrate.py b/reme/steps/evolve/dream/integrate.py index 015b26fc..763467f9 100644 --- a/reme/steps/evolve/dream/integrate.py +++ b/reme/steps/evolve/dream/integrate.py @@ -15,7 +15,11 @@ _TOOLS = ("node_search", "read", "frontmatter_read", "write", "edit", "frontmatt def _snapshot_digest(workspace: Path, digest_dir: str) -> dict[str, tuple[int, int]]: - """Capture all supported digest buckets for best-effort side-effect recovery.""" + """Capture lightweight digest fingerprints for side-effect recovery.""" + # Content hashes are more exact, but auto-dream snapshots the whole digest + # tree around agent attempts, so hashing would turn each snapshot into a + # full-tree read. mtime_ns + size is an intentional, best-effort mutation + # signal that keeps the existing file-side-effect recovery inexpensive. snapshot: dict[str, tuple[int, int]] = {} for bucket in DreamBucketEnum: root = workspace / digest_dir / bucket.value @@ -32,11 +36,21 @@ def _snapshot_digest(workspace: Path, digest_dir: str) -> dict[str, tuple[int, i return snapshot -def _changed_digest_paths(before: dict[str, tuple[int, int]], after: dict[str, tuple[int, int]]) -> list[str]: +def _changed_digest_paths( + before: dict[str, tuple[int, int]], + after: dict[str, tuple[int, int]], +) -> list[str]: """Return files created or changed during one integration attempt.""" return sorted(path for path, metadata in after.items() if before.get(path) != metadata) +def _record_modified_paths(state, paths: list[str]) -> None: + """Record detected file changes once while preserving discovery order.""" + for path in paths: + if path not in state.modified_paths: + state.modified_paths.append(path) + + @R.register("dream_integrate_step") class DreamIntegrateStep(BaseStep): """Integrate each extracted unit into digest memory.""" @@ -102,8 +116,8 @@ class DreamIntegrateStep(BaseStep): # Keep the original baseline across the retry so a file created by attempt one # cannot be misclassified as a pre-existing cross-bucket UPDATE on attempt two. unit_before = _snapshot_digest(workspace, digest_dir) + before = unit_before for attempt in range(2): - before = _snapshot_digest(workspace, digest_dir) try: result = await self.agent_wrapper.reply( self.prompt_format( @@ -124,7 +138,9 @@ class DreamIntegrateStep(BaseStep): job_tools=list(_TOOLS), ) except Exception as e: # noqa: BLE001 - changed = _changed_digest_paths(before, _snapshot_digest(workspace, digest_dir)) + after = _snapshot_digest(workspace, digest_dir) + changed = _changed_digest_paths(before, after) + _record_modified_paths(state, changed) created = len(changed) == 1 and changed[0] not in unit_before if len(changed) == 1 and self._valid_target( workspace, @@ -145,6 +161,7 @@ class DreamIntegrateStep(BaseStep): f"[{self.name}] unit {index}/{len(state.units)} attempt 1 had no recoverable file change; " f"retrying once: {type(e).__name__}: {e}", ) + before = after continue self._record_failure(state, unit, paths, e) self.logger.error(f"[{self.name}] unit {index}/{len(state.units)} failed: {type(e).__name__}: {e}") @@ -163,7 +180,9 @@ class DreamIntegrateStep(BaseStep): ): raise ValueError(f"invalid or missing digest target_path: {outcome.target_path!r}") except Exception as e: # noqa: BLE001 - changed = _changed_digest_paths(before, _snapshot_digest(workspace, digest_dir)) + after = _snapshot_digest(workspace, digest_dir) + changed = _changed_digest_paths(before, after) + _record_modified_paths(state, changed) created = len(changed) == 1 and changed[0] not in unit_before if len(changed) == 1 and self._valid_target( workspace, @@ -184,6 +203,7 @@ class DreamIntegrateStep(BaseStep): f"[{self.name}] unit {index}/{len(state.units)} attempt 1 returned an invalid receipt; " "retrying once", ) + before = after continue # After the bounded retry, checkpoint the source so malformed input cannot loop forever. self._record_skipped(state, unit, bucket, paths, e) @@ -193,6 +213,8 @@ class DreamIntegrateStep(BaseStep): ) return + after = _snapshot_digest(workspace, digest_dir) + _record_modified_paths(state, _changed_digest_paths(unit_before, after)) self._append_result( state, unit, diff --git a/reme/steps/evolve/dream/topics.py b/reme/steps/evolve/dream/topics.py index 9f843f30..9fbea087 100644 --- a/reme/steps/evolve/dream/topics.py +++ b/reme/steps/evolve/dream/topics.py @@ -94,9 +94,12 @@ class DreamTopicsStep(BaseStep): "diversity_days": diversity_days, "topics": topics, } + before_content = abs_path.read_bytes() if abs_path.is_file() else None self.logger.info(f"[{self.name}] write yaml start path={rel_path}") write_yaml(abs_path, payload) self.logger.info(f"[{self.name}] write yaml done path={rel_path}") + if before_content != abs_path.read_bytes() and rel_path not in state.modified_paths: + state.modified_paths.append(rel_path) self.logger.info(f"[{self.name}] refresh index start date={target_day} daily_dir={state.daily_dir}") await refresh_day_index(self.file_store, target_day, state.daily_dir) self.logger.info(f"[{self.name}] refresh index done date={target_day}") diff --git a/tests/unit/test_auto_dream.py b/tests/unit/test_auto_dream.py index c515dfa6..a47f40f8 100644 --- a/tests/unit/test_auto_dream.py +++ b/tests/unit/test_auto_dream.py @@ -17,7 +17,7 @@ from reme.components.runtime_context import RuntimeContext from reme.schema import DreamState, FileNode from reme.steps.evolve.dream.extract import DreamExtractStep from reme.steps.evolve.dream.finish import DreamFinishStep -from reme.steps.evolve.dream.integrate import DreamIntegrateStep +from reme.steps.evolve.dream.integrate import DreamIntegrateStep, _snapshot_digest from reme.steps.evolve.dream.proactive import ProactiveStep from reme.steps.evolve.dream.topics import DreamTopicsStep from reme.steps.evolve.dream.utils import load_yaml_topics, parse_structured_reply, recent_dates, scan_day_files @@ -321,12 +321,17 @@ def test_integrate_retries_one_invalid_receipt(tmp_path): ) step = DreamIntegrateStep(agent_wrapper=agent) - await step._integrate_one(state, unit, 1, tmp_path, "digest") # pylint: disable=protected-access + with patch( + "reme.steps.evolve.dream.integrate._snapshot_digest", + wraps=_snapshot_digest, + ) as snapshot: + await step._integrate_one(state, unit, 1, tmp_path, "digest") # pylint: disable=protected-access assert target.is_file() assert state.integrate_results[0]["target_path"] == "digest/wiki/unit.md" assert state.skipped_units == [] assert agent.calls == 2 + assert snapshot.call_count == 3 asyncio.run(run()) @@ -353,6 +358,7 @@ def test_integrate_invalid_receipt_recovers_one_changed_digest_file(tmp_path): assert state.integrate_results[0]["action"] == "CREATE" assert state.integrate_results[0]["target_path"] == "digest/wiki/unit.md" assert state.nodes_created == ["digest/wiki/unit.md"] + assert state.modified_paths == ["digest/wiki/unit.md"] assert len(state.warnings) == 1 asyncio.run(run()) @@ -379,6 +385,7 @@ def test_integrate_recovers_an_update_to_another_bucket(tmp_path): assert state.integrate_results[0]["action"] == "UPDATED" assert state.integrate_results[0]["target_path"] == "digest/procedure/unit.md" assert state.nodes_updated == ["digest/procedure/unit.md"] + assert state.modified_paths == ["digest/procedure/unit.md"] assert step.agent_wrapper.calls == 1 asyncio.run(run()) @@ -402,6 +409,7 @@ def test_integrate_does_not_recover_a_create_in_another_bucket(tmp_path): assert state.integrate_results == [] assert len(state.skipped_units) == 1 assert state.nodes_created == [] + assert state.modified_paths == ["digest/procedure/unit.md"] asyncio.run(run()) @@ -423,11 +431,17 @@ def test_integrate_accepts_cross_bucket_update_receipt(tmp_path): ) step = DreamIntegrateStep(agent_wrapper=agent) - await step._integrate_one(state, unit, 1, tmp_path, "digest") # pylint: disable=protected-access + with patch( + "reme.steps.evolve.dream.integrate._snapshot_digest", + wraps=_snapshot_digest, + ) as snapshot: + await step._integrate_one(state, unit, 1, tmp_path, "digest") # pylint: disable=protected-access assert state.skipped_units == [] assert state.integrate_results[0]["action"] == "REFINE" assert state.nodes_updated == ["digest/procedure/unit.md"] + assert state.modified_paths == [] + assert snapshot.call_count == 2 asyncio.run(run()) @@ -497,11 +511,61 @@ def test_topics_step_writes_only_target_date_interests(): assert target.is_file() assert old_interests.read_text(encoding="utf-8") == "date: 2026-05-26\ntopics: []\n" assert dream["interests_paths"] == ["daily/2026-05-28/interests.yaml"] + assert dream["modified_paths"] == ["daily/2026-05-28/interests.yaml"] assert yaml.safe_load(target.read_text(encoding="utf-8"))["date"] == "2026-05-28" asyncio.run(run()) +def test_topics_same_content_is_not_modified(tmp_path): + """Rewriting deterministic interests content does not count as a user-visible change.""" + + async def run(): + topic = {"title": "Topic", "reason": "Reason", "paths": ["daily/source.md"]} + step = DreamTopicsStep() + + with patch("reme.steps.evolve.dream.topics.refresh_day_index", return_value={}): + first = await step( + RuntimeContext( + dream=DreamState( + date="2026-05-28", + workspace=str(tmp_path), + daily_dir="daily", + topics=[topic], + ).model_dump(), + file_store=_FileStore(tmp_path), + ), + ) + second = await step( + RuntimeContext( + dream=DreamState( + date="2026-05-28", + workspace=str(tmp_path), + daily_dir="daily", + topics=[topic], + ).model_dump(), + file_store=_FileStore(tmp_path), + ), + ) + third = await step( + RuntimeContext( + dream=DreamState( + date="2026-05-28", + workspace=str(tmp_path), + daily_dir="daily", + topics=[topic], + ).model_dump(), + file_store=_FileStore(tmp_path), + ), + ) + + assert first.metadata["dream"]["modified_paths"] == ["daily/2026-05-28/interests.yaml"] + assert second.metadata["dream"]["modified_paths"] == ["daily/2026-05-28/interests.yaml"] + assert third.metadata["dream"]["modified_paths"] == [] + + asyncio.run(run()) + + def test_topics_agent_failure_falls_back_to_candidates(tmp_path): """Topic ranking remains best-effort when the optional agent is unavailable.""" @@ -700,6 +764,7 @@ def test_finish_does_not_checkpoint_failed_changed_paths(): changed_paths=[ok.relative_to(workspace).as_posix(), failed.relative_to(workspace).as_posix()], failed_paths=[failed.relative_to(workspace).as_posix()], interests_paths=[interests.relative_to(workspace).as_posix()], + modified_paths=["digest/procedure/example.md"], integrate_results=[ { "action": "CREATE", @@ -723,6 +788,7 @@ def test_finish_does_not_checkpoint_failed_changed_paths(): assert interests.relative_to(workspace).as_posix() in upserted assert day_index.relative_to(workspace).as_posix() in upserted assert catalog.dumps == 1 + assert resp.metadata["modified"] is True asyncio.run(run()) @@ -749,6 +815,7 @@ def test_finish_does_not_readd_a_failed_day_index(tmp_path): assert response.success is False assert not catalog.upserts assert response.metadata["dream"]["checkpoint_paths"] == [] + assert response.metadata["modified"] is False asyncio.run(run())