From 925ef0dad17adf2250b866716df2be00b4cafdc1 Mon Sep 17 00:00:00 2001 From: DiegoCluv7 <161216254+iluv7@users.noreply.github.com> Date: Thu, 22 Oct 2026 11:22:32 +0800 Subject: [PATCH] fix(auto-dream): preserve retries after invalid integration receipts (#654) --- reme/steps/evolve/dream/integrate.py | 20 +---- tests/unit/test_auto_dream.py | 72 ++++++++++++--- tests/unit/test_dream_checkpoint_versions.py | 95 +++++++++++++++++++- 3 files changed, 159 insertions(+), 28 deletions(-) diff --git a/reme/steps/evolve/dream/integrate.py b/reme/steps/evolve/dream/integrate.py index f43dbea7..f3946318 100644 --- a/reme/steps/evolve/dream/integrate.py +++ b/reme/steps/evolve/dream/integrate.py @@ -214,10 +214,10 @@ class DreamIntegrateStep(BaseStep): ) before = after continue - # After the bounded retry, checkpoint the source so malformed input cannot loop forever. - self._record_skipped(state, unit, bucket, paths, e) - self.logger.warning( - f"[{self.name}] unit {index}/{len(state.units)} skipped invalid receipt after retry: " + # Without a valid receipt or recoverable write, keep sources eligible for retry. + self._record_failure(state, unit, paths, e) + self.logger.error( + f"[{self.name}] unit {index}/{len(state.units)} failed invalid receipt after retry: " f"{type(e).__name__}: {e}", ) return @@ -339,18 +339,6 @@ class DreamIntegrateStep(BaseStep): note=f"Recovered from {type(error).__name__}: agent receipt unavailable", ) - @classmethod - def _record_skipped(cls, state, unit: dict, bucket: str, paths: list[str], error: Exception) -> None: - message = f"unit {unit.get('name', '')!r} skipped: unusable agent receipt ({type(error).__name__})" - key = cls._unit_key(unit, bucket, paths) - if not any( - cls._unit_key(item, str(item.get("bucket") or ""), [str(p) for p in item.get("paths") or []]) == key - for item in state.skipped_units - ): - state.skipped_units.append({**unit, "bucket": bucket, "paths": paths, "reason": message}) - if message not in state.warnings: - state.warnings.append(message) - @staticmethod def _record_failure(state, unit: dict, paths: list[str], error: Exception) -> None: message = f"{type(error).__name__}: {error}" diff --git a/tests/unit/test_auto_dream.py b/tests/unit/test_auto_dream.py index a808b615..788444fd 100644 --- a/tests/unit/test_auto_dream.py +++ b/tests/unit/test_auto_dream.py @@ -498,27 +498,71 @@ def test_extract_retries_one_unusable_receipt(tmp_path): asyncio.run(run()) -def test_integrate_invalid_receipt_without_file_change_is_skipped(tmp_path): - """An unusable integration receipt is a skipped unit, not a failed unit.""" +@pytest.mark.parametrize( + "receipt", + ["{}", "not a receipt", '{"action": "CREATE", "target_path": "digest/wiki/missing.md"}'], +) +def test_integrate_invalid_receipt_without_file_change_fails(tmp_path, receipt): + """An unusable integration receipt cannot acknowledge unprocessed source files.""" async def run(): unit = {"name": "unit", "bucket": "wiki", "summary": "summary", "paths": ["daily/a.md"]} state = DreamState(units=[unit]) step = DreamIntegrateStep() - step.agent_wrapper = _ReplyAgent() + step.agent_wrapper = _ReplyAgent({"result": receipt}) await step._integrate_one(state, unit, 1, tmp_path, "digest") # pylint: disable=protected-access assert state.integrate_results == [] - assert len(state.skipped_units) == 1 - assert state.failed_units == [] - assert not state.failed_paths - assert len(state.warnings) == 1 + assert state.skipped_units == [] + assert len(state.failed_units) == 1 + assert state.failed_units[0]["name"] == unit["name"] + assert state.failed_paths == unit["paths"] + assert len(state.errors) == 1 + assert state.warnings == [] assert step.agent_wrapper.calls == 2 asyncio.run(run()) +@pytest.mark.asyncio +@pytest.mark.parametrize("shared_source", [False, True]) +async def test_integrate_failure_only_blocks_affected_sources(tmp_path, monkeypatch, shared_source): + """A failed unit prevents checkpointing all its sources, even one used by a successful unit.""" + good, bad, target = "daily/good.md", "daily/bad.md", "digest/wiki/good.md" + observed = {rel: _touch(tmp_path / rel).stat().st_mtime for rel in (good, bad)} + failed_paths = [bad, good] if shared_source else [bad] + units = [ + {"name": "good", "bucket": "wiki", "summary": "good memory", "paths": [good]}, + {"name": "bad", "bucket": "wiki", "summary": "bad memory", "paths": failed_paths}, + ] + + def integrate_good(): + _touch(tmp_path / target, "integrated memory") + return {"result": json.dumps({"action": "CREATE", "target_path": target})} + + agent = _SequenceAgent(integrate_good, {"result": "{}"}, {"result": "{}"}) + app_context = ApplicationContext(workspace_dir=str(tmp_path)) + state = DreamState(units=units, changed_paths=[good, bad], existing=observed, workspace=str(tmp_path)) + catalog = _Catalog() + context = RuntimeContext(dream=state.model_dump(), file_catalog=catalog, file_store=_FileStore(tmp_path)) + monkeypatch.setattr("reme.steps.evolve.dream.integrate.llm_available", lambda _step: True) + + response = await DreamIntegrateStep(app_context=app_context, agent_wrapper=agent)(context) + assert response.success is False + response = await DreamFinishStep(app_context=app_context)(context) + + dream = response.metadata["dream"] + assert response.success is False + assert len(dream["integrate_results"]) == 1 + assert len(dream["failed_units"]) == 1 + assert set(dream["failed_paths"]) == set(failed_paths) + assert dream["checkpoint_paths"] == ([] if shared_source else [good]) + assert {node.path for node in catalog.upserts} == (set() if shared_source else {good}) + assert (tmp_path / target).read_text(encoding="utf-8") == "integrated memory" + assert agent.calls == 3 + + def test_integrate_retries_one_invalid_receipt(tmp_path): """A transient invalid integration receipt gets one bounded retry.""" @@ -622,7 +666,10 @@ def test_integrate_does_not_recover_a_create_in_another_bucket(tmp_path): await step._integrate_one(state, unit, 1, tmp_path, "digest") # pylint: disable=protected-access assert state.integrate_results == [] - assert len(state.skipped_units) == 1 + assert state.skipped_units == [] + assert len(state.failed_units) == 1 + assert state.failed_paths == unit["paths"] + assert state.errors assert state.nodes_created == [] assert state.modified_paths == ["digest/procedure/unit.md"] @@ -762,13 +809,13 @@ async def test_integrate_clears_changes_when_skipping_or_missing_llm(tmp_path, m @pytest.mark.asyncio @pytest.mark.parametrize("job_name", ["auto_dream", "dream_cron"]) @pytest.mark.parametrize("tag_fails", [False, True]) -@pytest.mark.parametrize("integrate_fails", [False, True]) +@pytest.mark.parametrize("integration_failure", [None, "exception", "invalid_receipt"]) async def test_dream_jobs_tag_outputs_and_preserve_checkpoint_results( tmp_path, monkeypatch, job_name, tag_fails, - integrate_fails, + integration_failure, ): """Both configured pipelines tag durable outputs without changing dream success.""" app_context = ApplicationContext(workspace_dir=str(tmp_path)) @@ -798,8 +845,10 @@ async def test_dream_jobs_tag_outputs_and_preserve_checkpoint_results( for path in targets: previous = (tmp_path / path).read_text(encoding="utf-8") if (tmp_path / path).exists() else "" _touch(tmp_path / path, previous + "ReMe evidence\n") - if integrate_fails: + if integration_failure == "exception": raise RuntimeError("integration unavailable") + if integration_failure == "invalid_receipt": + return {"result": "{}"} return {"result": json.dumps({"action": "CREATE", "target_path": targets[0]})} agent = _ReplyAgent() @@ -821,6 +870,7 @@ async def test_dream_jobs_tag_outputs_and_preserve_checkpoint_results( finally: await job.close() + integrate_fails = bool(integration_failure) assert response.success is not integrate_fails assert response.answer.startswith("AutoDream failed" if integrate_fails else "AutoDream completed") assert response.metadata["modified"] is True diff --git a/tests/unit/test_dream_checkpoint_versions.py b/tests/unit/test_dream_checkpoint_versions.py index 38afe875..ec97e112 100644 --- a/tests/unit/test_dream_checkpoint_versions.py +++ b/tests/unit/test_dream_checkpoint_versions.py @@ -16,7 +16,7 @@ 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.schema import DreamState, FileNode from reme.steps.evolve.dream.finish import DreamFinishStep DAY = "2026-05-28" @@ -141,6 +141,99 @@ async def test_dream_reprocesses_concurrent_source_edits_after_restart(tmp_path, await app.close() +@pytest.mark.asyncio +@pytest.mark.parametrize("older_checkpoint", [False, True]) +@pytest.mark.parametrize("recover_on_attempt", [None, 1, 2]) +async def test_invalid_integration_preserves_retry_after_restart(tmp_path, older_checkpoint, recover_on_attempt): + """Unrecoverable integration preserves catalog versions; actual writes can still recover.""" + source = tmp_path / SOURCE + source.parent.mkdir(parents=True) + original = frontmatter.dumps(frontmatter.Post(OLD, name="deployment", description="Deployment policy")) + source.write_text(original) + index = f"daily/{DAY}.md" + (tmp_path / index).write_text("day index") + + def agent_for(app, write_attempt): + calls = {"extract": 0, "integrate": 0} + + async def reply(_message, **kwargs): + if "list_tags" in kwargs["job_tools"]: + return {"result": "No tags added by the controlled provider"} + if kwargs["job_tools"] == ["read"]: + calls["extract"] += 1 + unit = {"name": "deployment", "bucket": "wiki", "summary": OLD, "paths": [SOURCE, index]} + return {"result": json.dumps({"units": [unit]})} + calls["integrate"] += 1 + if calls["integrate"] == write_attempt: + written = await app.run_job("write", path=TARGET, content=OLD) + assert written.success, written.answer + return {"result": "{}"} + + agent = AsyncMock(spec=BaseAgentWrapper) + agent.reply.side_effect = reply + return agent, calls + + app = _app(tmp_path) + await app.start() + previous = {rel: (tmp_path / rel).stat().st_mtime - 2 for rel in (SOURCE, index)} if older_checkpoint else {} + try: + catalog = app.context.components["file_catalog"]["dream"] + await catalog.upsert([FileNode(path=rel, st_mtime=mtime) for rel, mtime in previous.items()]) + agent, calls = agent_for(app, recover_on_attempt) + response = await app.run_job( + "auto_dream", + date=DAY, + scan_days=1, + agent_wrapper=agent, + as_llm=Mock(spec=ChatModelBase), + ) + state = response.metadata["dream"] + assert response.success is (recover_on_attempt is not None) + assert state["skipped_units"] == [] + assert calls == {"extract": 1, "integrate": recover_on_attempt or 2} + assert source.read_text() == original + if recover_on_attempt is None: + assert response.answer.startswith("AutoDream failed") + assert len(state["failed_units"]) == 1 + assert set(state["failed_paths"]) == {SOURCE, index} + assert state["checkpoint_paths"] == [] + assert state["errors"] + assert {n.path: n.st_mtime for n in await catalog.get_nodes()} == previous + assert not (tmp_path / TARGET).exists() + else: + assert response.answer.startswith("AutoDream completed with warnings") + assert state["failed_units"] == [] + assert state["failed_paths"] == [] + assert not state["errors"] + assert set(state["checkpoint_paths"]) == {SOURCE, index} + assert state["integrate_results"][0]["target_path"] == TARGET + assert OLD in (tmp_path / TARGET).read_text() + finally: + await app.close() + + app = _app(tmp_path) + await app.start() + try: + catalog = app.context.components["file_catalog"]["dream"] + if recover_on_attempt is None: + assert {n.path: n.st_mtime for n in await catalog.get_nodes()} == previous + agent, calls = agent_for(app, 1) + response = await app.run_job( + "auto_dream", + date=DAY, + scan_days=1, + agent_wrapper=agent, + as_llm=Mock(spec=ChatModelBase), + ) + assert response.success, response.answer + assert calls == {"extract": int(recover_on_attempt is None), "integrate": int(recover_on_attempt is None)} + assert {SOURCE, index} <= {n.path for n in await catalog.get_nodes()} + assert OLD in (tmp_path / TARGET).read_text() + assert source.read_text() == original + 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."""