From 3678dc6823d3003b3703b799f0b6715b3eab6e66 Mon Sep 17 00:00:00 2001 From: wang-qisen Date: Mon, 31 Aug 2026 18:13:16 +0800 Subject: [PATCH] refactor: split auto resource processors behind router --- README.md | 2 +- README_ZH.md | 2 +- docs/en/auto_resource.md | 9 + docs/en/reme-blog.md | 2 +- docs/en/reme_scene.md | 6 +- docs/zh/auto_resource.md | 7 + docs/zh/reme-blog.md | 2 +- docs/zh/reme_scene.md | 4 +- pyproject.toml | 8 +- reme/config/default.yaml | 15 +- reme/steps/evolve/__init__.py | 6 +- reme/steps/evolve/_auto_resource.py | 404 +++++++++++ reme/steps/evolve/auto_image.py | 150 +++-- reme/steps/evolve/auto_image.yaml | 17 +- reme/steps/evolve/auto_resource.py | 634 +++--------------- reme/steps/evolve/auto_text.py | 197 ++++++ .../{auto_resource.yaml => auto_text.yaml} | 1 + tests/integration/test_auto_resource.py | 3 +- tests/unit/test_auto_image_steps.py | 629 +++++++++++++++-- tests/unit/test_background_steps.py | 32 +- 20 files changed, 1479 insertions(+), 651 deletions(-) create mode 100644 reme/steps/evolve/_auto_resource.py create mode 100644 reme/steps/evolve/auto_text.py rename reme/steps/evolve/{auto_resource.yaml => auto_text.yaml} (99%) diff --git a/README.md b/README.md index dc0c18ce..23593b9f 100644 --- a/README.md +++ b/README.md @@ -340,7 +340,7 @@ These guides cover the main user workflows and the runtime contracts implemented | [Quick Start](docs/en/quick_start.md) | Install ReMe, start the service, and run the first file and memory operations. | | [Memory as File](docs/en/memory_as_file.md) | Understand workspace layers, frontmatter, wikilinks, chunks, and the file-as-source-of-truth model. | | [Auto Memory](docs/en/auto_memory.md) | Preserve source conversations and distill reusable daily memory cards. | -| [Auto Resource](docs/en/auto_resource.md) | Import supported text resources and turn them into source-linked daily cards. | +| [Auto Resource](docs/en/auto_resource.md) | Import supported text and image resources as source-linked daily cards. | | [Auto Dream](docs/en/auto_dream.md) and [Auto Link](docs/en/auto_link.md) | Consolidate daily notes into evolving digest nodes and readable wikilink relationships. | | [Memory Search](docs/en/memory_search.md) | Use BM25, optional vectors, RRF fusion, line-range recall, and progressive link expansion. | | [Proactive](docs/en/proactive.md) | Read interest topics safely and integrate them into a host agent's decision flow. | diff --git a/README_ZH.md b/README_ZH.md index 7583e4f7..bd7b6519 100644 --- a/README_ZH.md +++ b/README_ZH.md @@ -330,7 +330,7 @@ ReMe 通过 Agent 多轮搜索与读取的方式，评测多会话和超长上 | [快速开始](docs/zh/quick_start.md) | 安装 ReMe、启动服务，并执行首次文件和记忆操作。 | | [Memory as File](docs/zh/memory_as_file.md) | 理解 workspace 分层、frontmatter、wikilink、chunk 和文件事实来源模型。 | | [Auto Memory](docs/zh/auto_memory.md) | 保留过滤后的对话来源记录，并提炼可复用的 daily 记忆卡片。 | -| [Auto Resource](docs/zh/auto_resource.md) | 导入支持的文本资料，转换为可追溯来源的 daily 卡片。 | +| [Auto Resource](docs/zh/auto_resource.md) | 导入支持的文本与图像资料，转换为可追溯来源的 daily 卡片。 | | [Auto Dream](docs/zh/auto_dream.md) 与 [Auto Link](docs/zh/auto_link.md) | 将 daily 记忆整理为持续演化的 digest 节点和可读 wikilink 关系。 | | [记忆检索](docs/zh/memory_search.md) | 使用 BM25、可选向量、RRF 融合、行号范围召回和渐进式链接扩展。 | | [Proactive](docs/zh/proactive.md) | 安全读取兴趣主题，并将其接入宿主 Agent 的决策流程。 | diff --git a/docs/en/auto_resource.md b/docs/en/auto_resource.md index 716689db..503b3590 100644 --- a/docs/en/auto_resource.md +++ b/docs/en/auto_resource.md @@ -53,6 +53,11 @@ Text resources such as `md`, `txt`, `json`, `jsonl`, `csv`, `yaml`, and `html` a (`png`, `jpg`, `jpeg`, `webp`, `gif`, `bmp`, `tiff`, `heic`) produce caption cards as described in [Image Resources](#image-resources). +Internally, one `AutoResourceStep` receives each change batch and sends every item to the first configured processor +whose class-level matcher accepts it. `AutoImageResourceStep` handles image suffixes and `AutoTextResourceStep` is the +final fallback. A new modality can therefore add a registered processor, its prompt, and one `dispatch_steps` entry +without changing the router. + ## Image Resources Image files are interpreted the same way: a vision model writes a caption card that links back to the original image. @@ -65,6 +70,10 @@ provider-unfriendly formats are downscaled or re-encoded in memory for the reque `resource/` is never modified. When an image changes, its card is rewritten in place; when the image is deleted, the card is removed with it. +Image preprocessing uses Pillow from the `core` extra. HEIC resources additionally require the optional +`image-heif` extra: `pip install "reme-ai[image-heif]"`. Other supported image formats do not load or require the HEIF +plugin. + ## Resource Cards Each resource file produces one daily resource card. The system initially uses the resource file's stem as a temporary diff --git a/docs/en/reme-blog.md b/docs/en/reme-blog.md index bec49207..34732191 100644 --- a/docs/en/reme-blog.md +++ b/docs/en/reme-blog.md @@ -113,7 +113,7 @@ It is like having a recorder who is always present—not one that mechanically t Not all valuable information comes from conversations. Research materials, project documents, meeting notes, archived web pages, and structured data may all become part of a personal knowledge base. -Auto Resource provides a general entry point for external materials. After a resource enters `resource/`, ReMe preserves the original and organizes its topics, key facts, and actionable information into daily cards with `source_resource` links. It currently supports text-based resources including Markdown, plain text, JSON, JSONL, CSV, YAML, and HTML. +Auto Resource provides a general entry point for external materials. After a resource enters `resource/`, ReMe preserves the original and organizes its topics, key facts, and actionable information into daily cards with `source_resource` links. It supports text resources including Markdown, plain text, JSON, JSONL, CSV, YAML, and HTML, plus image resources that a vision model turns into caption cards. In other words, Auto Memory builds personal knowledge from conversations, while Auto Resource builds it from non-conversational materials. Both streams flow into the same daily memory layer, where ReMe indexes, consolidates, and retrieves them together. diff --git a/docs/en/reme_scene.md b/docs/en/reme_scene.md index c0dcf2a2..d95bf2c2 100644 --- a/docs/en/reme_scene.md +++ b/docs/en/reme_scene.md @@ -66,9 +66,9 @@ The corresponding flow is: - `auto_memory` saves a filtered source conversation record to `session/dialog/.jsonl`, then asks the agent to write important facts to a topic-named `daily//.md`. The note keeps `session_id` and `source_conversation` in frontmatter for stable lookup and provenance. -- `resource_watch_loop` watches text-file changes under `resource/` and triggers `auto_resource_step` to write a daily note - with `source_resource`. The agent suggests a content-based filename, which the system sanitizes and de-duplicates; it is - not guaranteed to match the resource filename. +- `resource_watch_loop` watches supported text and image changes under `resource/` and triggers `auto_resource_step` to + write a daily note with `source_resource`. Text resources use an agent, while images use a vision model. The generated + content-based filename is sanitized and de-duplicated; it is not guaranteed to match the resource filename. - Auto Memory, Auto Resource, and Auto Dream refresh `daily/.md` after writing. ### Day 1 evening: Auto Dream writes to Digest diff --git a/docs/zh/auto_resource.md b/docs/zh/auto_resource.md index 537a2e9f..00201a0b 100644 --- a/docs/zh/auto_resource.md +++ b/docs/zh/auto_resource.md @@ -48,12 +48,19 @@ workspace/ 当前 Beta 版本以文本类资源为主，例如 `md`、`txt`、`json`、`jsonl`、`csv`、`yaml`、`html`；图像资源（`png`、`jpg`、`jpeg`、`webp`、`gif`、`bmp`、`tiff`、`heic`）会生成 caption 卡片，见下文[图像资源](#图像资源)一节。 +内部由统一的 `AutoResourceStep` 接收每批变更，并将每一项交给配置中第一个匹配它的 processor。 +`AutoImageResourceStep` 声明图像后缀匹配规则，`AutoTextResourceStep` 作为最后的 fallback。后续新增模态时， +只需注册新的 processor、提供独立 prompt 并在 `dispatch_steps` 中增加一项，无需修改 router。 + ## 图像资源 图像文件的解读方式相同：视觉模型写入一张 caption 卡片并链接原图。卡片正文以 `![[resource/...]]` 嵌入链接开头，frontmatter 携带 `kind: image` 与 `media_type`，文本检索因此可以通过 caption 命中图像内容。 视觉模型优先使用配置中的 `as_llm` `vision` 实例，未配置时回退到 `default` 实例——默认模型具备视觉能力时无需额外配置。超过请求预算或格式不被模型接受的图像，仅在请求前于内存中降采样或转码；`resource/` 下的原图文件不会被修改。图像变更时卡片原地重写；图像删除时卡片随之删除。 +图像预处理使用 `core` extra 中的 Pillow。HEIC 资源还需要可选的 `image-heif` extra： +`pip install "reme-ai[image-heif]"`。其他受支持图像格式不会加载或依赖 HEIF 插件。 + ## 资源卡片 每个资源文件会生成一张 daily 资源卡片。创建时先使用资源文件 stem 作为临时路径，Agent 写入后，系统会根据 frontmatter `name` diff --git a/docs/zh/reme-blog.md b/docs/zh/reme-blog.md index a616b939..1af0d05f 100644 --- a/docs/zh/reme-blog.md +++ b/docs/zh/reme-blog.md @@ -119,7 +119,7 @@ daily/2026-08-07.md 当天索引，负责总览 并不是所有有价值的信息都来自对话。研究资料、项目文档、会议纪要、网页存档和结构化数据，同样可能成为个人知识库的一部分。 Auto Resource 提供了一条更通用的外部资料入口。资料进入 `resource/` 后，ReMe 保留原文，再把主题、关键事实和可行动信息整理为带有 -`source_resource` 链接的 daily 卡片。当前可以处理 Markdown、纯文本、JSON、JSONL、CSV、YAML 和 HTML 等文本类资料。 +`source_resource` 链接的 daily 卡片。它支持 Markdown、纯文本、JSON、JSONL、CSV、YAML 和 HTML 等文本类资料，也支持由视觉模型生成 caption 卡片的图像资源。 这意味着，Auto Memory 负责从对话建立个人知识，Auto Resource 负责从非对话资料建立个人知识。两条输入最终进入同一个 daily 记忆层，再由 ReMe 统一索引、整合和检索。 diff --git a/docs/zh/reme_scene.md b/docs/zh/reme_scene.md index 0a681de8..25db4bec 100644 --- a/docs/zh/reme_scene.md +++ b/docs/zh/reme_scene.md @@ -61,8 +61,8 @@ daily/ - `auto_memory` 保存对话来源消息到 `session/dialog/.jsonl`，再让 Agent 把重要事实写入按主题命名的 `daily//.md`；卡片 frontmatter 保留 `session_id` 和 `source_conversation` 用于稳定定位和追溯。 -- `resource_watch_loop` 监听 `resource/` 文本文件变化，并触发 `auto_resource_step` 写带 `source_resource` 的 daily note；文件名由 - Agent 根据内容建议，再由系统清洗并处理冲突，不保证与资源同名。 +- `resource_watch_loop` 监听 `resource/` 下支持的文本与图像变化，并触发 `auto_resource_step` 写带 + `source_resource` 的 daily note；文本资源由 Agent 处理，图像由视觉模型处理。根据内容生成的文件名会被系统清洗并处理冲突，不保证与资源同名。 - Auto Memory、Auto Resource 和 Auto Dream 都会在写入后刷新 `daily/.md` 当天索引页。 ### Day 1 晚上：Auto Dream 进入 Digest diff --git a/pyproject.toml b/pyproject.toml index 217031bb..234e5f53 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -30,8 +30,6 @@ dependencies = [ "mistletoe>=1.5.1", "numpy>=2.2.6", "openai>=2.26.0", - "pillow>=10.0.0", - "pillow-heif>=0.13.0", "psutil>=5.9", "pydantic>=2.12.5", "python-frontmatter>=1.1.0", @@ -51,6 +49,7 @@ web = [ ] core = [ "reme-ai[as]", + "pillow>=10.0.0", "claude-agent-sdk>=0.2.126", "dingtalk-stream>=0.24.3", "openai-codex>=0.144.4", @@ -64,6 +63,10 @@ core = [ "polars>=1.43.0", "reme_studio", ] +image-heif = [ + "pillow>=10.0.0", + "pillow-heif>=0.13.0", +] dev = [ "packaging>=24.2", "pre-commit>=4.6.1", @@ -73,6 +76,7 @@ dev = [ full = [ "reme-ai[core]", "reme-ai[dev]", + "reme-ai[image-heif]", ] [project.urls] diff --git a/reme/config/default.yaml b/reme/config/default.yaml index ff9edccc..0f741c50 100644 --- a/reme/config/default.yaml +++ b/reme/config/default.yaml @@ -31,13 +31,17 @@ jobs: - backend: update_catalog_step file_catalog: resource - backend: auto_resource_step - - backend: auto_image_step + dispatch_steps: + - auto_image_resource_step + - auto_text_resource_step - backend: watch_changes_step dispatch_steps: - backend: update_catalog_step file_catalog: resource - backend: auto_resource_step - - backend: auto_image_step + dispatch_steps: + - auto_image_resource_step + - auto_text_resource_step digest_watch_loop: backend: background @@ -189,6 +193,9 @@ jobs: - changes steps: - backend: auto_resource_step + dispatch_steps: + - auto_image_resource_step + - auto_text_resource_step auto_image: backend: base @@ -212,7 +219,7 @@ jobs: required: - changes steps: - - backend: auto_image_step + - backend: auto_image_resource_step proactive: backend: base @@ -781,7 +788,7 @@ components: max_tokens: 65536 thinking_enable: false - # Optional dedicated vision model for image resources (auto_image_step). + # Optional dedicated vision model for image resources (auto_image_resource_step). # Falls back to the "default" instance above when absent; uncomment to # decouple the vision model from the main LLM. # vision: diff --git a/reme/steps/evolve/__init__.py b/reme/steps/evolve/__init__.py index fec242a4..0b57349e 100644 --- a/reme/steps/evolve/__init__.py +++ b/reme/steps/evolve/__init__.py @@ -1,19 +1,21 @@ """Evolve steps.""" from ._evolve import now -from .auto_image import AutoImageStep +from .auto_image import AutoImageResourceStep from .auto_memory import AutoMemoryStep from .auto_memory_cc import AutoMemoryCCStep from .auto_resource import AutoResourceStep +from .auto_text import AutoTextResourceStep from .compressor import CompressorStep from .dream import DreamExtractStep, DreamFinishStep, DreamIntegrateStep, DreamTopicsStep, ProactiveStep __all__ = [ "now", - "AutoImageStep", + "AutoImageResourceStep", "AutoMemoryStep", "AutoMemoryCCStep", "AutoResourceStep", + "AutoTextResourceStep", "CompressorStep", "DreamExtractStep", "DreamFinishStep", diff --git a/reme/steps/evolve/_auto_resource.py b/reme/steps/evolve/_auto_resource.py new file mode 100644 index 00000000..9c90168c --- /dev/null +++ b/reme/steps/evolve/_auto_resource.py @@ -0,0 +1,404 @@ +"""Shared lifecycle and helpers for automatic resource processors.""" + +import hashlib +import re +from abc import abstractmethod +from collections.abc import Mapping +from pathlib import Path, PurePosixPath +from typing import Any + +import frontmatter +from watchfiles import Change + +from ..base_step import BaseStep +from ..file_io import refresh_day_index, validate_filename_component +from ._evolve import now + +_SOURCE_RESOURCE_KEY = "source_resource" +_DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$") +_UNSAFE_FILENAME_CHARS = re.compile(r'[<>:"/\\|?*\x00-\x1f]+') + + +def _compute_note_stem(filename: str) -> str: + """Return the daily note stem for a resource filename.""" + return PurePosixPath(filename).stem + + +def _parse_resource_path(file_path: str, resource_dir: str) -> tuple[str, str]: + """Extract (date, filename) from a resource path like 'resource/2026-06-06/report.pdf'. + + Returns (date_str, filename) where filename may contain subdirectories. + """ + parts = PurePosixPath(file_path).parts + # Strip leading resource_dir prefix + prefix_parts = PurePosixPath(resource_dir).parts + if parts[: len(prefix_parts)] != prefix_parts: + return "", "" + parts = parts[len(prefix_parts) :] + # First segment is date, rest is filename + date_str = parts[0] if parts else "" + if not _DATE_RE.match(date_str): + return "", "" + filename = str(PurePosixPath(*parts[1:])) if len(parts) > 1 else "" + return date_str, filename + + +def _loose_resource_filename(file_path: str, resource_dir: str) -> str: + """Return filename for a root-level resource path like 'resource/report.txt'.""" + parts = PurePosixPath(file_path).parts + prefix_parts = PurePosixPath(resource_dir).parts + if parts[: len(prefix_parts)] != prefix_parts: + return "" + rest = parts[len(prefix_parts) :] + if len(rest) != 1: + return "" + filename = rest[0] + return "" if filename in ("", ".", "..") else filename + + +def _results_answer(results: list[dict], processed_answer: str) -> str: + """Return the actual per-change answer while preserving a batch fallback.""" + answers = [str(item.get("answer") or "").strip() for item in results] + answers = [item for item in answers if item] + if len(answers) == 1: + return answers[0] + if len(answers) > 1: + return "\n\n".join(f"{index}. {answer}" for index, answer in enumerate(answers, start=1)) + return processed_answer + + +def _source_suffix(file_path: str) -> str: + """Return a short stable suffix for source-path collision handling.""" + return hashlib.sha1(file_path.encode("utf-8")).hexdigest()[:8] + + +def _sanitize_note_name(raw: str, fallback: str) -> str: + """Return a safe single filename component from an LLM-suggested name.""" + name = str(raw or "").strip() + name = _UNSAFE_FILENAME_CHARS.sub("-", name) + name = re.sub(r"\s+", " ", name).strip(" .") + if not name: + name = str(fallback or "").strip() + name = _UNSAFE_FILENAME_CHARS.sub("-", name).strip(" .") + if not name or validate_filename_component(name, kind="name"): + name = f"resource-{_source_suffix(fallback or raw or 'note')}" + if validate_filename_component(name, kind="name"): + name = f"resource-{_source_suffix(name)}" + return name + + +class BaseAutoResourceStep(BaseStep): + """Shared source-linked daily-note lifecycle for resource processors.""" + + resource_fallback = False + resource_suffixes: frozenset[str] = frozenset() + router_inherit_keys = frozenset({"file_store", "language"}) + + @classmethod + def matches_change(cls, change: Mapping[str, Any]) -> bool: + """Return whether this processor accepts a change before fallback. + + The predicate must stay synchronous and file-system independent because + deleted resources no longer exist when the router evaluates them. + """ + file_path = change.get("path") or change.get("file_path", "") + return Path(str(file_path)).suffix.lower() in cls.resource_suffixes + + def _normalize_change(self, raw) -> Change | None: + if isinstance(raw, Change): + return raw + if isinstance(raw, str): + return Change.__members__.get(raw) + return None + + def _today(self) -> str: + tz = self.app_context.app_config.timezone if self.app_context is not None else None + return now(tz).strftime("%Y-%m-%d") + + def _daily_note_path(self, day: str, name: str) -> str: + return f"{self.config_value('daily_dir')}/{day}/{name}.md" + + @staticmethod + def _source_resource_link(file_path: str) -> str: + return f"[[{file_path}]]" + + def _frontmatter(self, path: str) -> dict: + post = frontmatter.loads((self.file_store.workspace_path / path).read_text(encoding="utf-8")) + return dict(post.metadata or {}) + + def _note_bytes(self, path: str) -> bytes | None: + note_path = self.file_store.workspace_path / path + if not note_path.is_file(): + return None + return note_path.read_bytes() + + def _note_modified(self, before_path: str, before_bytes: bytes | None, after_path: str) -> bool: + if not after_path: + return False + after_bytes = self._note_bytes(after_path) + if after_bytes is None: + return before_bytes is not None + return after_path != before_path or before_bytes != after_bytes + + def _find_resource_note(self, notes: list[dict], file_path: str, fallback_path: str) -> dict | None: + source = self._source_resource_link(file_path) + for note in notes: + if str(note.get(_SOURCE_RESOURCE_KEY, "")).strip() == source: + return note + for note in notes: + if str(note.get("path", "")).strip() == fallback_path: + return note + return None + + async def _list_resource_note(self, day: str, file_path: str, fallback_path: str) -> dict | None: + list_response = await self.run_job("daily_list", date=day) + if not list_response.success: + raise RuntimeError(f"daily_list failed: {list_response.answer}") + notes = list_response.metadata.get("notes") or [] + return self._find_resource_note(notes, file_path, fallback_path) + + async def _ensure_resource_frontmatter(self, path: str, file_path: str) -> None: + metadata = {_SOURCE_RESOURCE_KEY: self._source_resource_link(file_path)} + current = self._frontmatter(path) + if all(current.get(key) == value for key, value in metadata.items()): + return + response = await self.run_job( + "frontmatter_update", + path=path, + metadata=metadata, + ) + if not response.success: + raise RuntimeError(f"frontmatter_update failed: {response.answer}") + + async def _set_frontmatter_name(self, path: str, name: str) -> None: + if self._frontmatter(path).get("name") == name: + return + response = await self.run_job("frontmatter_update", path=path, metadata={"name": name}) + if not response.success: + raise RuntimeError(f"frontmatter_update failed: {response.answer}") + + def _unique_daily_note_path(self, day: str, name: str, file_path: str, current_path: str) -> tuple[str, str]: + """Return a collision-free (name, path), preserving current_path when possible.""" + target_path = self._daily_note_path(day, name) + target_abs = self.file_store.workspace_path / target_path + if target_path == current_path or not target_abs.exists(): + return name, target_path + + suffixed = f"{name}--{_source_suffix(file_path)}" + target_path = self._daily_note_path(day, suffixed) + target_abs = self.file_store.workspace_path / target_path + if target_path == current_path or not target_abs.exists(): + return suffixed, target_path + + for index in range(2, 100): + candidate = f"{suffixed}-{index}" + target_path = self._daily_note_path(day, candidate) + target_abs = self.file_store.workspace_path / target_path + if target_path == current_path or not target_abs.exists(): + return candidate, target_path + raise RuntimeError(f"cannot allocate unique note name for: {name!r}") + + async def _rename_from_frontmatter_name( + self, + path: str, + day: str, + file_path: str, + fallback_name: str, + fallback_path: str, + *, + allow_rename: bool, + ) -> str: + meta = self._frontmatter(path) + current_name = PurePosixPath(path).stem + suggested_name = str(meta.get("name", "")).strip() + + if not allow_rename and path != fallback_path: + name = _sanitize_note_name(current_name, fallback_name) + if suggested_name != name: + await self._set_frontmatter_name(path, name) + return path + + name = _sanitize_note_name(suggested_name, fallback_name) + name, target_path = self._unique_daily_note_path(day, name, file_path, path) + if suggested_name != name: + await self._set_frontmatter_name(path, name) + + if target_path == path: + return path + + move_response = await self.run_job( + "move", + src_path=path, + dst_path=target_path, + overwrite=False, + retarget=True, + ) + if not move_response.success: + raise RuntimeError(f"move failed: {move_response.answer}") + return target_path + + async def _handle_delete(self, file_path: str, date_str: str, note_stem: str) -> None: + daily_dir = self.config_value("daily_dir") + fallback_path = f"{daily_dir}/{date_str}/{note_stem}.md" + try: + note = await self._list_resource_note(date_str, file_path, fallback_path) + except RuntimeError as exc: + self.context.response.success = False + self.context.response.answer = str(exc) + self.logger.info(f"[{self.name}] delete list failed file_path={file_path} answer={str(exc)!r}") + return + + note_rel = str(note["path"]) if note else fallback_path + note_abs = self.workspace_path / note_rel + note_existed = note_abs.is_file() + self.logger.info(f"[{self.name}] delete start note={note_rel}") + + if note_existed: + note_abs.unlink() + self.logger.info(f"[{self.name}] Deleted file: {note_rel}") + + await self.file_store.delete([note_rel]) + self.logger.info(f"[{self.name}] catalog delete done note={note_rel}") + self.logger.info(f"[{self.name}] refresh index start date={date_str} daily_dir={daily_dir}") + index_payload = await refresh_day_index(self.file_store, date_str, daily_dir) + self.logger.info(f"[{self.name}] refresh index done date={date_str}") + + self.context.response.success = True + self.context.response.answer = f"Deleted resource note: {note_rel}" + self.context.response.metadata.update( + { + "path": note_rel, + "session_id": note_stem, + "source_resource": self._source_resource_link(file_path), + "action": "deleted", + "modified": note_existed, + "index": index_payload, + }, + ) + + @abstractmethod + async def _handle_upsert( + self, + file_path: str, + date_str: str, + note_stem: str, + added: bool, + ) -> None: + """Interpret one added or modified resource into its daily note.""" + + async def _handle_change(self, file_path: str, raw_change) -> dict: + assert self.context is not None + # Handlers write item-scoped fields into the shared response. Start each + # change with a fresh mapping so one result cannot inherit another's metadata. + self.context.response.metadata = {} + file_path = self.to_workspace_relative(file_path) if file_path and Path(file_path).is_absolute() else file_path + if not file_path: + self.context.response.success = False + self.context.response.answer = "Missing file_path" + self.logger.warning(f"[{self.name}] missing file_path change={raw_change!r}") + return {"success": False, "path": file_path, "change": raw_change, "answer": self.context.response.answer} + + change = self._normalize_change(raw_change) + if change is None: + self.context.response.success = False + self.context.response.answer = f"Invalid change type: {raw_change}" + self.logger.warning(f"[{self.name}] invalid change file_path={file_path} change={raw_change!r}") + return {"success": False, "path": file_path, "change": raw_change, "answer": self.context.response.answer} + + resource_dir = self.config_value("resource_dir") + loose_filename = _loose_resource_filename(file_path, resource_dir) + if loose_filename: + date_str, filename = self._today(), loose_filename + self.logger.info(f"[{self.name}] loose resource file_path={file_path} date={date_str}") + else: + date_str, filename = _parse_resource_path(file_path, resource_dir) + + if not date_str or not filename: + self.context.response.success = False + self.context.response.answer = f"Cannot parse date/filename from: {file_path}" + self.logger.warning(f"[{self.name}] parse path failed file_path={file_path} resource_dir={resource_dir}") + return {"success": False, "path": file_path, "change": change.name, "answer": self.context.response.answer} + + note_stem = _compute_note_stem(filename) + self.logger.info(f"[{self.name}] {change.name} file_path={file_path} note_stem={note_stem}") + + if change == Change.deleted: + await self._handle_delete(file_path, date_str, note_stem) + else: + await self._handle_upsert( + file_path, + date_str, + note_stem, + change == Change.added, + ) + return { + "success": self.context.response.success, + "path": file_path, + "change": change.name, + "answer": self.context.response.answer, + "metadata": dict(self.context.response.metadata), + } + + def _failed_change_result(self, file_path: str, raw_change, exc: Exception) -> dict: + """Convert one unexpected processor error into an item-scoped result.""" + assert self.context is not None + file_path = str(file_path or "") + if Path(file_path).is_absolute(): + file_path = self.to_workspace_relative(file_path) + change = self._normalize_change(raw_change) + change_name = change.name if change is not None else str(raw_change) + answer = f"Failed to process resource: {file_path}: {exc}" + metadata = dict(self.context.response.metadata) + metadata.setdefault("path", file_path) + metadata.setdefault("modified", False) + metadata.update({"action": "failed", "error": str(exc)}) + self.context.response.success = False + self.context.response.answer = answer + self.context.response.metadata = metadata + self.logger.exception(f"[{self.name}] resource failed file_path={file_path} error={exc}") + return { + "success": False, + "path": file_path, + "change": change_name, + "answer": answer, + "metadata": dict(metadata), + } + + async def execute(self): + assert self.context is not None + changes = self.context.get("changes") + if not isinstance(changes, list): + self.context.response.success = False + self.context.response.answer = "AutoResourceStep requires changes: list[dict]" + self.logger.warning(f"[{self.name}] invalid changes payload type={type(changes).__name__}") + return self.context.response + + self.logger.info(f"[{self.name}] start changes={len(changes)}") + results = [] + for index, item in enumerate(changes, start=1): + if not isinstance(item, dict): + self.logger.warning(f"[{self.name}] skip invalid change item index={index} type={type(item).__name__}") + continue + self.logger.info(f"[{self.name}] process change {index}/{len(changes)}") + file_path = item.get("path") or item.get("file_path", "") + raw_change = item.get("change", "") + self.context.response.metadata = {} + try: + result = await self._handle_change(file_path, raw_change) + except Exception as exc: # pylint: disable=broad-except + result = self._failed_change_result(file_path, raw_change, exc) + results.append(result) + success_count = sum(1 for item in results if item.get("success")) + self.context.response.success = success_count == len(changes) + processed_answer = f"Processed {success_count}/{len(changes)} resource change(s)" + self.context.response.answer = _results_answer(results, processed_answer) + self.context.response.metadata["processed"] = len(results) + self.context.response.metadata["results"] = results + self.context.response.metadata["modified"] = any( + bool((item.get("metadata") or {}).get("modified")) for item in results + ) + self.logger.info( + f"[{self.name}] done success={success_count}/{len(changes)} " + f"processed={len(results)} modified={self.context.response.metadata['modified']}", + ) + return self.context.response diff --git a/reme/steps/evolve/auto_image.py b/reme/steps/evolve/auto_image.py index ca643751..50a5e7ab 100644 --- a/reme/steps/evolve/auto_image.py +++ b/reme/steps/evolve/auto_image.py @@ -1,36 +1,27 @@ -"""auto_image — interpret image resource files into source-linked daily notes via a VLM.""" +"""Image resource processor for the unified auto-resource router.""" import base64 import io import json import re -from pathlib import Path +from pathlib import Path, PurePosixPath import aiofiles from agentscope.message import Base64Source, DataBlock, TextBlock, UserMsg from agentscope.model import ChatModelBase -from PIL import Image from pydantic import BaseModel, Field from ..file_io import is_image_file, refresh_day_index -from ..file_io._path import IMAGE_MIME_BY_EXT -from .auto_resource import _SOURCE_RESOURCE_KEY, _sanitize_note_name, AutoResourceStep +from ..file_io._path import IMAGE_MIME_BY_EXT, IMAGE_SUFFIXES +from ._auto_resource import _SOURCE_RESOURCE_KEY, _sanitize_note_name, BaseAutoResourceStep from ...components import R from ...enumeration import ComponentEnum -try: - from pillow_heif import register_heif_opener - - register_heif_opener() -except ImportError: - # Without the plugin HEIC files stay passthrough and the provider rejects - # them per change; the pipeline itself keeps importing and running. - pass - DEFAULT_MAX_IMAGE_INPUT_BYTES = 50 * 1024 * 1024 MAX_IMAGE_REQUEST_DIMENSION = 2048 _JPEG_QUALITY = 85 -# Suffixes re-encoded to PNG for VLM requests; the stored resource file is never modified. +# Suffixes re-encoded to provider-friendly PNG/JPEG for VLM requests; +# the stored resource file is never modified. _CONVERT_SUFFIXES = {".bmp", ".tiff", ".heic"} _JSON_FENCE_RE = re.compile(r"^\s*```(?:json)?\s*(.*?)\s*```\s*$", re.DOTALL) @@ -38,36 +29,70 @@ _JSON_FENCE_RE = re.compile(r"^\s*```(?:json)?\s*(.*?)\s*```\s*$", re.DOTALL) class _CaptionOutput(BaseModel): """Structured caption contract enforced on the vision model.""" - name: str = Field(description="short kebab-case topic stem for the note filename; never include dates") - description: str = Field(description="one-sentence summary that conveys the key information on its own") - caption: str = Field(description="complete description / verbatim transcription of meaningful visible text") + name: str = Field( + description="short kebab-case topic stem based on visible content; filename is only a weak naming hint", + ) + description: str = Field(description="one-sentence summary of visible image content that stands on its own") + caption: str = Field( + description="complete description / verbatim transcription of meaningful content visible in the image", + ) + + +def _load_pillow(suffix: str): + """Load Pillow and optional HEIC support only when image processing runs.""" + try: + from PIL import Image # pylint: disable=import-outside-toplevel + except ImportError as exc: + raise RuntimeError("Image captioning requires Pillow; install reme-ai[core]") from exc + + if suffix == ".heic": + try: + from pillow_heif import register_heif_opener # pylint: disable=import-outside-toplevel + except ImportError as exc: + raise RuntimeError("HEIC image captioning requires pillow-heif; install reme-ai[image-heif]") from exc + try: + register_heif_opener() + except Exception as exc: # pylint: disable=broad-except + raise RuntimeError(f"Failed to initialize HEIC image support: {exc}") from exc + return Image def _normalize_image_bytes(data: bytes, suffix: str) -> tuple[bytes, str] | None: """Downscale or re-encode image bytes in memory for a VLM request. - Returns ``None`` to pass the original bytes through untouched (already a - reasonable size and format, or content that PIL cannot decode — left for - the provider to reject). The stored resource file is never modified. + Returns ``None`` only when valid image bytes already have a suitable size + and format. Missing dependencies and decode/convert failures are explicit. + The stored resource file is never modified. """ + image_module = _load_pillow(suffix) try: - with Image.open(io.BytesIO(data)) as image: - needs_resize = image.width > MAX_IMAGE_REQUEST_DIMENSION or image.height > MAX_IMAGE_REQUEST_DIMENSION - needs_convert = suffix in _CONVERT_SUFFIXES - if not needs_resize and not needs_convert: - return None + image = image_module.open(io.BytesIO(data)) + except Exception as exc: # pylint: disable=broad-except + raise RuntimeError(f"Failed to decode image ({suffix or 'unknown suffix'}): {exc}") from exc + + with image: + try: + image.load() + except Exception as exc: # pylint: disable=broad-except + raise RuntimeError(f"Failed to decode image ({suffix or 'unknown suffix'}): {exc}") from exc + + needs_resize = image.width > MAX_IMAGE_REQUEST_DIMENSION or image.height > MAX_IMAGE_REQUEST_DIMENSION + needs_convert = suffix in _CONVERT_SUFFIXES + if not needs_resize and not needs_convert: + return None + try: has_alpha = image.mode in ("RGBA", "LA", "P") frame = image.convert("RGBA" if has_alpha else "RGB") if needs_resize: - frame.thumbnail((MAX_IMAGE_REQUEST_DIMENSION, MAX_IMAGE_REQUEST_DIMENSION), Image.LANCZOS) + frame.thumbnail((MAX_IMAGE_REQUEST_DIMENSION, MAX_IMAGE_REQUEST_DIMENSION), image_module.LANCZOS) buffer = io.BytesIO() if frame.mode == "RGBA": frame.save(buffer, format="PNG") return buffer.getvalue(), "image/png" frame.save(buffer, format="JPEG", quality=_JPEG_QUALITY) return buffer.getvalue(), "image/jpeg" - except Exception: # pylint: disable=broad-except - return None + except Exception as exc: # pylint: disable=broad-except + raise RuntimeError(f"Failed to convert/resize image ({suffix or 'unknown suffix'}): {exc}") from exc def _build_image_request_payload(data: bytes, suffix: str) -> dict: @@ -152,21 +177,20 @@ def _parse_caption_json(text: str) -> dict: return {"name": "", "description": "", "caption": text.strip()} -@R.register("auto_image_step") -class AutoImageStep(AutoResourceStep): +@R.register("auto_image_resource_step") +class AutoImageResourceStep(BaseAutoResourceStep): """Interpret image resource files into daily notes via a direct VLM call. Unlike text resources (agent + file tools), the image interpretation is a single vision-model call. Images larger than the request budget or in provider-unfriendly formats are downscaled/re-encoded in memory for the request only; files under ``resource/`` are never modified. Note lookup, - renaming, deletion linkage, and day-index refresh reuse the inherited - AutoResourceStep helpers; only the interpretation differs. + renaming, deletion linkage, and day-index refresh reuse the shared + BaseAutoResourceStep lifecycle; only the interpretation differs. """ - def _skip_image_change(self, file_path: str) -> bool: - """This step interprets images, so the inherited image guard never applies.""" - return False + resource_suffixes = IMAGE_SUFFIXES + router_inherit_keys = BaseAutoResourceStep.router_inherit_keys | frozenset({"as_llm", "max_image_bytes"}) def _max_image_bytes(self) -> int: """Return the image read limit from Step or Job context.""" @@ -176,19 +200,18 @@ class AutoImageStep(AutoResourceStep): return int(value) if value is not None else DEFAULT_MAX_IMAGE_INPUT_BYTES def _vision_model(self) -> ChatModelBase | None: - """Resolve the vision model: an explicit instance, then the ``vision`` - named as_llm component, then the ``default`` one.""" - for source in (self.kwargs, self.context or {}): - value = source.get("as_llm") - if isinstance(value, ChatModelBase): - return value + """Resolve explicit ``as_llm`` through Ref, otherwise prefer vision/default.""" + context_model = self.context.get("as_llm") if self.context is not None else None + if "as_llm" in self.kwargs or isinstance(context_model, ChatModelBase): + return self.as_llm if self.app_context is None: return None models = self.app_context.components.get(ComponentEnum.AS_LLM, {}) - component = models.get("vision") or models.get("default") - if component is None: - return None - return getattr(component, "model", None) + for name in ("vision", "default"): + if name in models: + self.kwargs["as_llm"] = name + return self.as_llm + return None async def _caption_with_retry(self, model: ChatModelBase, user_message: UserMsg) -> dict: """Return the caption fields from the vision model. @@ -298,12 +321,12 @@ class AutoImageStep(AutoResourceStep): except Exception as exc: # pylint: disable=broad-except self.context.response.success = False self.context.response.answer = f"Failed to caption image resource: {file_path}: {exc}" + self.context.response.metadata.setdefault("path", file_path) + self.context.response.metadata.setdefault("modified", False) self.context.response.metadata.update( { - "path": file_path, "action": "failed", "error": str(exc), - "modified": False, }, ) self.logger.warning(f"[{self.name}] caption failed file_path={file_path} error={exc}") @@ -357,7 +380,15 @@ class AutoImageStep(AutoResourceStep): user_message = UserMsg( name="user", content=[ - TextBlock(text=self.prompt_format("user_message", file_path=file_path, date=date_str)), + TextBlock( + text=self.prompt_format( + "user_message", + file_path=file_path, + filename=PurePosixPath(file_path).name, + stem=note_stem, + date=date_str, + ), + ), DataBlock( source=Base64Source(data=payload["data_b64"], media_type=payload["mime"]), name="image", @@ -389,6 +420,13 @@ class AutoImageStep(AutoResourceStep): ) if not write_response.success: raise RuntimeError(f"write failed: {write_response.answer}") + self.context.response.metadata.update( + { + "path": note_path, + "created": note_created, + "modified": self._note_modified(before_note_path, before_note_bytes, note_path), + }, + ) if note_created: try: @@ -396,14 +434,21 @@ class AutoImageStep(AutoResourceStep): except RuntimeError as exc: self.context.response.success = False self.context.response.answer = str(exc) - self.context.response.metadata.update({"path": None, "created": note_created, "modified": False}) + self.context.response.metadata.update( + { + "path": note_path, + "created": note_created, + "modified": self._note_modified(before_note_path, before_note_bytes, note_path), + }, + ) self.logger.info(f"[{self.name}] post-create list failed file_path={file_path} answer={str(exc)!r}") return if note is None: self.context.response.success = True self.context.response.answer = f"Captioned image resource {file_path}" - self.context.response.metadata.update({"path": None, "created": False, "modified": False}) - self.logger.info(f"[{self.name}] done without note file_path={file_path} modified=False") + modified = self._note_modified(before_note_path, before_note_bytes, note_path) + self.context.response.metadata.update({"path": note_path, "created": False, "modified": modified}) + self.logger.info(f"[{self.name}] done without cataloged note file_path={file_path} modified={modified}") return note_path = str(note["path"]) @@ -431,6 +476,7 @@ class AutoImageStep(AutoResourceStep): return modified = self._note_modified(before_note_path, before_note_bytes, note_path) + self.context.response.metadata.update({"path": note_path, "created": note_created, "modified": modified}) index_payload = await refresh_day_index(self.file_store, date_str, daily_dir) self.context.response.success = True diff --git a/reme/steps/evolve/auto_image.yaml b/reme/steps/evolve/auto_image.yaml index 08e8e482..16ef930f 100644 --- a/reme/steps/evolve/auto_image.yaml +++ b/reme/steps/evolve/auto_image.yaml @@ -1,9 +1,17 @@ +# AutoImageResourceStep prompts. user_message: | Describe the attached image from the user's resource library for a memory knowledge base. - Resource image file: {file_path} + Resource image path: {file_path} + Filename: {filename} + Filename stem: {stem} Date: {date} + The filename and stem are weak hints for naming or disambiguation only. Do not + treat words in them as visible facts. If a filename conflicts with the image, + trust the visible image content. The description and caption must be grounded + in visible content, not inferred from the filename. + ## What to Record - **Visible facts**: people, objects, places, actions, and relationships that may matter in future conversations. @@ -32,9 +40,14 @@ user_message: | user_message_zh: | 为记忆知识库描述用户资源库中的这张图像。 - 资源图像文件:{file_path} + 资源图像路径:{file_path} + 文件名:{filename} + 文件名 stem:{stem} 日期:{date} + 文件名和 stem 只能作为命名或消歧时的弱提示,不能当作图中可见事实。如果文件名与图像内容冲突,以图像中的可见内容为准。 + description 和 caption 必须基于图像中的可见内容,不得从文件名推断事实。 + ## 记录什么 - **可见事实**:人物、物体、地点、动作及相互关系——未来对话中可能重要的信息。 - **图中的文字**:逐字转录有意义的可见文字(幻灯片、白板、截图、招牌、标签);保留原语言,不翻译、不纠错。 diff --git a/reme/steps/evolve/auto_resource.py b/reme/steps/evolve/auto_resource.py index 597d6b8b..1ce6aec1 100644 --- a/reme/steps/evolve/auto_resource.py +++ b/reme/steps/evolve/auto_resource.py @@ -1,246 +1,99 @@ -"""auto_resource — interpret resource files into source-linked daily notes via an agent.""" +"""Unified processor router for automatic resource interpretation.""" -import hashlib +import copy import inspect -import re -import uuid -from pathlib import Path, PurePosixPath -import aiofiles -import frontmatter -from watchfiles import Change - -from ..base_step import BaseStep -from ..file_io import is_image_file, refresh_day_index, validate_filename_component from ...components import R -from ._evolve import agent_reply_result_text, now +from ..base_step import BaseStep +from ._auto_resource import BaseAutoResourceStep, _results_answer -_SOURCE_RESOURCE_KEY = "source_resource" -_DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$") -_UNSAFE_FILENAME_CHARS = re.compile(r'[<>:"/\\|?*\x00-\x1f]+') - - -def _compute_agent_session_id(path: str) -> str: - """Return a stable UUID session id for agent backends.""" - return str(uuid.uuid5(uuid.NAMESPACE_URL, path)) - - -def _compute_note_stem(filename: str) -> str: - """Return the daily note stem for a resource filename.""" - return PurePosixPath(filename).stem - - -def _parse_resource_path(file_path: str, resource_dir: str) -> tuple[str, str]: - """Extract (date, filename) from a resource path like 'resource/2026-06-06/report.pdf'. - - Returns (date_str, filename) where filename may contain subdirectories. - """ - parts = PurePosixPath(file_path).parts - # Strip leading resource_dir prefix - prefix_parts = PurePosixPath(resource_dir).parts - if parts[: len(prefix_parts)] != prefix_parts: - return "", "" - parts = parts[len(prefix_parts) :] - # First segment is date, rest is filename - date_str = parts[0] if parts else "" - if not _DATE_RE.match(date_str): - return "", "" - filename = str(PurePosixPath(*parts[1:])) if len(parts) > 1 else "" - return date_str, filename - - -def _loose_resource_filename(file_path: str, resource_dir: str) -> str: - """Return filename for a root-level resource path like 'resource/report.txt'.""" - parts = PurePosixPath(file_path).parts - prefix_parts = PurePosixPath(resource_dir).parts - if parts[: len(prefix_parts)] != prefix_parts: - return "" - rest = parts[len(prefix_parts) :] - if len(rest) != 1: - return "" - filename = rest[0] - return "" if filename in ("", ".", "..") else filename - - -def _results_answer(results: list[dict], processed_answer: str) -> str: - """Return the actual per-change answer while preserving a batch fallback.""" - answers = [str(item.get("answer") or "").strip() for item in results] - answers = [item for item in answers if item] - if len(answers) == 1: - return answers[0] - if len(answers) > 1: - return "\n\n".join(f"{index}. {answer}" for index, answer in enumerate(answers, start=1)) - return processed_answer - - -def _source_suffix(file_path: str) -> str: - """Return a short stable suffix for source-path collision handling.""" - return hashlib.sha1(file_path.encode("utf-8")).hexdigest()[:8] - - -def _sanitize_note_name(raw: str, fallback: str) -> str: - """Return a safe single filename component from an LLM-suggested name.""" - name = str(raw or "").strip() - name = _UNSAFE_FILENAME_CHARS.sub("-", name) - name = re.sub(r"\s+", " ", name).strip(" .") - if not name: - name = str(fallback or "").strip() - name = _UNSAFE_FILENAME_CHARS.sub("-", name).strip(" .") - if not name or validate_filename_component(name, kind="name"): - name = f"resource-{_source_suffix(fallback or raw or 'note')}" - if validate_filename_component(name, kind="name"): - name = f"resource-{_source_suffix(name)}" - return name +_ProcessorSpec = str | dict +_IndexedChange = tuple[int, dict] +_ProcessorRoute = tuple[dict, type[BaseAutoResourceStep], list[_IndexedChange]] @R.register("auto_resource_step") class AutoResourceStep(BaseStep): - """Interpret resource files into daily notes via an Agent.""" + """Route each resource change to one configured processor and aggregate once.""" def __init__(self, **kwargs): + router_options = dict(kwargs) super().__init__(**kwargs) - self.create_tools: list[str] = ["write"] - self.update_tools: list[str] = ["read", "edit", "frontmatter_update", "write"] + self._router_options = router_options - def _normalize_change(self, raw) -> Change | None: - if isinstance(raw, Change): - return raw - if isinstance(raw, str): - return Change.__members__.get(raw) - return None + def _processor_spec(self, spec: _ProcessorSpec, step_cls: type[BaseAutoResourceStep]) -> dict: + """Merge router options declared by the processor into its explicit Step spec.""" + params = {"backend": spec} if isinstance(spec, str) else dict(spec) + inherited = { + key: self._router_options[key] + for key in step_cls.router_inherit_keys + if key in self._router_options and self._router_options[key] is not None + } + return {**inherited, **params} - def _today(self) -> str: - tz = self.app_context.app_config.timezone if self.app_context is not None else None - return now(tz).strftime("%Y-%m-%d") + def _processor_routes(self) -> list[_ProcessorRoute]: + """Resolve configured processors and allocate their per-invocation batches.""" + raw_specs = self.dispatch_step_specs + if not raw_specs: + raise RuntimeError("AutoResourceStep requires resource processors in dispatch_steps") - def _daily_note_path(self, day: str, name: str) -> str: - return f"{self.config_value('daily_dir')}/{day}/{name}.md" + routes: list[_ProcessorRoute] = [] + fallback_indexes: list[int] = [] + for index, raw_spec in enumerate(raw_specs): + step_cls, _ = self._resolve_dispatch_step(raw_spec) + if not isinstance(step_cls, type) or not issubclass(step_cls, BaseAutoResourceStep): + backend = raw_spec if isinstance(raw_spec, str) else raw_spec.get("backend", "") + raise TypeError(f"Resource processor '{backend}' must inherit BaseAutoResourceStep") + spec = self._processor_spec(raw_spec, step_cls) + if step_cls.resource_fallback: + fallback_indexes.append(index) + routes.append((spec, step_cls, [])) + if len(fallback_indexes) > 1: + raise ValueError("AutoResourceStep accepts at most one fallback processor") + if fallback_indexes and fallback_indexes[0] != len(routes) - 1: + raise ValueError("AutoResourceStep fallback processor must be last in dispatch_steps") + return routes + + async def _dispatch_processor( + self, + spec: dict, + indexed_changes: list[_IndexedChange], + result_slots: list[dict | None], + ) -> None: + """Dispatch one routed sub-batch and snapshot its shared Response immediately.""" + if not indexed_changes: + return + changes = [item for _, item in indexed_changes] + responses = await self.dispatch_steps([spec], changes=changes) + processor_response = responses[-1] + processor_results = copy.deepcopy(processor_response.metadata.get("results") or []) + if len(processor_results) != len(indexed_changes): + raise RuntimeError( + f"Resource processor returned {len(processor_results)} result(s) " + f"for {len(indexed_changes)} change(s)", + ) + for (index, _), result in zip(indexed_changes, processor_results): + result_slots[index] = result @staticmethod - def _source_resource_link(file_path: str) -> str: - return f"[[{file_path}]]" - - def _frontmatter(self, path: str) -> dict: - post = frontmatter.loads((self.file_store.workspace_path / path).read_text(encoding="utf-8")) - return dict(post.metadata or {}) - - def _note_bytes(self, path: str) -> bytes | None: - note_path = self.file_store.workspace_path / path - if not note_path.is_file(): - return None - return note_path.read_bytes() - - def _note_modified(self, before_path: str, before_bytes: bytes | None, after_path: str) -> bool: - if not after_path: - return False - after_bytes = self._note_bytes(after_path) - if after_bytes is None: - return before_bytes is not None - return after_path != before_path or before_bytes != after_bytes - - def _find_resource_note(self, notes: list[dict], file_path: str, fallback_path: str) -> dict | None: - source = self._source_resource_link(file_path) - for note in notes: - if str(note.get(_SOURCE_RESOURCE_KEY, "")).strip() == source: - return note - for note in notes: - if str(note.get("path", "")).strip() == fallback_path: - return note - return None - - async def _list_resource_note(self, day: str, file_path: str, fallback_path: str) -> dict | None: - list_response = await self.run_job("daily_list", date=day) - if not list_response.success: - raise RuntimeError(f"daily_list failed: {list_response.answer}") - notes = list_response.metadata.get("notes") or [] - return self._find_resource_note(notes, file_path, fallback_path) - - async def _ensure_resource_frontmatter(self, path: str, file_path: str) -> None: - metadata = {_SOURCE_RESOURCE_KEY: self._source_resource_link(file_path)} - current = self._frontmatter(path) - if all(current.get(key) == value for key, value in metadata.items()): - return - response = await self.run_job( - "frontmatter_update", - path=path, - metadata=metadata, - ) - if not response.success: - raise RuntimeError(f"frontmatter_update failed: {response.answer}") - - async def _set_frontmatter_name(self, path: str, name: str) -> None: - if self._frontmatter(path).get("name") == name: - return - response = await self.run_job("frontmatter_update", path=path, metadata={"name": name}) - if not response.success: - raise RuntimeError(f"frontmatter_update failed: {response.answer}") - - def _unique_daily_note_path(self, day: str, name: str, file_path: str, current_path: str) -> tuple[str, str]: - """Return a collision-free (name, path), preserving current_path when possible.""" - target_path = self._daily_note_path(day, name) - target_abs = self.file_store.workspace_path / target_path - if target_path == current_path or not target_abs.exists(): - return name, target_path - - suffixed = f"{name}--{_source_suffix(file_path)}" - target_path = self._daily_note_path(day, suffixed) - target_abs = self.file_store.workspace_path / target_path - if target_path == current_path or not target_abs.exists(): - return suffixed, target_path - - for index in range(2, 100): - candidate = f"{suffixed}-{index}" - target_path = self._daily_note_path(day, candidate) - target_abs = self.file_store.workspace_path / target_path - if target_path == current_path or not target_abs.exists(): - return candidate, target_path - raise RuntimeError(f"cannot allocate unique note name for: {name!r}") - - async def _rename_from_frontmatter_name( - self, - path: str, - day: str, - file_path: str, - fallback_name: str, - fallback_path: str, - *, - allow_rename: bool, - ) -> str: - meta = self._frontmatter(path) - current_name = PurePosixPath(path).stem - suggested_name = str(meta.get("name", "")).strip() - - if not allow_rename and path != fallback_path: - name = _sanitize_note_name(current_name, fallback_name) - if suggested_name != name: - await self._set_frontmatter_name(path, name) - return path - - name = _sanitize_note_name(suggested_name, fallback_name) - name, target_path = self._unique_daily_note_path(day, name, file_path, path) - if suggested_name != name: - await self._set_frontmatter_name(path, name) - - if target_path == path: - return path - - move_response = await self.run_job( - "move", - src_path=path, - dst_path=target_path, - overwrite=False, - retarget=True, - ) - if not move_response.success: - raise RuntimeError(f"move failed: {move_response.answer}") - return target_path + def _unsupported_result(item: dict, file_path: str) -> dict: + """Return a stable failure result when no configured processor accepts a resource.""" + answer = f"No configured resource processor accepts: {file_path}" + return { + "success": False, + "path": file_path, + "change": str(item.get("change", "")), + "answer": answer, + "metadata": { + "path": file_path, + "action": "failed", + "reason": "unsupported_resource", + "modified": False, + }, + } async def _emit_result_hook(self, *, changes: list[dict], results: list[dict]) -> None: - """Notify embedding hosts about the final auto-resource response. - - The hook is intentionally optional so standalone ReMe and old configs - keep the existing behavior. - """ + """Notify embedding hosts once with the final aggregate response.""" if self.app_context is None or self.context is None: return metadata = getattr(self.app_context, "metadata", None) @@ -267,292 +120,6 @@ class AutoResourceStep(BaseStep): except Exception: self.logger.exception(f"[{self.name}] result hook failed") - async def _handle_delete(self, file_path: str, date_str: str, note_stem: str) -> None: - daily_dir = self.config_value("daily_dir") - fallback_path = f"{daily_dir}/{date_str}/{note_stem}.md" - try: - note = await self._list_resource_note(date_str, file_path, fallback_path) - except RuntimeError as exc: - self.context.response.success = False - self.context.response.answer = str(exc) - self.logger.info(f"[{self.name}] delete list failed file_path={file_path} answer={str(exc)!r}") - return - - note_rel = str(note["path"]) if note else fallback_path - note_abs = self.workspace_path / note_rel - note_existed = note_abs.is_file() - self.logger.info(f"[{self.name}] delete start note={note_rel}") - - if note_existed: - note_abs.unlink() - self.logger.info(f"[{self.name}] Deleted file: {note_rel}") - - await self.file_store.delete([note_rel]) - self.logger.info(f"[{self.name}] catalog delete done note={note_rel}") - self.logger.info(f"[{self.name}] refresh index start date={date_str} daily_dir={daily_dir}") - index_payload = await refresh_day_index(self.file_store, date_str, daily_dir) - self.logger.info(f"[{self.name}] refresh index done date={date_str}") - - self.context.response.success = True - self.context.response.answer = f"Deleted resource note: {note_rel}" - self.context.response.metadata.update( - { - "path": note_rel, - "session_id": note_stem, - "source_resource": self._source_resource_link(file_path), - "action": "deleted", - "modified": note_existed, - "index": index_payload, - }, - ) - - async def _handle_upsert( - self, - file_path: str, - date_str: str, - note_stem: str, - added: bool, - ) -> None: - self.logger.info( - f"[{self.name}] upsert start file_path={file_path} date={date_str} " f"note_stem={note_stem} added={added}", - ) - daily_dir = self.config_value("daily_dir") - fallback_path = f"{daily_dir}/{date_str}/{note_stem}.md" - try: - note = await self._list_resource_note(date_str, file_path, fallback_path) - except RuntimeError as exc: - self.context.response.success = False - self.context.response.answer = str(exc) - self.logger.info(f"[{self.name}] list failed file_path={file_path} answer={str(exc)!r}") - return - - note_path = str(note["path"]) if note else fallback_path - note_created = note is None - before_note_path = note_path - before_note_bytes = self._note_bytes(note_path) - self.logger.info(f"[{self.name}] daily note lookup path={note_path} created={note_created}") - - # Read resource file content - abs_path = self.workspace_path / file_path - if not abs_path.is_file(): - self.context.response.success = False - self.context.response.answer = f"Resource file not found: {file_path}" - self.logger.warning(f"[{self.name}] resource missing file_path={file_path}") - return - - skip_read = False - try: - size_bytes = abs_path.stat().st_size - except OSError as exc: - self.context.response.success = False - self.context.response.answer = f"Failed to inspect resource file: {file_path}: {exc}" - self.context.response.metadata.update( - { - "path": file_path, - "action": "failed", - "error": str(exc), - "modified": False, - }, - ) - self.logger.warning(f"[{self.name}] resource stat failed file_path={file_path} error={exc}") - skip_read = True - if not skip_read: - max_file_bytes = self.max_file_bytes() - if size_bytes > max_file_bytes: - self.context.response.success = True - self.context.response.answer = ( - f"Skipped oversized resource file: {file_path} ({size_bytes} > {max_file_bytes} bytes)" - ) - self.context.response.metadata.update( - { - "path": file_path, - "action": "skipped", - "reason": "file_too_large", - "oversized": True, - "size_bytes": size_bytes, - "max_file_bytes": max_file_bytes, - "modified": False, - }, - ) - self.logger.warning( - f"[{self.name}] skip oversized resource file_path={file_path} " - f"size_bytes={size_bytes} max_file_bytes={max_file_bytes}", - ) - skip_read = True - if skip_read: - return - - self.logger.info(f"[{self.name}] read resource start file_path={file_path}") - async with aiofiles.open(abs_path, encoding="utf-8", errors="replace") as f: - file_content = await f.read() - self.logger.info(f"[{self.name}] read resource done file_path={file_path} chars={len(file_content)}") - - template_key = "user_message_create" if note_created else "user_message_update" - user_message = self.prompt_format( - template_key, - workspace_dir=str(self.workspace_path), - note_path=note_path, - note_stem=note_stem, - file_path=file_path, - source_resource=self._source_resource_link(file_path), - file_content=file_content, - date=date_str, - ) - - agent_session_id = _compute_agent_session_id(file_path) - self.logger.info( - f"[{self.name}] agent start file_path={file_path} note_path={note_path} " - f"agent_session_id={agent_session_id}", - ) - result = await self.agent_wrapper.reply( - user_message, - system_prompt=self.prompt_format("system_prompt"), - job_tools=self.create_tools if note_created else self.update_tools, - session_id=agent_session_id, - ) - self.logger.info(f"[{self.name}] agent done file_path={file_path} has_result={bool(result.get('result'))}") - - if note_created: - try: - note = await self._list_resource_note(date_str, file_path, fallback_path) - except RuntimeError as exc: - self.context.response.success = False - self.context.response.answer = str(exc) - self.context.response.metadata.update({"path": None, "created": note_created, "modified": False}) - self.logger.info(f"[{self.name}] post-create list failed file_path={file_path} answer={str(exc)!r}") - return - if note is None: - self.context.response.success = True - self.context.response.answer = agent_reply_result_text(result) - self.context.response.metadata.update({"path": None, "created": False, "modified": False}) - self.logger.info(f"[{self.name}] done without note file_path={file_path} modified=False") - return - note_path = str(note["path"]) - - try: - await self._ensure_resource_frontmatter(note_path, file_path) - note_path = await self._rename_from_frontmatter_name( - note_path, - date_str, - file_path, - note_stem, - fallback_path, - allow_rename=note_created, - ) - except RuntimeError as exc: - self.context.response.success = False - self.context.response.answer = str(exc) - self.context.response.metadata.update( - { - "path": note_path, - "created": note_created, - "modified": self._note_modified(before_note_path, before_note_bytes, note_path), - }, - ) - self.logger.info(f"[{self.name}] post-agent failed path={note_path} answer={str(exc)!r}") - return - - modified = self._note_modified(before_note_path, before_note_bytes, note_path) - self.logger.info(f"[{self.name}] refresh index start date={date_str} daily_dir={daily_dir}") - index_payload = await refresh_day_index(self.file_store, date_str, daily_dir) - self.logger.info(f"[{self.name}] refresh index done date={date_str}") - - self.context.response.success = True - self.context.response.answer = agent_reply_result_text(result) - self.context.response.metadata.update( - { - "path": note_path, - "created": note_created, - "modified": modified, - "session_id": note_stem, - "source_resource": self._source_resource_link(file_path), - "agent_session_id": agent_session_id, - "action": "added" if added else "modified", - "index": index_payload, - }, - ) - self.logger.info(f"[{self.name}] done {note_path} modified={modified}") - - def _skip_image_change(self, file_path: str) -> bool: - """Return True for image changes, which auto_image_step interprets.""" - return is_image_file(file_path) - - async def _handle_change(self, file_path: str, raw_change) -> dict: - assert self.context is not None - # Handlers write item-scoped fields into the shared response. Start each - # change with a fresh mapping so one result cannot inherit another's metadata. - self.context.response.metadata = {} - file_path = self.to_workspace_relative(file_path) if file_path and Path(file_path).is_absolute() else file_path - if not file_path: - self.context.response.success = False - self.context.response.answer = "Missing file_path" - self.logger.warning(f"[{self.name}] missing file_path change={raw_change!r}") - return {"success": False, "path": file_path, "change": raw_change, "answer": self.context.response.answer} - - change = self._normalize_change(raw_change) - if change is None: - self.context.response.success = False - self.context.response.answer = f"Invalid change type: {raw_change}" - self.logger.warning(f"[{self.name}] invalid change file_path={file_path} change={raw_change!r}") - return {"success": False, "path": file_path, "change": raw_change, "answer": self.context.response.answer} - - resource_dir = self.config_value("resource_dir") - loose_filename = _loose_resource_filename(file_path, resource_dir) - if loose_filename: - date_str, filename = self._today(), loose_filename - self.logger.info(f"[{self.name}] loose resource file_path={file_path} date={date_str}") - else: - date_str, filename = _parse_resource_path(file_path, resource_dir) - - if not date_str or not filename: - self.context.response.success = False - self.context.response.answer = f"Cannot parse date/filename from: {file_path}" - self.logger.warning(f"[{self.name}] parse path failed file_path={file_path} resource_dir={resource_dir}") - return {"success": False, "path": file_path, "change": change.name, "answer": self.context.response.answer} - - note_stem = _compute_note_stem(filename) - self.logger.info(f"[{self.name}] {change.name} file_path={file_path} note_stem={note_stem}") - - if self._skip_image_change(file_path): - # Image resources are binary; the text interpretation below would - # read them as mojibake. They are handled by auto_image_step. - answer = f"Skipped image resource file: {file_path}" - self.context.response.success = True - self.context.response.answer = answer - self.context.response.metadata.update( - { - "path": file_path, - "action": "skipped", - "reason": "image_file", - "modified": False, - }, - ) - self.logger.info(f"[{self.name}] skip change file_path={file_path} reason=image_file") - return { - "success": True, - "path": file_path, - "change": change.name, - "answer": answer, - "metadata": dict(self.context.response.metadata), - } - - if change == Change.deleted: - await self._handle_delete(file_path, date_str, note_stem) - else: - await self._handle_upsert( - file_path, - date_str, - note_stem, - change == Change.added, - ) - return { - "success": self.context.response.success, - "path": file_path, - "change": change.name, - "answer": self.context.response.answer, - "metadata": dict(self.context.response.metadata), - } - async def execute(self): assert self.context is not None changes = self.context.get("changes") @@ -562,25 +129,46 @@ class AutoResourceStep(BaseStep): self.logger.warning(f"[{self.name}] invalid changes payload type={type(changes).__name__}") return self.context.response - self.logger.info(f"[{self.name}] start changes={len(changes)}") - results = [] - for index, item in enumerate(changes, start=1): + routes = self._processor_routes() + fallback = next((route for route in routes if route[1].resource_fallback), None) + specific_routes = [route for route in routes if not route[1].resource_fallback] + result_slots: list[dict | None] = [None] * len(changes) + for index, item in enumerate(changes): if not isinstance(item, dict): - self.logger.warning(f"[{self.name}] skip invalid change item index={index} type={type(item).__name__}") + self.logger.warning( + f"[{self.name}] skip invalid change item index={index + 1} type={type(item).__name__}", + ) continue - self.logger.info(f"[{self.name}] process change {index}/{len(changes)}") - results.append( - await self._handle_change(item.get("path") or item.get("file_path", ""), item.get("change", "")), + file_path = item.get("path") or item.get("file_path", "") + target = next( + (route for route in specific_routes if route[1].matches_change(item)), + fallback, ) + if target is None: + result_slots[index] = self._unsupported_result(item, file_path) + continue + target[2].append((index, item)) + + route_counts = ", ".join(f"{spec['backend']}={len(batch)}" for spec, _, batch in routes) + self.logger.info(f"[{self.name}] route changes={len(changes)} processors=({route_counts})") + try: + for spec, _, indexed_changes in routes: + await self._dispatch_processor(spec, indexed_changes, result_slots) + finally: + # dispatch_steps merges the sub-batch into the shared context. + # Downstream steps and the result hook must see the original batch. + self.context["changes"] = changes + + results = [item for item in result_slots if item is not None] success_count = sum(1 for item in results if item.get("success")) - self.context.response.success = success_count == len(changes) + self.context.response.success = len(results) == len(changes) and success_count == len(changes) processed_answer = f"Processed {success_count}/{len(changes)} resource change(s)" self.context.response.answer = _results_answer(results, processed_answer) - self.context.response.metadata["processed"] = len(results) - self.context.response.metadata["results"] = results - self.context.response.metadata["modified"] = any( - bool((item.get("metadata") or {}).get("modified")) for item in results - ) + self.context.response.metadata = { + "processed": len(results), + "results": results, + "modified": any(bool((item.get("metadata") or {}).get("modified")) for item in results), + } await self._emit_result_hook(changes=changes, results=results) self.logger.info( f"[{self.name}] done success={success_count}/{len(changes)} " diff --git a/reme/steps/evolve/auto_text.py b/reme/steps/evolve/auto_text.py new file mode 100644 index 00000000..c89349e7 --- /dev/null +++ b/reme/steps/evolve/auto_text.py @@ -0,0 +1,197 @@ +"""Text resource processor for the unified auto-resource router.""" + +import uuid + +import aiofiles + +from ...components import R +from ..file_io import refresh_day_index +from ._evolve import agent_reply_result_text +from ._auto_resource import BaseAutoResourceStep + + +def _compute_agent_session_id(path: str) -> str: + """Return a stable UUID session id for agent backends.""" + return str(uuid.uuid5(uuid.NAMESPACE_URL, path)) + + +@R.register("auto_text_resource_step") +class AutoTextResourceStep(BaseAutoResourceStep): + """Interpret text resource files into daily notes via an Agent.""" + + resource_fallback = True + router_inherit_keys = BaseAutoResourceStep.router_inherit_keys | frozenset( + {"agent_wrapper", "max_file_bytes", "prompt_dict"}, + ) + + def __init__(self, **kwargs): + super().__init__(**kwargs) + self.create_tools: list[str] = ["write"] + self.update_tools: list[str] = ["read", "edit", "frontmatter_update", "write"] + + async def _handle_upsert( + self, + file_path: str, + date_str: str, + note_stem: str, + added: bool, + ) -> None: + self.logger.info( + f"[{self.name}] upsert start file_path={file_path} date={date_str} " f"note_stem={note_stem} added={added}", + ) + daily_dir = self.config_value("daily_dir") + fallback_path = f"{daily_dir}/{date_str}/{note_stem}.md" + try: + note = await self._list_resource_note(date_str, file_path, fallback_path) + except RuntimeError as exc: + self.context.response.success = False + self.context.response.answer = str(exc) + self.logger.info(f"[{self.name}] list failed file_path={file_path} answer={str(exc)!r}") + return + + note_path = str(note["path"]) if note else fallback_path + note_created = note is None + before_note_path = note_path + before_note_bytes = self._note_bytes(note_path) + self.logger.info(f"[{self.name}] daily note lookup path={note_path} created={note_created}") + + # Read resource file content + abs_path = self.workspace_path / file_path + if not abs_path.is_file(): + self.context.response.success = False + self.context.response.answer = f"Resource file not found: {file_path}" + self.logger.warning(f"[{self.name}] resource missing file_path={file_path}") + return + + skip_read = False + try: + size_bytes = abs_path.stat().st_size + except OSError as exc: + self.context.response.success = False + self.context.response.answer = f"Failed to inspect resource file: {file_path}: {exc}" + self.context.response.metadata.update( + { + "path": file_path, + "action": "failed", + "error": str(exc), + "modified": False, + }, + ) + self.logger.warning(f"[{self.name}] resource stat failed file_path={file_path} error={exc}") + skip_read = True + if not skip_read: + max_file_bytes = self.max_file_bytes() + if size_bytes > max_file_bytes: + self.context.response.success = True + self.context.response.answer = ( + f"Skipped oversized resource file: {file_path} ({size_bytes} > {max_file_bytes} bytes)" + ) + self.context.response.metadata.update( + { + "path": file_path, + "action": "skipped", + "reason": "file_too_large", + "oversized": True, + "size_bytes": size_bytes, + "max_file_bytes": max_file_bytes, + "modified": False, + }, + ) + self.logger.warning( + f"[{self.name}] skip oversized resource file_path={file_path} " + f"size_bytes={size_bytes} max_file_bytes={max_file_bytes}", + ) + skip_read = True + if skip_read: + return + + self.logger.info(f"[{self.name}] read resource start file_path={file_path}") + async with aiofiles.open(abs_path, encoding="utf-8", errors="replace") as f: + file_content = await f.read() + self.logger.info(f"[{self.name}] read resource done file_path={file_path} chars={len(file_content)}") + + template_key = "user_message_create" if note_created else "user_message_update" + user_message = self.prompt_format( + template_key, + workspace_dir=str(self.workspace_path), + note_path=note_path, + note_stem=note_stem, + file_path=file_path, + source_resource=self._source_resource_link(file_path), + file_content=file_content, + date=date_str, + ) + + agent_session_id = _compute_agent_session_id(file_path) + self.logger.info( + f"[{self.name}] agent start file_path={file_path} note_path={note_path} " + f"agent_session_id={agent_session_id}", + ) + result = await self.agent_wrapper.reply( + user_message, + system_prompt=self.prompt_format("system_prompt"), + job_tools=self.create_tools if note_created else self.update_tools, + session_id=agent_session_id, + ) + self.logger.info(f"[{self.name}] agent done file_path={file_path} has_result={bool(result.get('result'))}") + + if note_created: + try: + note = await self._list_resource_note(date_str, file_path, fallback_path) + except RuntimeError as exc: + self.context.response.success = False + self.context.response.answer = str(exc) + self.context.response.metadata.update({"path": None, "created": note_created, "modified": False}) + self.logger.info(f"[{self.name}] post-create list failed file_path={file_path} answer={str(exc)!r}") + return + if note is None: + self.context.response.success = True + self.context.response.answer = agent_reply_result_text(result) + self.context.response.metadata.update({"path": None, "created": False, "modified": False}) + self.logger.info(f"[{self.name}] done without note file_path={file_path} modified=False") + return + note_path = str(note["path"]) + + try: + await self._ensure_resource_frontmatter(note_path, file_path) + note_path = await self._rename_from_frontmatter_name( + note_path, + date_str, + file_path, + note_stem, + fallback_path, + allow_rename=note_created, + ) + except RuntimeError as exc: + self.context.response.success = False + self.context.response.answer = str(exc) + self.context.response.metadata.update( + { + "path": note_path, + "created": note_created, + "modified": self._note_modified(before_note_path, before_note_bytes, note_path), + }, + ) + self.logger.info(f"[{self.name}] post-agent failed path={note_path} answer={str(exc)!r}") + return + + modified = self._note_modified(before_note_path, before_note_bytes, note_path) + self.logger.info(f"[{self.name}] refresh index start date={date_str} daily_dir={daily_dir}") + index_payload = await refresh_day_index(self.file_store, date_str, daily_dir) + self.logger.info(f"[{self.name}] refresh index done date={date_str}") + + self.context.response.success = True + self.context.response.answer = agent_reply_result_text(result) + self.context.response.metadata.update( + { + "path": note_path, + "created": note_created, + "modified": modified, + "session_id": note_stem, + "source_resource": self._source_resource_link(file_path), + "agent_session_id": agent_session_id, + "action": "added" if added else "modified", + "index": index_payload, + }, + ) + self.logger.info(f"[{self.name}] done {note_path} modified={modified}") diff --git a/reme/steps/evolve/auto_resource.yaml b/reme/steps/evolve/auto_text.yaml similarity index 99% rename from reme/steps/evolve/auto_resource.yaml rename to reme/steps/evolve/auto_text.yaml index e9716379..3b859cb6 100644 --- a/reme/steps/evolve/auto_resource.yaml +++ b/reme/steps/evolve/auto_text.yaml @@ -1,3 +1,4 @@ +# AutoTextResourceStep prompts. system_prompt: | You are an automatic resource interpretation system. Your job is to read a resource file and record a structured summary into a daily note at the specified path. Think about what information in this file would be most useful for future retrieval and understanding. diff --git a/tests/integration/test_auto_resource.py b/tests/integration/test_auto_resource.py index d8f50b46..5538b2d3 100644 --- a/tests/integration/test_auto_resource.py +++ b/tests/integration/test_auto_resource.py @@ -25,7 +25,8 @@ sys.path.insert(0, str(INTEGRATION_DIR)) # pylint: disable=wrong-import-position from _workspace_fixture import workspace_env # noqa: E402 -from reme.steps.evolve.auto_resource import _compute_agent_session_id, _compute_note_stem # noqa: E402 +from reme.steps.evolve._auto_resource import _compute_note_stem # noqa: E402 +from reme.steps.evolve.auto_text import _compute_agent_session_id # noqa: E402 RESOURCE_FILENAME = "project-roadmap.md" RESOURCE_CONTENT_V1 = """\ diff --git a/tests/unit/test_auto_image_steps.py b/tests/unit/test_auto_image_steps.py index 6a5c5fa5..6c32f26c 100644 --- a/tests/unit/test_auto_image_steps.py +++ b/tests/unit/test_auto_image_steps.py @@ -1,4 +1,4 @@ -"""Tests for AutoImageStep: image resource files become caption daily notes. +"""Tests for AutoImageResourceStep: image resource files become caption daily notes. The vision model boundary is faked (no network); test images are synthesized with PIL inside a temporary workspace. @@ -12,21 +12,34 @@ import hashlib import io import json import os +import subprocess +import sys import tempfile +import tomllib from pathlib import Path from types import SimpleNamespace -from unittest.mock import MagicMock +from unittest.mock import MagicMock, patch import frontmatter import pytest +import yaml from agentscope.model import ChatModelBase from PIL import Image +from reme.components import R from reme.components.agent_wrapper import BaseAgentWrapper from reme.components.file_store import LocalFileStore from reme.components.runtime_context import RuntimeContext -from reme.steps.evolve.auto_image import AutoImageStep, _parse_caption_json +from reme.enumeration import ComponentEnum +from reme.steps.evolve._auto_resource import BaseAutoResourceStep +from reme.steps.evolve.auto_image import ( + AutoImageResourceStep, + _build_image_request_payload, + _normalize_image_bytes, + _parse_caption_json, +) from reme.steps.evolve.auto_resource import AutoResourceStep +from reme.steps.evolve.auto_text import AutoTextResourceStep from reme.steps.file_io import DailyListStep, FrontmatterUpdateStep, MoveStep, WriteStep @@ -53,11 +66,27 @@ class _FakeAgentWrapper(BaseAgentWrapper): super().__init__() self.inputs = "" - async def reply(self, inputs, **kwargs) -> dict: + async def reply(self, inputs, **_kwargs) -> dict: + """Capture the text processor input and return a successful result.""" self.inputs = inputs return {"result": "ok"} +class _FlakyAgentWrapper(BaseAgentWrapper): + """Fail one text item, then succeed so per-resource isolation is observable.""" + + def __init__(self): + super().__init__() + self.calls = 0 + + async def reply(self, _inputs, **_kwargs) -> dict: + """Raise on the first call and return normally on later calls.""" + self.calls += 1 + if self.calls == 1: + raise RuntimeError("text provider unavailable") + return {"result": "recovered"} + + class _FakeVisionModel(ChatModelBase): """Capture VLM calls and return a canned text response (plain-call path).""" @@ -74,6 +103,24 @@ class _FakeVisionModel(ChatModelBase): return SimpleNamespace(content=[{"type": "text", "text": self.text}]) +class _FakeAudioResourceStep(BaseAutoResourceStep): + """Minimal third-modality processor used to verify the router extension contract.""" + + resource_suffixes = frozenset({".wav"}) + + async def _handle_upsert(self, file_path: str, date_str: str, note_stem: str, added: bool) -> None: + self.context.response.success = True + self.context.response.answer = f"Processed audio resource: {file_path}" + self.context.response.metadata.update( + { + "path": f"daily/{date_str}/{note_stem}.md", + "action": "added" if added else "modified", + "processor": "audio", + "modified": True, + }, + ) + + class _FlakyVisionModel(ChatModelBase): """Fail the first plain call, succeed afterwards.""" @@ -207,7 +254,7 @@ def test_auto_image_creates_caption_note(): try: source = _write_binary(cwd / "resource" / "2026-01-01" / "img.png", _png_bytes()) model = _FakeVisionModel(_caption_json("red-square", "A red square", "An 8x8 solid red square.")) - step = AutoImageStep(app_context=app_ctx, file_store=fs, as_llm=model) + step = AutoImageResourceStep(app_context=app_ctx, file_store=fs, as_llm=model) resp = await _run_step(step, [{"change": "added", "path": str(source)}]) assert resp.success is True @@ -241,14 +288,14 @@ def test_auto_image_parses_fenced_json_and_falls_back_to_raw_text(): try: fenced = "```json\n" + _caption_json("fenced-note", "Fenced", "Fenced caption body.") + "\n```" source = _write_binary(cwd / "resource" / "2026-01-01" / "fenced.png", _png_bytes()) - step = AutoImageStep(app_context=app_ctx, file_store=fs, as_llm=_FakeVisionModel(fenced)) + step = AutoImageResourceStep(app_context=app_ctx, file_store=fs, as_llm=_FakeVisionModel(fenced)) resp = await _run_step(step, [{"change": "added", "path": str(source)}]) assert resp.success is True fenced_post = frontmatter.loads((cwd / "daily" / "2026-01-01" / "fenced-note.md").read_text("utf-8")) assert "Fenced caption body." in fenced_post.content raw = _write_binary(cwd / "resource" / "2026-01-01" / "photo.png", _png_bytes(color=(30, 30, 200))) - step = AutoImageStep( + step = AutoImageResourceStep( app_context=app_ctx, file_store=fs, as_llm=_FakeVisionModel("A plain description."), @@ -280,7 +327,7 @@ def test_auto_image_updates_existing_note_in_place(): "[[resource/2026-01-01/img.png]]", ) model = _FakeVisionModel(_caption_json("red-square", "Updated", "The updated caption.")) - step = AutoImageStep(app_context=app_ctx, file_store=fs, as_llm=model) + step = AutoImageResourceStep(app_context=app_ctx, file_store=fs, as_llm=model) resp = await _run_step(step, [{"change": "modified", "path": str(source)}]) assert resp.success is True @@ -310,7 +357,7 @@ def test_auto_image_deletes_linked_note(): cwd / "daily" / "2026-01-01" / "red-square.md", "[[resource/2026-01-01/img.png]]", ) - step = AutoImageStep(app_context=app_ctx, file_store=fs, as_llm=_FakeVisionModel("{}")) + step = AutoImageResourceStep(app_context=app_ctx, file_store=fs, as_llm=_FakeVisionModel("{}")) resp = await _run_step(step, [{"change": "deleted", "path": str(source)}]) assert resp.success is True @@ -338,7 +385,7 @@ def test_auto_image_downscales_oversized_image_for_request_only(): ) stored_bytes = source.read_bytes() model = _FakeVisionModel(_caption_json("huge-image", "Big", "A big image.")) - step = AutoImageStep(app_context=app_ctx, file_store=fs, as_llm=model) + step = AutoImageResourceStep(app_context=app_ctx, file_store=fs, as_llm=model) resp = await _run_step(step, [{"change": "added", "path": str(source)}]) assert resp.success is True @@ -368,7 +415,7 @@ def test_auto_image_skips_oversized_file(): try: source = _write_binary(cwd / "resource" / "2026-01-01" / "img.png", _png_bytes()) model = _FakeVisionModel(_caption_json("x", "y", "z")) - step = AutoImageStep(app_context=app_ctx, file_store=fs, as_llm=model) + step = AutoImageResourceStep(app_context=app_ctx, file_store=fs, as_llm=model) resp = await _run_step(step, [{"change": "added", "path": str(source)}], max_image_bytes=8) result = resp.metadata["results"][0] @@ -396,7 +443,7 @@ def test_auto_image_skips_without_vision_model(): _install_file_jobs(app_ctx, fs) try: source = _write_binary(cwd / "resource" / "2026-01-01" / "img.png", _png_bytes()) - step = AutoImageStep(app_context=app_ctx, file_store=fs) + step = AutoImageResourceStep(app_context=app_ctx, file_store=fs) resp = await _run_step(step, [{"change": "added", "path": str(source)}]) result = resp.metadata["results"][0] @@ -409,8 +456,381 @@ def test_auto_image_skips_without_vision_model(): asyncio.run(run()) -def test_image_and_text_changes_are_routed_by_suffix(): - """auto_image skips text changes; auto_resource skips image changes.""" +def test_auto_resource_router_preserves_mixed_result_order_and_emits_one_hook(): + """The unified router sends each suffix to one processor and aggregates once.""" + + async def run(): + with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): + cwd = Path.cwd() + app_ctx = _make_app_context(cwd) + app_ctx.registry = R.copy() + app_ctx.registry.add("fake_audio_resource_step", _FakeAudioResourceStep, owner=__name__) + fs = LocalFileStore(name="test_store", embedding_store="") + await fs.start() + _install_file_jobs(app_ctx, fs) + try: + image = _write_binary(cwd / "resource" / "2026-01-01" / "img.png", _png_bytes()) + text = cwd / "resource" / "2026-01-01" / "note.txt" + text.parent.mkdir(parents=True, exist_ok=True) + text.write_text("hello", encoding="utf-8") + wrapper = _FakeAgentWrapper() + model = _FakeVisionModel(_caption_json("red-square", "Red", "A red square.")) + hook_calls = [] + + async def hook(**kwargs): + hook_calls.append(kwargs) + + app_ctx.metadata = {"qwenpaw_memory_result_hook": hook} + changes = [ + {"change": "added", "path": str(image)}, + {"change": "added", "path": str(cwd / "resource" / "2026-01-01" / "clip.wav")}, + {"change": "added", "path": str(text)}, + ] + step = AutoResourceStep( + app_context=app_ctx, + file_store=fs, + agent_wrapper=wrapper, + as_llm=model, + dispatch_steps=[ + "auto_image_resource_step", + "fake_audio_resource_step", + "auto_text_resource_step", + ], + ) + context = RuntimeContext(changes=changes) + resp = await step(context) + + assert resp.success is True + assert [item["path"] for item in resp.metadata["results"]] == [ + "resource/2026-01-01/img.png", + "resource/2026-01-01/clip.wav", + "resource/2026-01-01/note.txt", + ] + assert all( + (item.get("metadata") or {}).get("reason") not in {"image_file", "non_image_file"} + for item in resp.metadata["results"] + ) + assert len(model.calls) == 1 + assert "hello" in wrapper.inputs + assert context.get("changes") == changes + assert len(hook_calls) == 1 + assert hook_calls[0]["kwargs"] == {"changes": changes} + assert len(hook_calls[0]["metadata"]["results"]) == 3 + finally: + await fs.close() + + asyncio.run(run()) + + +def test_auto_resource_router_does_not_mask_text_failure_with_image_success(): + """A later successful image result cannot overwrite an earlier text failure.""" + + async def run(): + with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): + cwd = Path.cwd() + app_ctx = _make_app_context(cwd) + app_ctx.registry = R + fs = LocalFileStore(name="test_store", embedding_store="") + await fs.start() + _install_file_jobs(app_ctx, fs) + try: + missing_text = cwd / "resource" / "2026-01-01" / "missing.txt" + image = _write_binary(cwd / "resource" / "2026-01-01" / "img.png", _png_bytes()) + model = _FakeVisionModel(_caption_json("red-square", "Red", "A red square.")) + changes = [ + {"change": "added", "path": str(missing_text)}, + {"change": "added", "path": str(image)}, + ] + step = AutoResourceStep( + app_context=app_ctx, + dispatch_steps=[ + {"backend": "auto_image_resource_step", "file_store": fs, "as_llm": model}, + { + "backend": "auto_text_resource_step", + "file_store": fs, + "agent_wrapper": _FakeAgentWrapper(), + }, + ], + ) + resp = await step(RuntimeContext(changes=changes)) + + assert resp.success is False + assert resp.metadata["results"][0]["success"] is False + assert "Resource file not found" in resp.metadata["results"][0]["answer"] + assert resp.metadata["results"][1]["success"] is True + assert (cwd / "daily" / "2026-01-01" / "red-square.md").is_file() + finally: + await fs.close() + + asyncio.run(run()) + + +def test_auto_resource_router_isolates_text_exception_and_preserves_image_result(): + """One text exception cannot discard an earlier image write or stop later resources.""" + + async def run(): + with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): + cwd = Path.cwd() + app_ctx = _make_app_context(cwd) + app_ctx.registry = R.copy() + fs = LocalFileStore(name="test_store", embedding_store="") + await fs.start() + _install_file_jobs(app_ctx, fs) + try: + image = _write_binary(cwd / "resource" / "2026-01-01" / "img.png", _png_bytes()) + first_text = cwd / "resource" / "2026-01-01" / "first.txt" + second_text = cwd / "resource" / "2026-01-01" / "second.txt" + first_text.write_text("first", encoding="utf-8") + second_text.write_text("second", encoding="utf-8") + model = _FakeVisionModel(_caption_json("red-square", "Red", "A red square.")) + wrapper = _FlakyAgentWrapper() + hook_calls = [] + + async def hook(**kwargs): + hook_calls.append(kwargs) + + app_ctx.metadata = {"qwenpaw_memory_result_hook": hook} + changes = [ + {"change": "added", "path": str(first_text)}, + {"change": "added", "path": str(image)}, + {"change": "added", "path": str(second_text)}, + ] + context = RuntimeContext(changes=changes) + step = AutoResourceStep( + app_context=app_ctx, + dispatch_steps=[ + {"backend": "auto_image_resource_step", "file_store": fs, "as_llm": model}, + { + "backend": "auto_text_resource_step", + "file_store": fs, + "agent_wrapper": wrapper, + }, + ], + ) + response = await step(context) + + results = response.metadata["results"] + assert response.success is False + assert response.metadata["processed"] == 3 + assert response.metadata["modified"] is True + assert [item["path"] for item in results] == [ + "resource/2026-01-01/first.txt", + "resource/2026-01-01/img.png", + "resource/2026-01-01/second.txt", + ] + assert [item["success"] for item in results] == [False, True, True] + assert results[0]["metadata"] == { + "path": "resource/2026-01-01/first.txt", + "modified": False, + "action": "failed", + "error": "text provider unavailable", + } + assert results[1]["metadata"]["modified"] is True + assert "error" not in results[2]["metadata"] + assert wrapper.calls == 2 + assert (cwd / "daily" / "2026-01-01" / "red-square.md").is_file() + assert context.get("changes") == changes + assert len(hook_calls) == 1 + assert hook_calls[0]["metadata"]["results"] == results + finally: + await fs.close() + + asyncio.run(run()) + + +def test_auto_resource_router_inherits_declared_options_with_child_override(): + """Each processor selects inherited router options; explicit child values win.""" + file_store = object() + agent_wrapper = object() + vision_model = object() + prompt_dict = {"system_prompt": "legacy text prompt"} + step = AutoResourceStep( + file_store=file_store, + agent_wrapper=agent_wrapper, + as_llm=vision_model, + language="zh", + prompt_dict=prompt_dict, + max_file_bytes=4, + max_image_bytes=8, + dispatch_steps=[ + {"backend": "auto_image_resource_step", "max_image_bytes": 32}, + {"backend": "auto_text_resource_step", "max_file_bytes": 16}, + ], + ) + + specs = {spec["backend"]: spec for spec, _, _ in step._processor_routes()} + assert specs["auto_text_resource_step"] == { + "backend": "auto_text_resource_step", + "file_store": file_store, + "agent_wrapper": agent_wrapper, + "language": "zh", + "prompt_dict": prompt_dict, + "max_file_bytes": 16, + } + assert specs["auto_image_resource_step"] == { + "backend": "auto_image_resource_step", + "file_store": file_store, + "as_llm": vision_model, + "language": "zh", + "max_image_bytes": 32, + } + + +def test_auto_resource_router_accepts_a_registered_third_modality_without_code_changes(): + """A new processor only needs registration, a matcher, and dispatch configuration.""" + + async def run(): + with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): + cwd = Path.cwd() + app_ctx = _make_app_context(cwd) + app_ctx.registry = R.copy() + app_ctx.registry.add("fake_audio_resource_step", _FakeAudioResourceStep, owner=__name__) + step = AutoResourceStep( + app_context=app_ctx, + dispatch_steps=["fake_audio_resource_step", "auto_text_resource_step"], + ) + response = await step( + RuntimeContext( + changes=[{"change": "added", "path": str(cwd / "resource" / "2026-01-01" / "clip.WAV")}], + ), + ) + + assert response.success is True + assert response.metadata["results"][0]["metadata"]["processor"] == "audio" + assert response.metadata["results"][0]["path"] == "resource/2026-01-01/clip.WAV" + + unsupported = AutoResourceStep( + app_context=app_ctx, + dispatch_steps=["fake_audio_resource_step"], + ) + response = await unsupported( + RuntimeContext( + changes=[{"change": "added", "path": "resource/2026-01-01/archive.bin"}], + ), + ) + + assert response.success is False + assert response.metadata["results"][0]["metadata"]["reason"] == "unsupported_resource" + + asyncio.run(run()) + + +def test_auto_resource_router_requires_the_fallback_processor_to_be_last(): + """Processor ordering stays deterministic and first-match routing remains extensible.""" + step = AutoResourceStep( + dispatch_steps=["auto_text_resource_step", "auto_image_resource_step"], + ) + + with pytest.raises(ValueError, match="fallback processor must be last"): + step._processor_routes() + + +def test_resource_processors_have_canonical_registrations_and_isolated_prompts(): + """Each modality owns one backend and loads only its module-local prompts.""" + assert R.get(ComponentEnum.STEP, "auto_resource_step") is AutoResourceStep + assert R.get(ComponentEnum.STEP, "auto_text_resource_step") is AutoTextResourceStep + assert R.get(ComponentEnum.STEP, "auto_image_resource_step") is AutoImageResourceStep + assert R.get(ComponentEnum.STEP, "auto_image_step") is None + + text_step = AutoTextResourceStep() + image_step = AutoImageResourceStep() + assert text_step.prompt.has_prompt("system_prompt") + assert text_step.prompt.has_prompt("user_message_create") + assert not text_step.prompt.has_prompt("user_message") + assert image_step.prompt.has_prompt("user_message") + assert not image_step.prompt.has_prompt("system_prompt") + assert not image_step.prompt.has_prompt("user_message_create") + + +def test_auto_image_named_model_uses_standard_ref_resolution(): + """A configured ``as_llm`` component name is honored instead of ignored.""" + app_ctx = _make_app_context(Path.cwd()) + named = _FakeVisionModel("named") + vision = _FakeVisionModel("vision") + default = _FakeVisionModel("default") + app_ctx.components = { + ComponentEnum.AS_LLM: { + "my_vlm": SimpleNamespace(model=named), + "vision": SimpleNamespace(model=vision), + "default": SimpleNamespace(model=default), + }, + } + step = AutoImageResourceStep(app_context=app_ctx, as_llm="my_vlm") + step.context = RuntimeContext() + + assert step._vision_model() is named + + implicit = AutoImageResourceStep(app_context=app_ctx) + implicit.context = RuntimeContext() + assert implicit._vision_model() is vision + + missing = AutoImageResourceStep(app_context=app_ctx, as_llm="missing_vlm") + missing.context = RuntimeContext() + with pytest.raises(KeyError, match="missing_vlm"): + missing._vision_model() + + +def test_image_preprocessing_reports_dependency_and_decode_errors(): + """Lazy image dependencies and corrupt bytes produce actionable errors.""" + real_import = __import__ + + def import_without_pillow(name, *args, **kwargs): + if name == "PIL": + raise ImportError("blocked Pillow") + return real_import(name, *args, **kwargs) + + with patch("builtins.__import__", side_effect=import_without_pillow): + with pytest.raises(RuntimeError, match=r"Pillow.*reme-ai\[core\]"): + _normalize_image_bytes(_png_bytes(), ".png") + + def import_without_heif(name, *args, **kwargs): + if name == "pillow_heif": + raise ImportError("blocked pillow-heif") + return real_import(name, *args, **kwargs) + + with patch("builtins.__import__", side_effect=import_without_heif): + assert _normalize_image_bytes(_png_bytes(), ".png") is None + with pytest.raises(RuntimeError, match=r"pillow-heif.*reme-ai\[image-heif\]"): + _normalize_image_bytes(_png_bytes(), ".heic") + + with pytest.raises(RuntimeError, match="Failed to decode image"): + _build_image_request_payload(b"not an image", ".png") + + bmp_data = _img_bytes("BMP") + with patch("PIL.Image.Image.save", side_effect=OSError("encoder failed")): + with pytest.raises(RuntimeError, match="Failed to convert/resize image"): + _normalize_image_bytes(bmp_data, ".bmp") + + +def test_auto_image_module_imports_without_pillow(): + """Importing the registered image Step does not eagerly require Pillow.""" + root = Path(__file__).resolve().parents[2] + script = """ +import builtins + +real_import = builtins.__import__ + +def import_without_pillow(name, *args, **kwargs): + if name == "PIL" or name.startswith("PIL."): + raise ImportError("Pillow unavailable") + return real_import(name, *args, **kwargs) + +builtins.__import__ = import_without_pillow +import reme.steps.evolve.auto_image +""" + completed = subprocess.run( + [sys.executable, "-c", script], + cwd=root, + check=False, + capture_output=True, + text=True, + ) + + assert completed.returncode == 0, completed.stderr + + +def test_auto_image_prompt_treats_filename_as_a_weak_hint(): + """The VLM prompt separates filename hints from visible image evidence.""" async def run(): with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): @@ -420,27 +840,88 @@ def test_image_and_text_changes_are_routed_by_suffix(): await fs.start() _install_file_jobs(app_ctx, fs) try: - image_step = AutoImageStep(app_context=app_ctx, file_store=fs, as_llm=_FakeVisionModel("{}")) - resp = await _run_step( - image_step, - [{"change": "added", "path": str(cwd / "resource" / "2026-01-01" / "note.txt")}], - ) - assert resp.metadata["results"][0]["metadata"]["reason"] == "non_image_file" + source = _write_binary(cwd / "resource" / "2026-01-01" / "cat-at-beach.png", _png_bytes()) + model = _FakeVisionModel(_caption_json("red-square", "Red", "A red square.")) + step = AutoImageResourceStep(app_context=app_ctx, file_store=fs, as_llm=model) + resp = await _run_step(step, [{"change": "added", "path": str(source)}]) - wrapper = _FakeAgentWrapper() - resource_step = AutoResourceStep(app_context=app_ctx, file_store=fs, agent_wrapper=wrapper) - resp = await _run_step( - resource_step, - [{"change": "added", "path": str(cwd / "resource" / "2026-01-01" / "img.png")}], - ) - assert resp.metadata["results"][0]["metadata"]["reason"] == "image_file" - assert wrapper.inputs == "" + assert resp.success is True + prompt = model.calls[0][0].content[0].text + assert "Filename: cat-at-beach.png" in prompt + assert "Filename stem: cat-at-beach" in prompt + assert "weak hints" in prompt + assert "trust the visible image content" in prompt finally: await fs.close() asyncio.run(run()) +def test_auto_image_reports_modified_when_index_refresh_fails_after_write(): + """A post-write failure keeps the actual on-disk modification state.""" + + async def run(): + with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): + cwd = Path.cwd() + app_ctx = _make_app_context(cwd) + fs = LocalFileStore(name="test_store", embedding_store="") + await fs.start() + _install_file_jobs(app_ctx, fs) + try: + source = _write_binary(cwd / "resource" / "2026-01-01" / "img.png", _png_bytes()) + model = _FakeVisionModel(_caption_json("red-square", "Red", "A red square.")) + step = AutoImageResourceStep(app_context=app_ctx, file_store=fs, as_llm=model) + + async def fail_refresh(*_args, **_kwargs): + raise RuntimeError("index refresh failed") + + with patch("reme.steps.evolve.auto_image.refresh_day_index", new=fail_refresh): + resp = await _run_step(step, [{"change": "added", "path": str(source)}]) + + result = resp.metadata["results"][0] + assert resp.success is False + assert result["metadata"]["action"] == "failed" + assert result["metadata"]["modified"] is True + assert "index refresh failed" in result["metadata"]["error"] + assert (cwd / "daily" / "2026-01-01" / "red-square.md").is_file() + finally: + await fs.close() + + asyncio.run(run()) + + +def test_default_resource_watcher_dispatches_only_the_unified_router(): + """Both init and live resource producers call one auto-resource router.""" + root = Path(__file__).resolve().parents[2] + config = yaml.safe_load((root / "reme" / "config" / "default.yaml").read_text(encoding="utf-8")) + steps = config["jobs"]["resource_watch_loop"]["steps"] + + for producer in steps: + backends = [item["backend"] for item in producer["dispatch_steps"]] + assert backends == ["update_catalog_step", "auto_resource_step"] + router = producer["dispatch_steps"][1] + assert router["dispatch_steps"] == ["auto_image_resource_step", "auto_text_resource_step"] + + auto_resource = config["jobs"]["auto_resource"]["steps"][0] + assert auto_resource["backend"] == "auto_resource_step" + assert auto_resource["dispatch_steps"] == ["auto_image_resource_step", "auto_text_resource_step"] + assert config["jobs"]["auto_image"]["steps"] == [{"backend": "auto_image_resource_step"}] + + +def test_image_dependencies_keep_heif_support_optional(): + """Pillow is core, while pillow-heif stays isolated in its opt-in extra.""" + root = Path(__file__).resolve().parents[2] + project = tomllib.loads((root / "pyproject.toml").read_text(encoding="utf-8"))["project"] + base = [item.lower() for item in project["dependencies"]] + optional = project["optional-dependencies"] + + assert not any(item.startswith("pillow") for item in base) + assert any(item.lower().startswith("pillow>") for item in optional["core"]) + assert not any(item.lower().startswith("pillow-heif") for item in optional["core"]) + assert any(item.lower().startswith("pillow-heif") for item in optional["image-heif"]) + assert "reme-ai[image-heif]" in optional["full"] + + def test_auto_image_model_failure_is_isolated_per_change(): """A failing VLM call marks one change failed while the rest of the batch continues.""" @@ -455,7 +936,7 @@ def test_auto_image_model_failure_is_isolated_per_change(): first = _write_binary(cwd / "resource" / "2026-01-01" / "img-a.png", _png_bytes()) second = _write_binary(cwd / "resource" / "2026-01-01" / "img-b.png", _png_bytes(color=(20, 90, 200))) model = _FlakyVisionModel(_caption_json("blue-square", "Blue", "A blue square.")) - step = AutoImageStep(app_context=app_ctx, file_store=fs, as_llm=model) + step = AutoImageResourceStep(app_context=app_ctx, file_store=fs, as_llm=model) resp = await _run_step( step, [ @@ -477,6 +958,41 @@ def test_auto_image_model_failure_is_isolated_per_change(): asyncio.run(run()) +def test_auto_image_decode_failure_is_isolated_per_change(): + """Corrupt image bytes fail one item without blocking a later valid image.""" + + async def run(): + with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): + cwd = Path.cwd() + app_ctx = _make_app_context(cwd) + fs = LocalFileStore(name="test_store", embedding_store="") + await fs.start() + _install_file_jobs(app_ctx, fs) + try: + corrupt = _write_binary(cwd / "resource" / "2026-01-01" / "bad.png", b"not an image") + valid = _write_binary(cwd / "resource" / "2026-01-01" / "good.png", _png_bytes()) + model = _FakeVisionModel(_caption_json("good-image", "Good", "A valid image.")) + step = AutoImageResourceStep(app_context=app_ctx, file_store=fs, as_llm=model) + resp = await _run_step( + step, + [ + {"change": "added", "path": str(corrupt)}, + {"change": "added", "path": str(valid)}, + ], + ) + + assert resp.success is False + assert resp.metadata["results"][0]["metadata"]["action"] == "failed" + assert "Failed to decode image" in resp.metadata["results"][0]["metadata"]["error"] + assert resp.metadata["results"][1]["success"] is True + assert len(model.calls) == 1 + assert (cwd / "daily" / "2026-01-01" / "good-image.md").is_file() + finally: + await fs.close() + + asyncio.run(run()) + + def test_auto_image_uniquifies_conflicting_note_name(): """A name collision with an unrelated note falls back to the sha1-suffixed path.""" @@ -491,7 +1007,7 @@ def test_auto_image_uniquifies_conflicting_note_name(): source = _write_binary(cwd / "resource" / "2026-01-01" / "img.png", _png_bytes()) _write_note(cwd / "daily" / "2026-01-01" / "red-square.md", "[[resource/2026-01-01/other.png]]") model = _FakeVisionModel(_caption_json("red-square", "A red square", "An 8x8 solid red square.")) - step = AutoImageStep(app_context=app_ctx, file_store=fs, as_llm=model) + step = AutoImageResourceStep(app_context=app_ctx, file_store=fs, as_llm=model) resp = await _run_step(step, [{"change": "added", "path": str(source)}]) assert resp.success is True @@ -540,7 +1056,7 @@ def test_auto_image_note_body_stays_clean_when_caption_field_missing(): try: source = _write_binary(cwd / "resource" / "2026-01-01" / "img.png", _png_bytes()) model = _FakeVisionModel('{"file": "resource/2026-01-01/img.png", "description": "A tall waterfall."}') - step = AutoImageStep(app_context=app_ctx, file_store=fs, as_llm=model) + step = AutoImageResourceStep(app_context=app_ctx, file_store=fs, as_llm=model) resp = await _run_step(step, [{"change": "added", "path": str(source)}]) assert resp.success is True @@ -571,7 +1087,7 @@ def test_auto_image_uses_structured_output_first(): model = _StructuredVisionModel( content={"name": "red-square", "description": "A red square.", "caption": "An 8x8 red square."}, ) - step = AutoImageStep(app_context=app_ctx, file_store=fs, as_llm=model) + step = AutoImageResourceStep(app_context=app_ctx, file_store=fs, as_llm=model) resp = await _run_step(step, [{"change": "added", "path": str(source)}]) assert resp.success is True @@ -603,7 +1119,7 @@ def test_auto_image_retries_with_plain_call_when_structured_fails(): error=RuntimeError("provider rejects tool_choice"), plain_text=_caption_json("plain-note", "Plain", "Plain-call caption."), ) - step = AutoImageStep(app_context=app_ctx, file_store=fs, as_llm=model) + step = AutoImageResourceStep(app_context=app_ctx, file_store=fs, as_llm=model) resp = await _run_step(step, [{"change": "added", "path": str(source)}]) assert resp.success is True @@ -634,7 +1150,7 @@ def test_auto_image_falls_back_when_structured_content_empty(): content={"name": "", "description": "", "caption": ""}, plain_text=_caption_json("empty-note", "Empty", "Recovered by plain call."), ) - step = AutoImageStep(app_context=app_ctx, file_store=fs, as_llm=model) + step = AutoImageResourceStep(app_context=app_ctx, file_store=fs, as_llm=model) resp = await _run_step(step, [{"change": "added", "path": str(source)}]) assert resp.success is True @@ -647,11 +1163,10 @@ def test_auto_image_falls_back_when_structured_content_empty(): asyncio.run(run()) -def test_auto_image_converts_bmp_tiff_heic_requests(): - """bmp/tiff/heic resources are re-encoded for the request; notes record the source media_type.""" +def test_auto_image_converts_bmp_tiff_requests(): + """Core BMP/TIFF conversions run without the optional HEIC dependency.""" async def run(): - pytest.importorskip("pillow_heif") with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): cwd = Path.cwd() app_ctx = _make_app_context(cwd) @@ -663,20 +1178,19 @@ def test_auto_image_converts_bmp_tiff_heic_requests(): for stem, fmt, suffix in ( ("photo", "BMP", ".bmp"), ("scan", "TIFF", ".tiff"), - ("phone", "HEIF", ".heic"), ): sources.append(_write_binary(cwd / "resource" / "2026-01-01" / f"{stem}{suffix}", _img_bytes(fmt))) model = _StructuredVisionModel(content={"name": "", "description": "d", "caption": "converted caption"}) - step = AutoImageStep(app_context=app_ctx, file_store=fs, as_llm=model) + step = AutoImageResourceStep(app_context=app_ctx, file_store=fs, as_llm=model) resp = await _run_step(step, [{"change": "added", "path": str(p)} for p in sources]) assert resp.success is True - assert len(model.structured_calls) == 3 + assert len(model.structured_calls) == 2 sent_mimes = set() for call in model.structured_calls: sent_mimes.add(call[0].content[1].source.media_type) assert sent_mimes <= {"image/png", "image/jpeg"} - for stem, expected_mime in (("photo", "image/bmp"), ("scan", "image/tiff"), ("phone", "image/heic")): + for stem, expected_mime in (("photo", "image/bmp"), ("scan", "image/tiff")): note = frontmatter.loads((cwd / "daily" / "2026-01-01" / f"{stem}.md").read_text(encoding="utf-8")) assert note.metadata["media_type"] == expected_mime assert "converted caption" in note.content @@ -684,3 +1198,36 @@ def test_auto_image_converts_bmp_tiff_heic_requests(): await fs.close() asyncio.run(run()) + + +def test_auto_image_converts_heic_request_when_extra_is_installed(): + """HEIC conversion is covered independently when the image-heif extra exists.""" + pytest.importorskip("pillow_heif") + + async def run(): + with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): + cwd = Path.cwd() + app_ctx = _make_app_context(cwd) + fs = LocalFileStore(name="test_store", embedding_store="") + await fs.start() + _install_file_jobs(app_ctx, fs) + try: + source = _write_binary( + cwd / "resource" / "2026-01-01" / "phone.heic", + _img_bytes("HEIF"), + ) + model = _StructuredVisionModel( + content={"name": "", "description": "d", "caption": "converted caption"}, + ) + step = AutoImageResourceStep(app_context=app_ctx, file_store=fs, as_llm=model) + resp = await _run_step(step, [{"change": "added", "path": str(source)}]) + + assert resp.success is True + assert model.structured_calls[0][0].content[1].source.media_type in {"image/png", "image/jpeg"} + note = frontmatter.loads((cwd / "daily" / "2026-01-01" / "phone.md").read_text(encoding="utf-8")) + assert note.metadata["media_type"] == "image/heic" + assert "converted caption" in note.content + finally: + await fs.close() + + asyncio.run(run()) diff --git a/tests/unit/test_background_steps.py b/tests/unit/test_background_steps.py index 342e42b2..2e89eb4e 100644 --- a/tests/unit/test_background_steps.py +++ b/tests/unit/test_background_steps.py @@ -31,7 +31,9 @@ from reme.components.file_store import LocalFileStore from reme.components.runtime_context import RuntimeContext from reme.enumeration import ComponentEnum from reme.steps.evolve.auto_memory import AutoMemoryStep -from reme.steps.evolve.auto_resource import AutoResourceStep, _compute_note_stem +from reme.steps.evolve._auto_resource import _compute_note_stem +from reme.steps.evolve.auto_resource import AutoResourceStep +from reme.steps.evolve.auto_text import AutoTextResourceStep from reme.steps.file_io.daily_list import DailyListStep from reme.steps.file_io.frontmatter_update import FrontmatterUpdateStep from reme.steps.file_io.move import MoveStep @@ -1198,7 +1200,7 @@ def test_auto_resource_batch_deleted_changes(): "---\nname: test\nsource_resource: '[[resource/2026-01-01/file.md]]'\n---\nbody\n", ) - step = AutoResourceStep(app_context=app_ctx, file_store=fs) + step = AutoTextResourceStep(app_context=app_ctx, file_store=fs) ctx = RuntimeContext( changes=[ {"change": "deleted", "path": str(cwd / "resource" / "2026-01-01" / filename)}, @@ -1231,7 +1233,7 @@ def test_auto_resource_skips_oversized_file_before_reading(): _install_file_jobs(app_ctx, fs) try: source = write_file(cwd / "resource" / "2026-01-01" / "large.txt", "too large") - step = AutoResourceStep(app_context=app_ctx, file_store=fs, agent_wrapper=wrapper) + step = AutoTextResourceStep(app_context=app_ctx, file_store=fs, agent_wrapper=wrapper) resp = await step( RuntimeContext( changes=[{"change": "added", "path": str(source)}], @@ -1266,7 +1268,7 @@ def test_auto_resource_batch_keeps_result_metadata_isolated(): large = write_file(cwd / "resource" / "2026-01-01" / "large.txt", "too large") small = write_file(cwd / "resource" / "2026-01-01" / "small.txt", "ok") second_large = write_file(cwd / "resource" / "2026-01-01" / "second-large.txt", "also large") - step = AutoResourceStep(app_context=app_ctx, file_store=fs, agent_wrapper=wrapper) + step = AutoTextResourceStep(app_context=app_ctx, file_store=fs, agent_wrapper=wrapper) resp = await step( RuntimeContext( changes=[ @@ -1318,7 +1320,7 @@ def test_auto_resource_handles_file_removed_before_stat(): raise FileNotFoundError("file disappeared") return original_stat(path, *args, **kwargs) - step = AutoResourceStep(app_context=app_ctx, file_store=fs) + step = AutoTextResourceStep(app_context=app_ctx, file_store=fs) with patch.object(Path, "stat", disappearing_stat): resp = await step( RuntimeContext(changes=[{"change": "added", "path": str(source)}]), @@ -1346,7 +1348,7 @@ def test_auto_resource_accepts_loose_root_resource(): today = datetime.datetime.now().strftime("%Y-%m-%d") captured = {} - step = AutoResourceStep(app_context=app_ctx) + step = AutoTextResourceStep(app_context=app_ctx) async def fake_upsert(file_path, date_str, note_stem, created): captured.update( @@ -1385,7 +1387,7 @@ def test_auto_resource_loose_root_resource_keeps_existing_dated_resource(): source = write_file(workspace / "resource" / "report.txt", "new") captured = {} - step = AutoResourceStep(app_context=app_ctx) + step = AutoTextResourceStep(app_context=app_ctx) async def fake_upsert(file_path, date_str, note_stem, created): captured.update( @@ -1429,7 +1431,7 @@ def test_auto_resource_modified_missing_note_uses_create_tools(): cwd / "daily" / "2026-01-01" / "report.md", "---\nname: resource-summary\nsource_resource: '[[resource/2026-01-01/report.txt]]'\n---\nbody\n", ) - step = AutoResourceStep(app_context=app_ctx, file_store=fs, agent_wrapper=wrapper) + step = AutoTextResourceStep(app_context=app_ctx, file_store=fs, agent_wrapper=wrapper) resp = await step( RuntimeContext(changes=[{"change": "modified", "path": "resource/2026-01-01/report.txt"}]), ) @@ -1466,7 +1468,7 @@ def test_auto_resource_sanitizes_invalid_generated_name(): cwd / "daily" / "2026-01-01" / "report.md", "---\nname: bad/name\nsource_resource: '[[resource/2026-01-01/report.txt]]'\n---\nbody\n", ) - step = AutoResourceStep(app_context=app_ctx, file_store=fs, agent_wrapper=wrapper) + step = AutoTextResourceStep(app_context=app_ctx, file_store=fs, agent_wrapper=wrapper) resp = await step( RuntimeContext(changes=[{"change": "added", "path": "resource/2026-01-01/report.txt"}]), ) @@ -1503,7 +1505,7 @@ def test_auto_resource_uniquifies_conflicting_generated_name(): cwd / "daily" / "2026-01-01" / "report.md", "---\nname: resource-summary\nsource_resource: '[[resource/2026-01-01/report.txt]]'\n---\nbody\n", ) - step = AutoResourceStep(app_context=app_ctx, file_store=fs, agent_wrapper=wrapper) + step = AutoTextResourceStep(app_context=app_ctx, file_store=fs, agent_wrapper=wrapper) resp = await step( RuntimeContext(changes=[{"change": "added", "path": "resource/2026-01-01/report.txt"}]), ) @@ -1540,7 +1542,7 @@ def test_auto_resource_update_finds_renamed_note_by_source_resource(): cwd / "daily" / "2026-01-01" / "generated-name.md", "---\nname: generated-name\nsource_resource: '[[resource/2026-01-01/report.txt]]'\n---\nold body\n", ) - step = AutoResourceStep(app_context=app_ctx, file_store=fs, agent_wrapper=wrapper) + step = AutoTextResourceStep(app_context=app_ctx, file_store=fs, agent_wrapper=wrapper) resp = await step( RuntimeContext(changes=[{"change": "modified", "path": "resource/2026-01-01/report.txt"}]), ) @@ -1583,7 +1585,7 @@ def test_auto_resource_update_keeps_existing_renamed_path(): ) wrapper.on_reply = rewrite_frontmatter - step = AutoResourceStep(app_context=app_ctx, file_store=fs, agent_wrapper=wrapper) + step = AutoTextResourceStep(app_context=app_ctx, file_store=fs, agent_wrapper=wrapper) resp = await step( RuntimeContext(changes=[{"change": "modified", "path": "resource/2026-01-01/report.txt"}]), ) @@ -1619,7 +1621,7 @@ def test_auto_resource_reports_unmodified_when_agent_skips_existing_note(): cwd / "daily" / "2026-01-01" / "report.md", "---\nname: report\nsource_resource: '[[resource/2026-01-01/report.txt]]'\n---\nbody\n", ) - step = AutoResourceStep(app_context=app_ctx, file_store=fs, agent_wrapper=wrapper) + step = AutoTextResourceStep(app_context=app_ctx, file_store=fs, agent_wrapper=wrapper) resp = await step( RuntimeContext(changes=[{"change": "modified", "path": "resource/2026-01-01/report.txt"}]), ) @@ -1648,7 +1650,7 @@ def test_auto_resource_deletes_loose_root_resource_note_for_today(): try: today = datetime.datetime.now().strftime("%Y-%m-%d") note_path = write_file(workspace / "daily" / today / "report.md", "---\nname: report\n---\nbody\n") - step = AutoResourceStep(app_context=app_ctx, file_store=fs) + step = AutoTextResourceStep(app_context=app_ctx, file_store=fs) resp = await step(RuntimeContext(changes=[{"change": "deleted", "path": "resource/report.txt"}])) @@ -1679,7 +1681,7 @@ def test_auto_resource_deletes_renamed_note_by_source_resource(): workspace / "daily" / "2026-01-01" / "generated-name.md", "---\nname: generated-name\nsource_resource: '[[resource/2026-01-01/report.txt]]'\n---\nbody\n", ) - step = AutoResourceStep(app_context=app_ctx, file_store=fs) + step = AutoTextResourceStep(app_context=app_ctx, file_store=fs) resp = await step( RuntimeContext(changes=[{"change": "deleted", "path": "resource/2026-01-01/report.txt"}]),