From 05958d4d8bc2d87df764438d80304bb67f5ab9ba Mon Sep 17 00:00:00 2001 From: Sen Huang <48879559+ployts@users.noreply.github.com> Date: Fri, 11 Sep 2026 14:08:29 +0800 Subject: [PATCH] feat(evolve): tag auto-dream outputs (#538) --- docs/en/auto_dream.md | 17 ++- docs/en/auto_link.md | 1 + docs/en/auto_memory.md | 8 ++ docs/en/reme_scene.md | 4 +- docs/zh/auto_dream.md | 15 ++- docs/zh/auto_link.md | 1 + docs/zh/auto_memory.md | 6 + docs/zh/reme_scene.md | 5 +- reme/config/default.yaml | 2 + reme/steps/evolve/dream/integrate.py | 9 ++ tests/unit/test_auto_dream.py | 181 ++++++++++++++++++++++++++- tests/unit/test_auto_memory_cc.py | 113 +++++++++++++++++ 12 files changed, 355 insertions(+), 7 deletions(-) diff --git a/docs/en/auto_dream.md b/docs/en/auto_dream.md index 1f0a0f5d..33b0efa1 100644 --- a/docs/en/auto_dream.md +++ b/docs/en/auto_dream.md @@ -37,6 +37,7 @@ auto_dream: - backend: dream_integrate_step - backend: dream_finish_step file_catalog: dream + - backend: auto_tag_step ``` Parameters: @@ -71,7 +72,7 @@ The main outputs are: | `digest/wiki/*.md` | General knowledge, concepts, observations, and decision precedents. | | `metadata/file_catalog/dream*` | Dream-specific catalog used to detect changes in daily inputs. | -## Three Stages +## Four Stages ### 1. Extract @@ -134,6 +135,17 @@ 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. +### 4. Auto Tag + +After Finish, both `auto_dream` and `dream_cron` run `auto_tag_step` on Markdown digest files actually created or modified +during integration, including writes recovered after agent errors. Repeated writes to one file are tagged once. The +Step uses the same request-scoped `changes` contract as [Auto Memory](./auto_memory.md) and writes entity tags to the +configured frontmatter key, `memory_tags` by default. Unchanged files and daily source notes are not tagged by Dream. + +Tagging diagnostics appear in `metadata.auto_tag`. Per-file tagging failures preserve the dream answer, success status, +and checkpoint decisions. A later run without file changes does not automatically retry failed tagging. Tag-index +updates follow the existing asynchronous file watcher. + ## Running Auto Dream CLI: @@ -167,6 +179,7 @@ jobs: - backend: dream_integrate_step - backend: dream_finish_step file_catalog: dream + - backend: auto_tag_step ``` ## Important Boundaries @@ -182,4 +195,4 @@ the workspace-relative wikilink semantics described in `auto_dream` does not invent an overview from nothing. Only content that actually appears in daily input and is extracted as a memory unit can enter digest. -The complete flow depends on an LLM for Extract and Integrate. +The complete flow depends on an LLM for Extract, Integrate, and Auto Tag. diff --git a/docs/en/auto_link.md b/docs/en/auto_link.md index d704b0fa..0b3bde0f 100644 --- a/docs/en/auto_link.md +++ b/docs/en/auto_link.md @@ -19,6 +19,7 @@ auto_dream: - dream_extract_step - dream_integrate_step # where auto_link actually happens - dream_finish_step + - auto_tag_step ``` The Integrate stage processes each unit independently. A unit is written to exactly one target digest node, but that diff --git a/docs/en/auto_memory.md b/docs/en/auto_memory.md index ffe3db95..6f0235bc 100644 --- a/docs/en/auto_memory.md +++ b/docs/en/auto_memory.md @@ -110,5 +110,13 @@ reme auto_memory \ ## What Happens Next +The default `auto_memory` and `auto_memory_cc` jobs run `auto_tag_step` after recording memory. Only a daily note that +was actually created or modified is tagged, using its final path after any rename. Claude Code callers still pass only +`session_id`; repeated Stop events with no new messages skip both memory generation and tagging. + +Tags describe the document's central entities and are stored in the configured frontmatter key (`memory_tags` by default). +Per-file tagging failures are reported in `metadata.auto_tag` while preserving the memory response. Calls without note +changes do not automatically retry failed tagging; the existing file watcher updates the tag index asynchronously. + Auto Memory only creates memory in the daily layer. To distill this material further into long-term `digest/` nodes, use [Auto Dream](./auto_dream.md). To search daily and digest content, use [Memory Search](./memory_search.md). diff --git a/docs/en/reme_scene.md b/docs/en/reme_scene.md index 6ee0b5ed..764caeea 100644 --- a/docs/en/reme_scene.md +++ b/docs/en/reme_scene.md @@ -81,7 +81,7 @@ Run: reme auto_dream date=2026-05-18 ``` -`auto_dream` is a three-step pipeline: +`auto_dream` is a four-step pipeline: ```text dream_extract_step @@ -92,6 +92,8 @@ dream_integrate_step decide CREATE / CORROBORATE / REFINE / CORRECT dream_finish_step checkpoint successfully processed daily inputs +auto_tag_step + tag the entities in created or modified digest notes ``` Outputs in this scenario: diff --git a/docs/zh/auto_dream.md b/docs/zh/auto_dream.md index d1bda782..2e132ddb 100644 --- a/docs/zh/auto_dream.md +++ b/docs/zh/auto_dream.md @@ -35,6 +35,7 @@ auto_dream: - backend: dream_integrate_step - backend: dream_finish_step file_catalog: dream + - backend: auto_tag_step ``` 参数含义: @@ -68,7 +69,7 @@ Auto Dream 只扫描 Markdown 日期索引和笔记,不读取 proactive 状态 | `digest/wiki/*.md` | 通用知识、概念、观察、决策先例。 | | `metadata/file_catalog/dream*` | dream 专用 catalog,用于判断 daily 输入是否变化。 | -## 三个阶段 +## 四个阶段 ### 1. Extract @@ -125,6 +126,15 @@ Auto Dream 不读取或写入 proactive 状态和 `interests.yaml`。这些文 失败路径不会被 checkpoint。这样下一次 `auto_dream` 仍会把它们视作 changed input,直到整合成功。 +### 4. Auto Tag + +Finish 后,`auto_dream` 和 `dream_cron` 都会通过 `auto_tag_step` 为本轮整合实际新增或修改的 Markdown digest 文件打标, +包括 Agent 异常后恢复的落盘结果。同一文件被多次写入时只打标一次。该 Step 复用 [Auto Memory](./auto_memory.md) 的请求级 +`changes` 协议,将实体标签写入配置的 frontmatter 字段,默认为 `memory_tags`。Dream 不为未变化文件或 daily 来源笔记打标。 + +打标诊断记录在 `metadata.auto_tag`。单文件打标失败保留 dream 原有的摘要、成功状态和 checkpoint 决策;后续没有文件变化的 +调用不会自动重试失败的打标。标签索引通过现有文件 watcher 异步更新。 + ## 运行方式 CLI: @@ -158,6 +168,7 @@ jobs: - backend: dream_integrate_step - backend: dream_finish_step file_catalog: dream + - backend: auto_tag_step ``` ## 关键边界 @@ -170,4 +181,4 @@ jobs: `auto_dream` 不凭空生成总览。只有 daily 输入中确实出现、并被抽取为 memory unit 的内容,才会进入 digest。 -完整流程依赖 LLM 完成 Extract 和 Integrate。 +完整流程依赖 LLM 完成 Extract、Integrate 和 Auto Tag。 diff --git a/docs/zh/auto_link.md b/docs/zh/auto_link.md index bc59af86..e9fa0d31 100644 --- a/docs/zh/auto_link.md +++ b/docs/zh/auto_link.md @@ -16,6 +16,7 @@ auto_dream: - dream_extract_step - dream_integrate_step # auto_link 的实际发生位置 - dream_finish_step + - auto_tag_step ``` Integrate 阶段对每个 unit 独立运行。一个 unit 只落到一个目标 digest 节点,但这个目标节点可以链接多个来源和多个相关 digest diff --git a/docs/zh/auto_memory.md b/docs/zh/auto_memory.md index 909c247f..8b969219 100644 --- a/docs/zh/auto_memory.md +++ b/docs/zh/auto_memory.md @@ -100,5 +100,11 @@ reme auto_memory \ ## 后续流向 +默认的 `auto_memory` 和 `auto_memory_cc` Job 会在记录记忆后执行 `auto_tag_step`,只为实际新增或修改的 daily 笔记打标, +并使用重命名后的最终路径。Claude Code 调用方仍只需传入 `session_id`;重复 Stop 没有新增消息时,记忆生成和打标都会跳过。 + +标签描述文档的核心实体,写入配置的 frontmatter 字段,默认为 `memory_tags`。单文件打标失败记录在 `metadata.auto_tag`, +保留原有记忆响应;没有笔记变化的调用不会自动重试失败的打标。标签索引通过现有文件 watcher 异步更新。 + Auto Memory 只生成 daily 层记忆。要把这些材料进一步沉淀为长期 `digest/` 节点,使用 [Auto Dream](./auto_dream.md);要搜索 daily 和 digest,使用 [Memory Search](./memory_search.md)。 diff --git a/docs/zh/reme_scene.md b/docs/zh/reme_scene.md index efb8bed5..cd51c158 100644 --- a/docs/zh/reme_scene.md +++ b/docs/zh/reme_scene.md @@ -75,7 +75,7 @@ daily/ reme auto_dream date=2026-05-18 ``` -`auto_dream` 是三步管线: +`auto_dream` 是四步管线: ```text dream_extract_step @@ -88,6 +88,9 @@ dream_integrate_step dream_finish_step checkpoint 成功处理的 daily 输入 + +auto_tag_step + 为新增或修改的 digest 笔记生成实体标签 ``` 本场景中的产物: diff --git a/reme/config/default.yaml b/reme/config/default.yaml index 3b7753a0..0624a709 100644 --- a/reme/config/default.yaml +++ b/reme/config/default.yaml @@ -72,6 +72,7 @@ jobs: - backend: dream_integrate_step - backend: dream_finish_step file_catalog: dream + - backend: auto_tag_step optimize_index_cron: backend: cron @@ -161,6 +162,7 @@ jobs: - backend: dream_integrate_step - backend: dream_finish_step file_catalog: dream + - backend: auto_tag_step auto_memory: backend: base diff --git a/reme/steps/evolve/dream/integrate.py b/reme/steps/evolve/dream/integrate.py index 763467f9..12569e7e 100644 --- a/reme/steps/evolve/dream/integrate.py +++ b/reme/steps/evolve/dream/integrate.py @@ -57,6 +57,7 @@ class DreamIntegrateStep(BaseStep): async def execute(self): assert self.context is not None + self.context["changes"] = [] state = state_from_context(self) if not state.units: self.logger.info(f"[{self.name}] skip no units") @@ -81,8 +82,16 @@ class DreamIntegrateStep(BaseStep): for bucket in DreamBucketEnum: (workspace / digest_dir / bucket.value).mkdir(parents=True, exist_ok=True) self.logger.info(f"[{self.name}] digest dirs ready buckets={len(list(DreamBucketEnum))}") + # Compare the whole run so retries and later units updating a newly + # created file still emit one "added" change for the final file. + before = _snapshot_digest(workspace, digest_dir) for i, unit in enumerate(state.units, start=1): await self._integrate_one(state, unit, i, workspace, digest_dir) + after = _snapshot_digest(workspace, digest_dir) + self.context["changes"] = [ + {"change": "modified" if path in before else "added", "path": path} + for path in _changed_digest_paths(before, after) + ] state.failed_paths = sorted(set(state.failed_paths)) answer = ( f"Integrated {len(state.integrate_results)} unit(s); skipped {len(state.skipped_units)} unit(s); " diff --git a/tests/unit/test_auto_dream.py b/tests/unit/test_auto_dream.py index e87b82b8..b31ea9e0 100644 --- a/tests/unit/test_auto_dream.py +++ b/tests/unit/test_auto_dream.py @@ -3,22 +3,30 @@ # pylint: disable=protected-access import asyncio +import json import tempfile from pathlib import Path -from unittest.mock import patch +from unittest.mock import AsyncMock, patch +import frontmatter +import pytest import yaml from reme.components.application_context import ApplicationContext from reme.components.agent_wrapper import BaseAgentWrapper from reme.components.file_catalog import BaseFileCatalog from reme.components.file_store import BaseFileStore +from reme.components.job import BaseJob from reme.components.runtime_context import RuntimeContext +from reme.components.tag_index import LocalTagIndex +from reme.config import resolve_app_config from reme.schema import DreamState, FileNode +from reme.steps.evolve.auto_tag import AutoTagStep from reme.steps.evolve.dream.extract import DreamExtractStep from reme.steps.evolve.dream.finish import DreamFinishStep from reme.steps.evolve.dream.integrate import DreamIntegrateStep, _snapshot_digest from reme.steps.evolve.dream.utils import parse_structured_reply, recent_dates, scan_day_files +from reme.steps.file_io.frontmatter_update import FrontmatterUpdateStep def _touch(path: Path, text: str = "x") -> Path: @@ -110,6 +118,8 @@ class _SequenceAgent(BaseAgentWrapper): async def reply(self, _message, **_kwargs): self.calls += 1 outcome = self.outcomes.pop(0) + if callable(outcome): + outcome = outcome() if isinstance(outcome, Exception): raise outcome return outcome @@ -497,6 +507,175 @@ def test_integrate_uses_one_application_wide_lock(tmp_path): assert first._integration_lock() is second._integration_lock() # pylint: disable=protected-access +@pytest.mark.asyncio +async def test_integrate_emits_actual_changes_once_across_units(tmp_path, monkeypatch): + """Creation followed by updates stays added; receipt-only updates emit nothing.""" + created = "digest/wiki/created.md" + updated = "digest/procedure/updated.md" + untouched = "digest/personal/untouched.md" + _touch(tmp_path / updated, "old") + _touch(tmp_path / untouched, "unchanged") + + def create(): + _touch(tmp_path / created, "new") + return {"result": json.dumps({"action": "CREATE", "target_path": created})} + + def update(): + _touch(tmp_path / created, "new with more evidence") + _touch(tmp_path / updated, "updated existing memory") + return {"result": json.dumps({"action": "REFINE", "target_path": created})} + + agent = _SequenceAgent( + create, + update, + {"result": json.dumps({"action": "CORROBORATE", "target_path": untouched})}, + ) + state = DreamState( + units=[{"name": str(i), "bucket": "wiki", "paths": ["daily/source.md"]} for i in range(3)], + ) + context = RuntimeContext(dream=state.model_dump(), file_store=_FileStore(tmp_path)) + step = DreamIntegrateStep(app_context=ApplicationContext(workspace_dir=str(tmp_path)), agent_wrapper=agent) + monkeypatch.setattr("reme.steps.evolve.dream.integrate.llm_available", lambda _step: True) + + await step(context) + + assert context.response.success is True + assert context["changes"] == [ + {"change": "modified", "path": updated}, + {"change": "added", "path": created}, + ] + assert untouched in context.response.metadata["dream"]["nodes_updated"] + assert untouched not in context.response.metadata["dream"]["modified_paths"] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("agent_error", [False, True]) +async def test_integrate_keeps_added_changes_across_retry_recovery(tmp_path, monkeypatch, agent_error): + """Files from both attempts remain eligible for tagging after receipt recovery.""" + first, second = "digest/wiki/first.md", "digest/wiki/second.md" + + def attempt_one(): + _touch(tmp_path / first, "first") + _touch(tmp_path / second, "second") + return RuntimeError("agent failed") if agent_error else {"result": "{}"} + + def attempt_two(): + _touch(tmp_path / first, "first with more evidence") + return RuntimeError("agent failed") if agent_error else {"result": "{}"} + + agent = _SequenceAgent(attempt_one, attempt_two) + state = DreamState(units=[{"name": "unit", "bucket": "wiki", "paths": ["daily/source.md"]}]) + context = RuntimeContext(dream=state.model_dump(), file_store=_FileStore(tmp_path)) + step = DreamIntegrateStep(app_context=ApplicationContext(workspace_dir=str(tmp_path)), agent_wrapper=agent) + monkeypatch.setattr("reme.steps.evolve.dream.integrate.llm_available", lambda _step: True) + + await step(context) + + assert agent.calls == 2 + assert context.response.success is True + assert context.response.metadata["dream"]["warnings"] + assert context["changes"] == [{"change": "added", "path": path} for path in (first, second)] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("with_units", [False, True]) +async def test_integrate_clears_changes_when_skipping_or_missing_llm(tmp_path, monkeypatch, with_units): + """Early returns cannot pass caller-supplied paths into AutoTag.""" + app_context = ApplicationContext(workspace_dir=str(tmp_path)) + state = DreamState(units=[{"name": "unit", "paths": ["daily/source.md"]}] if with_units else []) + agent = _ReplyAgent() + context = RuntimeContext(dream=state.model_dump(), changes=[{"change": "added", "path": "daily/source.md"}]) + monkeypatch.setattr("reme.steps.evolve.dream.integrate.llm_available", lambda _step: False) + + await DreamIntegrateStep(app_context=app_context, agent_wrapper=agent)(context) + response = await AutoTagStep(app_context=app_context, agent_wrapper=agent)(context) + + assert context["changes"] == [] + assert response.success is not with_units + assert response.metadata["auto_tag"]["processed"] == 0 + assert agent.calls == 0 + + +@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]) +async def test_dream_jobs_tag_outputs_and_preserve_checkpoint_results( + tmp_path, + monkeypatch, + job_name, + tag_fails, + integrate_fails, +): + """Both configured pipelines tag durable outputs without changing dream success.""" + app_context = ApplicationContext(workspace_dir=str(tmp_path)) + catalog, store = _Catalog(), _FileStore(tmp_path) + store.tag_index = LocalTagIndex() + source = "daily/2026-09-11/source.md" + targets = ["digest/wiki/first.md", "digest/wiki/second.md"] + _touch(tmp_path / source, "Source about ReMe") + unit = {"name": "unit", "bucket": "wiki", "summary": "ReMe memory", "paths": [source]} + + async def reply(_message, **kwargs): + if kwargs["job_tools"] == ["read"]: + return {"result": json.dumps({"units": [unit]})} + if "list_tags" in kwargs["job_tools"]: + injected = kwargs["injected_job_kwargs"] + path = injected["_allowed_paths"][0] + if tag_fails and path == targets[0]: + raise RuntimeError("tagging unavailable") + update_context = RuntimeContext( + path=path, + metadata={"memory_tags": ["ReMe"]}, + **injected, + ) + await FrontmatterUpdateStep(file_store=store)(update_context) + assert update_context.response.success + return {"result": "Tagged ReMe"} + 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: + raise RuntimeError("integration unavailable") + return {"result": json.dumps({"action": "CREATE", "target_path": targets[0]})} + + agent = _ReplyAgent() + monkeypatch.setattr(agent, "reply", AsyncMock(side_effect=reply)) + monkeypatch.setattr("reme.steps.evolve.dream.extract.llm_available", lambda _step: True) + monkeypatch.setattr("reme.steps.evolve.dream.integrate.llm_available", lambda _step: True) + config = resolve_app_config(config="default", log_config=False)["jobs"][job_name] + # Execute the cron's configured steps once without starting a scheduler. + job = BaseJob(name=job_name, steps=config["steps"], app_context=app_context) + await job.start() + try: + response = await job( + date="2026-09-11", + scan_days=1, + agent_wrapper=agent, + file_store=store, + file_catalog=catalog, + ) + finally: + await job.close() + + assert response.success is not integrate_fails + assert response.answer.startswith("AutoDream completed") + assert response.metadata["modified"] is True + dream, tagging = response.metadata["dream"], response.metadata["auto_tag"] + assert (source in dream["checkpoint_paths"]) is not integrate_fails + assert (source in dream["failed_paths"]) is integrate_fails + assert tagging["processed"] == 2 + assert tagging["failed"] == int(tag_fails) + assert tagging["succeeded"] == 2 - int(tag_fails) + assert [{"change": item["change"], "path": item["path"]} for item in tagging["results"]] == [ + {"change": "added", "path": path} for path in targets + ] + for path in targets: + post = frontmatter.loads((tmp_path / path).read_text(encoding="utf-8")) + assert post.metadata.get("memory_tags") == (None if tag_fails and path == targets[0] else ["ReMe"]) + assert (tmp_path / source).read_text(encoding="utf-8") == "Source about ReMe" + + def test_extract_without_llm_marks_changed_paths_failed(tmp_path): """A missing LLM must not let finish checkpoint unprocessed source files.""" diff --git a/tests/unit/test_auto_memory_cc.py b/tests/unit/test_auto_memory_cc.py index 46deeece..9ae05c1c 100644 --- a/tests/unit/test_auto_memory_cc.py +++ b/tests/unit/test_auto_memory_cc.py @@ -3,9 +3,17 @@ # pylint: disable=protected-access from types import SimpleNamespace +from unittest.mock import AsyncMock +import frontmatter import pytest +from reme.components.agent_wrapper import BaseAgentWrapper +from reme.components.application_context import ApplicationContext +from reme.components.file_store import LocalFileStore +from reme.components.job import BaseJob +from reme.config import resolve_app_config +from reme.enumeration import ComponentEnum from reme.steps.evolve.auto_memory_cc import AutoMemoryCCStep @@ -27,3 +35,108 @@ async def test_reme_cc_store_preserves_existing_session_layout(tmp_path): assert increment == [{"uuid": "new"}] assert step._session_link(session_id) == f"[[session/claude_code/{session_id}.jsonl]]" assert not (tmp_path / "session" / "claude_code" / "claude_code").exists() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("tag_fails", [False, True]) +async def test_cc_job_tags_created_updated_and_renamed_notes_only(tmp_path, monkeypatch, tag_fails): + """The configured CC job inherits changes, skips no-ops, and preserves its main result.""" + monkeypatch.chdir(tmp_path) + app_context = ApplicationContext(workspace_dir=str(tmp_path)) + store = LocalFileStore(name="default", embedding_store="", tag_index="default") + app_context.components[ComponentEnum.FILE_STORE] = {"default": store} + configs = resolve_app_config(config="default", log_config=False)["jobs"] + session_id, day = "session-1", "2026-09-11" + source = f"[[session/claude_code/{session_id}.jsonl]]" + entries = [] + monkeypatch.setattr(AutoMemoryCCStep, "_load_cc_session", AsyncMock(return_value=entries)) + mode = "create" + tag_paths = [] + memory_calls = [] + + async def reply(_message, **kwargs): + if "list_tags" in kwargs["job_tools"]: + injected = kwargs["injected_job_kwargs"] + path = injected["_allowed_paths"][0] + tag_paths.append(path) + if tag_fails: + raise RuntimeError("tagging unavailable") + result = await app_context.jobs["frontmatter_update"]( + path=path, + metadata={"memory_tags": ["ReMe"]}, + **injected, + ) + assert result.success + return {"result": "Tagged ReMe"} + memory_calls.append(mode) + if mode == "create": + target = tmp_path / f"daily/{day}/note.md" + target.parent.mkdir(parents=True) + post = frontmatter.Post("ReMe memory", name="note", session_id=session_id, source_conversation=source) + elif mode == "unchanged": + return {"result": "No new facts"} + else: + target = tmp_path / kwargs["injected_job_kwargs"]["_allowed_paths"][0] + post = frontmatter.loads(target.read_text(encoding="utf-8")) + post.content += "\nMore ReMe evidence" + if mode == "rename": + post.metadata["name"] = "renamed" + target.write_text(frontmatter.dumps(post), encoding="utf-8") + return {"result": "Recorded memory"} + + agent = AsyncMock(spec=BaseAgentWrapper) + agent.reply.side_effect = reply + await store.start() + try: + for name in ("daily_list", "move", "frontmatter_update", "auto_memory_cc"): + job = BaseJob( + name=name, + steps=configs[name]["steps"], + app_context=app_context, + agent_wrapper=agent, + date=day, + ) + app_context.jobs[name] = job + await job.start() + job = app_context.jobs["auto_memory_cc"] + for index, mode in enumerate(("create", "unchanged", "update", "rename")): + entries.append( + {"uuid": str(index), "type": "user", "message": {"role": "user", "content": f"ReMe fact {index}"}}, + ) + before_tags = len(tag_paths) + response = await job(session_id=session_id) + assert response.success is True + assert response.metadata["source_conversation"] == source + if mode == "unchanged": + assert response.answer == "No new facts" + assert response.metadata["modified"] is False + assert response.metadata["auto_tag"]["processed"] == 0 + assert len(tag_paths) == before_tags + else: + path = f"daily/{day}/{'renamed' if mode == 'rename' else 'note'}.md" + assert response.answer == "Recorded memory" + assert response.metadata["path"] == path + assert response.metadata["modified"] is True + assert response.metadata["created"] is (mode == "create") + tagging = response.metadata["auto_tag"] + assert tagging["processed"] == 1 + assert tagging["failed"] == int(tag_fails) + assert tagging["results"][0]["change"] == ("added" if mode == "create" else "modified") + assert tag_paths[-1] == path + post = frontmatter.loads((tmp_path / path).read_text(encoding="utf-8")) + assert post.metadata.get("memory_tags") == (None if tag_fails else ["ReMe"]) + assert post.metadata["source_conversation"] == source + + # Repeating a Stop with the same UUIDs must run neither agent again. + before_calls = (len(memory_calls), len(tag_paths)) + repeated = await job(session_id=session_id) + assert repeated.success is True + assert repeated.answer == "Skipped: no messages" + assert repeated.metadata["auto_tag"]["processed"] == 0 + assert (len(memory_calls), len(tag_paths)) == before_calls + assert not (tmp_path / f"daily/{day}/note.md").exists() + assert not (tmp_path / "session/dialog").exists() + finally: + for job in reversed(list(app_context.jobs.values())): + await job.close() + await store.close()