fix(auto-dream): preserve retries after invalid integration receipts (#654)

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}"

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

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."""