refactor: split auto resource processors behind router

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

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 的决策流程。                    |

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

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.
 

docs/en/reme_scene.md

@@ -66,9 +66,9 @@ The corresponding flow is:
 - `auto_memory` saves a filtered source conversation record to `session/dialog/<session_id>.jsonl`, then asks the agent to write
   important facts to a topic-named `daily/<date>/<generated_name>.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/<date>.md` after writing.
 
 ### Day 1 evening: Auto Dream writes to Digest

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`

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 统一索引、整合和检索。

docs/zh/reme_scene.md

@@ -61,8 +61,8 @@ daily/
 
 - `auto_memory` 保存对话来源消息到 `session/dialog/<session_id>.jsonl`，再让 Agent 把重要事实写入按主题命名的
   `daily/<date>/<generated_name>.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/<date>.md` 当天索引页。
 
 ### Day 1 晚上：Auto Dream 进入 Digest

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]

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:

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

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

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

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 必须基于图像中的可见内容,不得从文件名推断事实。
+
   ## 记录什么
   - **可见事实**:人物、物体、地点、动作及相互关系——未来对话中可能重要的信息。
   - **图中的文字**:逐字转录有意义的可见文字(幻灯片、白板、截图、招牌、标签);保留原语言,不翻译、不纠错。

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

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

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.
 

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

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

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"}]),