scriptoriumadmin/ReMe
1
0
Fork 0
mirror of https://github.com/agentscope-ai/ReMe.git synced 2026-10-08 03:10:24 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

feat(evolve): report auto-dream file changes (#458)

Browse source 
* feat(evolve): report auto-dream content changes

* perf(evolve): use lightweight dream snapshots
This commit is contained in:
jinliyl 2026-08-19 15:44:06 +08:00 • committed by GitHub
parent 6b9a75267b
commit d3aee1adf5
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 105 additions and 8 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
reme/schema/dream.py
View file

@ -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 = ""

1
reme/steps/evolve/dream/finish.py
View file

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

32
reme/steps/evolve/dream/integrate.py
View file

@ -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,

3
reme/steps/evolve/dream/topics.py
View file

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

73
tests/unit/test_auto_dream.py
View file

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