fix: address auto resource review concerns

docs/en/auto_resource.md

@@ -91,9 +91,9 @@ The resource card links to the original file through frontmatter:
 source_resource: "[[resource/2026-06-20/market-report.md]]"
 ```
 
-When a resource changes, Auto Resource finds and updates the corresponding card through `source_resource`. When a
-resource is deleted, its daily note is also removed. The older `daily/YYYY-MM-DD/<resource_stem>.md` naming convention
-remains supported as a fallback.
+When a resource changes, Auto Resource finds and updates the corresponding card through an exact `source_resource`
+match. When a resource is deleted, only the explicitly linked daily note is removed. A same-stem note without that
+provenance marker is treated as user-owned and left untouched; new resource cards use a collision-free path instead.
 
 ## Daily Index
 

docs/zh/auto_resource.md

@@ -78,8 +78,8 @@ daily/2026-06-20/市场报告要点.md
 source_resource: "[[resource/2026-06-20/market-report.md]]"
 ```
 
-如果资源文件更新，Auto Resource 会通过 `source_resource` 找到对应卡片并更新；如果资源文件删除，对应的 daily note 也会被清理。旧版本按
-stem 生成的 `daily/YYYY-MM-DD/<resource_stem>.md` 仍作为 fallback 兼容。
+如果资源文件更新，Auto Resource 只会通过精确匹配的 `source_resource` 找到对应卡片并更新；如果资源文件删除，也只会清理显式关联的
+daily note。缺少该来源标记的同 stem 笔记会被视为用户笔记并保留，新资源卡片则会使用无冲突路径。
 
 ## 当天索引
 

reme/config/default.yaml

@@ -197,30 +197,6 @@ jobs:
           - auto_image_resource_step
           - auto_text_resource_step
 
-  auto_image:
-    backend: base
-    description: "Auto-image: interpret image resource files into daily caption notes"
-    parameters:
-      type: object
-      properties:
-        changes:
-          type: array
-          description: "image resource change batch, each item has path/file_path and change"
-          items:
-            type: object
-            properties:
-              path:
-                type: string
-              file_path:
-                type: string
-              change:
-                type: string
-                description: "added/modified/deleted"
-      required:
-        - changes
-    steps:
-      - backend: auto_image_resource_step
-
   proactive:
     backend: base
     description: "Proactive: read daily/<date>/interests.yaml and expose the latest user-interest topics."

reme/steps/evolve/auto_image_resource.py

@@ -162,6 +162,7 @@ def _parse_caption_json(text: str) -> dict:
     fence = _JSON_FENCE_RE.match(cleaned)
     if fence:
         cleaned = fence.group(1)
+    parsed_json = False
     for candidate in (cleaned, cleaned[cleaned.find("{") : cleaned.rfind("}") + 1]):
         if not candidate:
             continue
@@ -169,11 +170,14 @@ def _parse_caption_json(text: str) -> dict:
             parsed = json.loads(candidate)
         except (json.JSONDecodeError, ValueError):
             continue
+        parsed_json = True
         if isinstance(parsed, dict):
             normalized = _normalize_caption_fields(parsed)
             if normalized["caption"] or normalized["description"]:
                 return normalized
-    return {"name": "", "description": "", "caption": text.strip()}
+    if parsed_json:
+        return {"name": "", "description": "", "caption": ""}
+    return {"name": "", "description": "", "caption": cleaned.strip()}
 
 
 @R.register("auto_image_resource_step")
@@ -234,18 +238,20 @@ class AutoImageResourceStep(BaseAutoResourceStep):
         except Exception as exc:  # pylint: disable=broad-except
             self.logger.warning(f"[{self.name}] structured caption failed ({exc}); retrying with a plain call")
         result = await model([user_message])
-        return _parse_caption_json(await _response_text(result))
+        parsed = _parse_caption_json(await _response_text(result))
+        if not parsed["caption"] and not parsed["description"]:
+            raise RuntimeError("Vision model returned no usable caption")
+        return parsed
 
-    async def _read_image(self, file_path: str) -> dict | None:
+    async def _read_image(self, file_path: str, source_path: Path) -> dict | None:
         """Read the image file and build the VLM request payload.
 
         Returns ``None`` when the change must be skipped (stat failure or
         oversized file); the skip outcome is already recorded on the response.
         """
-        abs_path = self.workspace_path / file_path
         max_image_bytes = self._max_image_bytes()
         try:
-            size_bytes = abs_path.stat().st_size
+            size_bytes = source_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}"
@@ -282,7 +288,7 @@ class AutoImageResourceStep(BaseAutoResourceStep):
             return None
 
         self.logger.info(f"[{self.name}] read image start file_path={file_path}")
-        async with aiofiles.open(abs_path, "rb") as f:
+        async with aiofiles.open(source_path, "rb") as f:
             data = await f.read()
         payload = _build_image_request_payload(data, Path(file_path).suffix.lower())
         self.logger.info(
@@ -291,68 +297,17 @@ class AutoImageResourceStep(BaseAutoResourceStep):
         )
         return payload
 
-    async def _handle_change(self, file_path: str, raw_change) -> dict:
-        """Skip non-image changes; isolate per-change failures so the batch continues."""
-        if file_path and not self.matches_change({"path": file_path}):
-            file_path = self.to_workspace_relative(file_path) if Path(file_path).is_absolute() else file_path
-            self.context.response.metadata = {}
-            answer = f"Skipped non-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": "non_image_file",
-                    "modified": False,
-                },
-            )
-            self.logger.info(f"[{self.name}] skip change file_path={file_path} reason=non_image_file")
-            return {
-                "success": True,
-                "path": file_path,
-                "change": str(raw_change),
-                "answer": answer,
-                "metadata": dict(self.context.response.metadata),
-            }
-        try:
-            return await super()._handle_change(file_path, raw_change)
-        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(
-                {
-                    "action": "failed",
-                    "error": str(exc),
-                },
-            )
-            self.logger.warning(f"[{self.name}] caption failed file_path={file_path} error={exc}")
-            return {
-                "success": False,
-                "path": file_path,
-                "change": str(raw_change),
-                "answer": self.context.response.answer,
-                "metadata": dict(self.context.response.metadata),
-            }
-
-    async def _handle_upsert(self, file_path: str, date_str: str, note_stem: str, added: bool) -> None:
+    async def _handle_upsert(
+        self,
+        file_path: str,
+        date_str: str,
+        note_stem: str,
+        added: bool,
+        source_path: Path,
+    ) -> None:
         """Caption the image and write/refresh its note (image counterpart of the text upsert)."""
-        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)
+        note_state = await self._prepare_resource_note(date_str, file_path, note_stem)
+        note_path = note_state.path
         self.logger.info(
             f"[{self.name}] upsert start file_path={file_path} date={date_str} " f"note_stem={note_stem} added={added}",
         )
@@ -372,7 +327,7 @@ class AutoImageResourceStep(BaseAutoResourceStep):
             self.logger.warning(f"[{self.name}] no vision model configured file_path={file_path}")
             return
 
-        payload = await self._read_image(file_path)
+        payload = await self._read_image(file_path, source_path)
         if payload is None:
             return
 
@@ -419,77 +374,21 @@ class AutoImageResourceStep(BaseAutoResourceStep):
         )
         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),
-            },
+        note_path = await self._finalize_resource_note(
+            note_state,
+            date_str,
+            file_path,
+            note_stem,
+            added,
         )
-
-        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": 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}"
-                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"])
-
-        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-write failed path={note_path} answer={str(exc)!r}")
-            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 self._refresh_day_index(date_str)
+        if note_path is None:
+            raise RuntimeError(f"Image caption note was not written: {file_path}")
 
         self.context.response.success = True
         self.context.response.answer = f"Captioned image resource {file_path} -> {note_path}"
         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),
-                "action": "added" if added else "modified",
                 "media_type": payload["source_mime"],
-                "index": index_payload,
             },
         )
-        self.logger.info(f"[{self.name}] done {note_path} modified={modified}")
+        self.logger.info(f"[{self.name}] done {note_path} modified={self.context.response.metadata['modified']}")

reme/steps/evolve/auto_text_resource.py

@@ -1,6 +1,7 @@
 """Text resource processor for the unified auto-resource router."""
 
 import uuid
+from pathlib import Path
 
 import aiofiles
 
@@ -36,29 +37,18 @@ class AutoTextResourceStep(BaseAutoResourceStep):
         date_str: str,
         note_stem: str,
         added: bool,
+        source_path: Path,
     ) -> 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)
+        note_state = await self._prepare_resource_note(date_str, file_path, note_stem)
+        note_path = note_state.path
+        note_created = note_state.created
         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():
+        if not source_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}")
@@ -66,7 +56,7 @@ class AutoTextResourceStep(BaseAutoResourceStep):
 
         skip_read = False
         try:
-            size_bytes = abs_path.stat().st_size
+            size_bytes = source_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}"
@@ -107,7 +97,7 @@ class AutoTextResourceStep(BaseAutoResourceStep):
             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:
+        async with aiofiles.open(source_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)}")
 
@@ -136,61 +126,24 @@ class AutoTextResourceStep(BaseAutoResourceStep):
         )
         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}")
+        note_path = await self._finalize_resource_note(
+            note_state,
+            date_str,
+            file_path,
+            note_stem,
+            added,
+        )
+        if note_path is None:
+            self.context.response.success = True
+            self.context.response.answer = agent_reply_result_text(result)
+            self.logger.info(f"[{self.name}] done without note file_path={file_path} modified=False")
             return
 
-        modified = self._note_modified(before_note_path, before_note_bytes, note_path)
-        index_payload = await self._refresh_day_index(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}")
+        self.logger.info(f"[{self.name}] done {note_path} modified={self.context.response.metadata['modified']}")

reme/steps/evolve/base_auto_resource.py

@@ -4,6 +4,7 @@ import hashlib
 import re
 from abc import abstractmethod
 from collections.abc import Mapping
+from dataclasses import dataclass
 from pathlib import Path, PurePosixPath
 from typing import Any
 
@@ -12,6 +13,7 @@ from watchfiles import Change
 
 from ..base_step import BaseStep
 from ..file_io import refresh_day_index, validate_filename_component
+from ..file_io._path import is_relative_to, resolve_path
 from ._evolve import now
 
 _SOURCE_RESOURCE_KEY = "source_resource"
@@ -19,6 +21,15 @@ _DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$")
 _UNSAFE_FILENAME_CHARS = re.compile(r'[<>:"/\\|?*\x00-\x1f]+')
 
 
+@dataclass(frozen=True)
+class _ResourceNoteState:
+    """Snapshot a linked note, or a collision-free path reserved for a new one."""
+
+    path: str
+    created: bool
+    before_bytes: bytes | None
+
+
 def _compute_note_stem(filename: str) -> str:
     """Return the daily note stem for a resource filename."""
     return PurePosixPath(filename).stem
@@ -118,6 +129,68 @@ class BaseAutoResourceStep(BaseStep):
     def _daily_note_path(self, day: str, name: str) -> str:
         return f"{self.config_value('daily_dir')}/{day}/{name}.md"
 
+    def _resource_directory(self) -> tuple[str, Path]:
+        """Return the workspace-relative identity and resolved resource root."""
+        workspace = self.workspace_path.resolve()
+        configured = str(self.config_value("resource_dir"))
+        resolved, error = resolve_path(workspace, configured)
+        if error or resolved is None:
+            raise ValueError(f"invalid resource_dir {configured!r}: {error or 'cannot resolve path'}")
+
+        logical = Path(configured)
+        if logical.is_absolute():
+            for workspace_variant in (self.workspace_path.absolute(), workspace):
+                try:
+                    logical = logical.relative_to(workspace_variant)
+                    break
+                except ValueError:
+                    continue
+            else:
+                raise ValueError("resource_dir must stay inside the workspace")
+        return logical.as_posix(), resolved
+
+    def _resolve_resource_source(self, raw_path: str) -> tuple[str, Path, str]:
+        """Return a logical resource path, safe read path, and logical resource root.
+
+        The logical path is kept for ``source_resource`` provenance. The
+        resolved path is used for stat/read so a symlink cannot escape the
+        configured resource directory.
+        """
+        value = str(raw_path or "").strip()
+        if not value:
+            raise ValueError("resource path is required")
+
+        supplied = Path(value)
+        if any(part in {".", ".."} for part in supplied.parts):
+            raise ValueError(f"resource path cannot contain '.' or '..': {value!r}")
+
+        workspace = self.workspace_path.resolve()
+        resource_dir, resource_root = self._resource_directory()
+
+        resolved, path_error = resolve_path(workspace, value)
+        if path_error or resolved is None:
+            raise ValueError(f"invalid resource path {value!r}: {path_error or 'cannot resolve path'}")
+        if not is_relative_to(resolved, resource_root):
+            raise ValueError("resource path must stay inside the configured resource directory")
+
+        if supplied.is_absolute():
+            try:
+                logical = supplied.relative_to(self.workspace_path.absolute())
+            except ValueError:
+                try:
+                    logical = supplied.relative_to(workspace)
+                except ValueError as exc:
+                    raise ValueError("resource path must stay inside the workspace") from exc
+        else:
+            logical = supplied
+
+        resource_logical = Path(resource_dir)
+        try:
+            logical.relative_to(resource_logical)
+        except ValueError as exc:
+            raise ValueError("resource path must stay inside the configured resource directory") from exc
+        return logical.as_posix(), resolved, resource_dir
+
     @staticmethod
     def _source_resource_link(file_path: str) -> str:
         return f"[[{file_path}]]"
@@ -148,22 +221,57 @@ class BaseAutoResourceStep(BaseStep):
         self.logger.info(f"[{self.name}] refresh index done date={day}")
         return index_payload
 
-    def _find_resource_note(self, notes: list[dict], file_path: str, fallback_path: str) -> dict | None:
+    def _find_resource_note(self, notes: list[dict], file_path: str) -> dict | None:
+        """Return only a note explicitly owned by ``file_path``."""
         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:
+    async def _list_resource_note(self, day: str, file_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)
+        return self._find_resource_note(notes, file_path)
+
+    async def _prepare_resource_note(self, day: str, file_path: str, note_stem: str) -> _ResourceNoteState:
+        """Find the owned note or allocate a safe path before the first write."""
+        note = await self._list_resource_note(day, file_path)
+        if note is not None:
+            note_path = str(note["path"])
+            return _ResourceNoteState(path=note_path, created=False, before_bytes=self._note_bytes(note_path))
+
+        _, note_path = self._unique_daily_note_path(day, note_stem, file_path, current_path="")
+        return _ResourceNoteState(path=note_path, created=True, before_bytes=None)
+
+    async def _resolve_written_note(
+        self,
+        state: _ResourceNoteState,
+        day: str,
+        file_path: str,
+    ) -> str | None:
+        """Resolve a processor write without claiming a pre-existing same-stem note."""
+        if not state.created:
+            if self._note_bytes(state.path) is None:
+                raise RuntimeError(f"linked resource note disappeared: {state.path}")
+            return state.path
+
+        note = await self._list_resource_note(day, file_path)
+        if note is not None:
+            return str(note["path"])
+        if self._note_bytes(state.path) is None:
+            return None
+
+        # The path was absent when this invocation allocated it, so a note
+        # created there without provenance can be repaired conservatively. A
+        # different explicit owner is always a conflict and is never claimed.
+        source = str(self._frontmatter(state.path).get(_SOURCE_RESOURCE_KEY, "")).strip()
+        expected = self._source_resource_link(file_path)
+        if source and source != expected:
+            raise RuntimeError(f"resource note path is owned by another source: {state.path}")
+        return state.path
 
     async def _ensure_resource_frontmatter(self, path: str, file_path: str) -> None:
         metadata = {_SOURCE_RESOURCE_KEY: self._source_resource_link(file_path)}
@@ -212,7 +320,6 @@ class BaseAutoResourceStep(BaseStep):
         day: str,
         file_path: str,
         fallback_name: str,
-        fallback_path: str,
         *,
         allow_rename: bool,
     ) -> str:
@@ -220,7 +327,7 @@ class BaseAutoResourceStep(BaseStep):
         current_name = PurePosixPath(path).stem
         suggested_name = str(meta.get("name", "")).strip()
 
-        if not allow_rename and path != fallback_path:
+        if not allow_rename:
             name = _sanitize_note_name(current_name, fallback_name)
             if suggested_name != name:
                 await self._set_frontmatter_name(path, name)
@@ -245,18 +352,73 @@ class BaseAutoResourceStep(BaseStep):
             raise RuntimeError(f"move failed: {move_response.answer}")
         return target_path
 
+    async def _finalize_resource_note(
+        self,
+        state: _ResourceNoteState,
+        day: str,
+        file_path: str,
+        note_stem: str,
+        added: bool,
+    ) -> str | None:
+        """Resolve, source-link, rename, index, and report one processor write."""
+        staged_bytes = self._note_bytes(state.path)
+        self.context.response.metadata.update(
+            {
+                "path": state.path if staged_bytes is not None else None,
+                "created": state.created and staged_bytes is not None,
+                "modified": self._note_modified(state.path, state.before_bytes, state.path),
+            },
+        )
+        note_path = await self._resolve_written_note(state, day, file_path)
+        if note_path is None:
+            self.context.response.metadata.update({"path": None, "created": False, "modified": False})
+            return None
+
+        modified = self._note_modified(state.path, state.before_bytes, note_path)
+        self.context.response.metadata.update({"path": note_path, "created": state.created, "modified": modified})
+        await self._ensure_resource_frontmatter(note_path, file_path)
+        note_path = await self._rename_from_frontmatter_name(
+            note_path,
+            day,
+            file_path,
+            note_stem,
+            allow_rename=state.created,
+        )
+        modified = self._note_modified(state.path, state.before_bytes, note_path)
+        self.context.response.metadata.update({"path": note_path, "created": state.created, "modified": modified})
+        index_payload = await self._refresh_day_index(day)
+        self.context.response.metadata.update(
+            {
+                "path": note_path,
+                "created": state.created,
+                "modified": modified,
+                "session_id": note_stem,
+                "source_resource": self._source_resource_link(file_path),
+                "action": "added" if added else "modified",
+                "index": index_payload,
+            },
+        )
+        return note_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}")
+        note = await self._list_resource_note(date_str, file_path)
+        if note is None:
+            self.context.response.success = True
+            self.context.response.answer = f"No linked resource note to delete: {file_path}"
+            self.context.response.metadata.update(
+                {
+                    "path": None,
+                    "session_id": note_stem,
+                    "source_resource": self._source_resource_link(file_path),
+                    "action": "skipped",
+                    "reason": "resource_note_not_found",
+                    "modified": False,
+                },
+            )
+            self.logger.info(f"[{self.name}] delete skipped; no owned note file_path={file_path}")
             return
 
-        note_rel = str(note["path"]) if note else fallback_path
+        note_rel = str(note["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}")
@@ -265,12 +427,6 @@ class BaseAutoResourceStep(BaseStep):
             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}")
-        index_payload = await self._refresh_day_index(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,
@@ -278,9 +434,15 @@ class BaseAutoResourceStep(BaseStep):
                 "source_resource": self._source_resource_link(file_path),
                 "action": "deleted",
                 "modified": note_existed,
-                "index": index_payload,
             },
         )
+        await self.file_store.delete([note_rel])
+        self.logger.info(f"[{self.name}] catalog delete done note={note_rel}")
+        index_payload = await self._refresh_day_index(date_str)
+
+        self.context.response.success = True
+        self.context.response.answer = f"Deleted resource note: {note_rel}"
+        self.context.response.metadata["index"] = index_payload
 
     @abstractmethod
     async def _handle_upsert(
@@ -289,6 +451,7 @@ class BaseAutoResourceStep(BaseStep):
         date_str: str,
         note_stem: str,
         added: bool,
+        source_path: Path,
     ) -> None:
         """Interpret one added or modified resource into its daily note."""
 
@@ -297,7 +460,6 @@ class BaseAutoResourceStep(BaseStep):
         # 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"
@@ -311,7 +473,23 @@ class BaseAutoResourceStep(BaseStep):
             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")
+        try:
+            file_path, source_path, resource_dir = self._resolve_resource_source(file_path)
+        except ValueError as exc:
+            self.context.response.success = False
+            self.context.response.answer = str(exc)
+            self.context.response.metadata.update(
+                {"path": str(file_path), "action": "failed", "error": str(exc), "modified": False},
+            )
+            self.logger.warning(f"[{self.name}] invalid resource path file_path={file_path!r} error={exc}")
+            return {
+                "success": False,
+                "path": str(file_path),
+                "change": change.name,
+                "answer": self.context.response.answer,
+                "metadata": dict(self.context.response.metadata),
+            }
+
         loose_filename = _loose_resource_filename(file_path, resource_dir)
         if loose_filename:
             date_str, filename = self._today(), loose_filename
@@ -336,6 +514,7 @@ class BaseAutoResourceStep(BaseStep):
                 date_str,
                 note_stem,
                 change == Change.added,
+                source_path,
             )
         return {
             "success": self.context.response.success,

reme/steps/index/_watch_rules.py

@@ -92,6 +92,7 @@ def match_file(file_path: str, rules: list[WatchRule]) -> bool:
 
 def _match_rule(p: Path, rule: WatchRule) -> bool:
     """Check if a single path matches a rule's suffix constraint."""
-    if rule.suffixes and not any(p.name.endswith("." + s.strip(".")) for s in rule.suffixes):
+    filename = p.name.casefold()
+    if rule.suffixes and not any(filename.endswith("." + suffix.strip(".").casefold()) for suffix in rule.suffixes):
         return False
     return True

tests/unit/auto_resource_test_support.py

@@ -0,0 +1,269 @@
+"""Shared test harness for auto-resource processor and router tests."""
+
+import io
+import json
+from dataclasses import dataclass
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+import pytest_asyncio
+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_resource import AutoImageResourceStep
+from reme.steps.evolve.auto_resource import AutoResourceStep
+from reme.steps.evolve.base_auto_resource import BaseAutoResourceStep
+from reme.steps.file_io import DailyListStep, FrontmatterUpdateStep, MoveStep, WriteStep
+
+
+class FakeAgentWrapper(BaseAgentWrapper):
+    """Capture text-processor calls without invoking a real model."""
+
+    def __init__(self):
+        super().__init__()
+        self.inputs = ""
+
+    async def reply(self, inputs, **_kwargs) -> dict:
+        """Record and accept one text-processor request."""
+        self.inputs = inputs
+        return {"result": "ok"}
+
+
+class FlakyAgentWrapper(BaseAgentWrapper):
+    """Fail one text item, then succeed."""
+
+    def __init__(self):
+        super().__init__()
+        self.calls = 0
+
+    async def reply(self, _inputs, **_kwargs) -> dict:
+        """Fail the first request and accept subsequent ones."""
+        self.calls += 1
+        if self.calls == 1:
+            raise RuntimeError("text provider unavailable")
+        return {"result": "recovered"}
+
+
+class FakeVisionModel(ChatModelBase):
+    """Capture VLM calls and return canned plain text."""
+
+    def __init__(self, text: str):
+        self.text = text
+        self.calls: list = []
+
+    async def generate_structured_output(self, messages, structured_model, **kwargs):  # pylint: disable=unused-argument
+        """Force callers through the plain fallback."""
+        raise NotImplementedError("structured path not faked")
+
+    async def __call__(self, messages, **kwargs):
+        """Record a call and return the canned plain text."""
+        self.calls.append(messages)
+        return SimpleNamespace(content=[{"type": "text", "text": self.text}])
+
+
+class FlakyVisionModel(ChatModelBase):
+    """Fail the first plain call, then succeed."""
+
+    def __init__(self, text: str):
+        self.text = text
+        self.calls = 0
+
+    async def generate_structured_output(self, messages, structured_model, **kwargs):  # pylint: disable=unused-argument
+        """Force callers through the plain fallback."""
+        raise NotImplementedError("structured path not faked")
+
+    async def __call__(self, messages, **kwargs):
+        """Fail once, then return the canned plain text."""
+        self.calls += 1
+        if self.calls == 1:
+            raise RuntimeError("vision backend unavailable")
+        return SimpleNamespace(content=[{"type": "text", "text": self.text}])
+
+
+class StructuredVisionModel(ChatModelBase):
+    """Serve structured output and count fallback plain calls."""
+
+    def __init__(self, content: dict | None = None, error: Exception | None = None, plain_text: str = "plain"):
+        self.content = content
+        self.error = error
+        self.plain_text = plain_text
+        self.structured_calls: list = []
+        self.plain_calls: list = []
+
+    async def generate_structured_output(self, messages, structured_model, **kwargs):  # pylint: disable=unused-argument
+        """Return or fail with the configured structured response."""
+        self.structured_calls.append(messages)
+        if self.error is not None:
+            raise self.error
+        return SimpleNamespace(content=dict(self.content or {}))
+
+    async def __call__(self, messages, **kwargs):  # pylint: disable=unused-argument
+        """Record and return the configured plain fallback."""
+        self.plain_calls.append(messages)
+        return SimpleNamespace(content=[{"type": "text", "text": self.plain_text}])
+
+
+class FakeAudioResourceStep(BaseAutoResourceStep):
+    """Minimal third modality 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,
+        source_path: Path,
+    ) -> None:
+        del source_path
+        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 _StepJob:
+    """Tiny job adapter for tests that need ``BaseStep.run_job``."""
+
+    def __init__(self, step_cls, app_context, file_store):
+        self.step_cls = step_cls
+        self.app_context = app_context
+        self.file_store = file_store
+
+    async def __call__(self, **kwargs):
+        step = self.step_cls(app_context=self.app_context, file_store=self.file_store)
+        result = await step(**kwargs)
+        return result or step.context.response
+
+
+def make_app_context(workspace: Path):
+    """Create the minimal application context used by resource tests."""
+    context = MagicMock()
+    context.app_config.workspace_dir = str(workspace)
+    context.app_config.daily_dir = "daily"
+    context.app_config.digest_dir = "digest"
+    context.app_config.resource_dir = "resource"
+    context.app_config.session_dir = "session"
+    context.app_config.timezone = None
+    return context
+
+
+def _install_file_jobs(app_context, file_store) -> None:
+    app_context.jobs = {
+        "daily_list": _StepJob(DailyListStep, app_context, file_store),
+        "frontmatter_update": _StepJob(FrontmatterUpdateStep, app_context, file_store),
+        "move": _StepJob(MoveStep, app_context, file_store),
+        "write": _StepJob(WriteStep, app_context, file_store),
+    }
+
+
+def image_bytes(image_format: str = "PNG", size=(8, 8), color=(200, 30, 30)) -> bytes:
+    """Synthesize a small image in a Pillow-supported format."""
+    if image_format == "HEIF":
+        from pillow_heif import register_heif_opener
+
+        register_heif_opener()
+    image = Image.new("RGB", size, color)
+    buffer = io.BytesIO()
+    image.save(buffer, format=image_format)
+    return buffer.getvalue()
+
+
+def png_bytes(width: int = 8, height: int = 8, color=(200, 30, 30)) -> bytes:
+    """Compatibility shorthand for PNG-focused assertions."""
+    return image_bytes("PNG", (width, height), color)
+
+
+def write_binary(path: Path, data: bytes) -> Path:
+    """Write test bytes, creating parent directories."""
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_bytes(data)
+    return path
+
+
+def write_note(path: Path, source_resource: str, body: str = "old caption") -> Path:
+    """Write a minimal source-owned image note."""
+    content = (
+        f"---\nname: {path.stem}\ndescription: old\n"
+        f'source_resource: "{source_resource}"\nkind: image\n'
+        f"media_type: image/png\n---\n![[{source_resource[2:-2]}]]\n\n## Caption\n\n{body}\n"
+    )
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(content, encoding="utf-8")
+    return path
+
+
+def caption_json(name: str, description: str, caption: str) -> str:
+    """Build a plain-call caption payload."""
+    return json.dumps({"name": name, "description": description, "caption": caption})
+
+
+def image_processor(app_context, file_store, model, *, routed: bool):
+    """Build either the image processor or the public unified-router path."""
+    if not routed:
+        return AutoImageResourceStep(app_context=app_context, file_store=file_store, as_llm=model)
+    app_context.registry = R
+    return AutoResourceStep(
+        app_context=app_context,
+        dispatch_steps=[
+            {"backend": "auto_image_resource_step", "file_store": file_store, "as_llm": model},
+            {
+                "backend": "auto_text_resource_step",
+                "file_store": file_store,
+                "agent_wrapper": FakeAgentWrapper(),
+            },
+        ],
+    )
+
+
+@dataclass
+class AutoResourceTestEnv:
+    """Started, isolated workspace shared by one test invocation."""
+
+    workspace: Path
+    app_context: object
+    file_store: LocalFileStore
+
+    def write_binary(self, relative_path: str, data: bytes) -> Path:
+        """Write bytes relative to this workspace."""
+        return write_binary(self.workspace / relative_path, data)
+
+    def write_note(self, relative_path: str, source_resource: str, body: str = "old caption") -> Path:
+        """Write a source-owned note relative to this workspace."""
+        return write_note(self.workspace / relative_path, source_resource, body)
+
+    def processor(self, model, *, routed: bool = False):
+        """Build the direct processor or unified router for this workspace."""
+        return image_processor(self.app_context, self.file_store, model, routed=routed)
+
+    async def run(self, step, changes, **context_kwargs):
+        """Run one processor invocation with a fresh runtime context."""
+        return await step(RuntimeContext(changes=changes, **context_kwargs))
+
+
+@pytest_asyncio.fixture
+async def auto_resource_env(tmp_path, monkeypatch):
+    """Yield a started resource-test workspace and always close its file store."""
+    workspace = tmp_path / "workspace"
+    workspace.mkdir()
+    monkeypatch.chdir(workspace)
+    app_context = make_app_context(workspace)
+    file_store = LocalFileStore(name="test_store", embedding_store="")
+    await file_store.start()
+    _install_file_jobs(app_context, file_store)
+    try:
+        yield AutoResourceTestEnv(workspace, app_context, file_store)
+    finally:
+        await file_store.close()

tests/unit/conftest.py

@@ -0,0 +1,5 @@
+"""Fixtures shared by focused unit-test modules."""
+
+from .auto_resource_test_support import auto_resource_env
+
+__all__ = ["auto_resource_env"]

tests/unit/test_auto_resource_review_regressions.py

@@ -0,0 +1,168 @@
+"""Regression tests for the safety findings from the Auto Resource PR review."""
+
+import frontmatter
+import pytest
+
+from .auto_resource_test_support import (
+    FakeVisionModel,
+    StructuredVisionModel,
+    caption_json,
+    image_bytes,
+    write_binary,
+    write_note,
+)
+
+pytestmark = pytest.mark.asyncio
+
+
+@pytest.mark.parametrize("routed", [False, True], ids=["image", "unified-router"])
+async def test_image_rejects_paths_outside_the_resource_tree(routed, auto_resource_env, tmp_path):
+    """Traversal and external paths fail before image reads."""
+    env = auto_resource_env
+    outside = write_binary(tmp_path / "outside.png", image_bytes())
+    nonresource = env.write_binary("private.png", image_bytes())
+
+    model = FakeVisionModel(caption_json("unsafe", "Unsafe", "Must not be read."))
+    response = await env.run(
+        env.processor(model, routed=routed),
+        [
+            {"change": "added", "path": "resource/2026-01-01/../../../outside.png"},
+            {"change": "added", "path": str(outside)},
+            {"change": "added", "path": str(nonresource)},
+        ],
+    )
+
+    results = response.metadata["results"]
+    assert response.success is False
+    assert len(results) == 3
+    assert all(item["metadata"]["action"] == "failed" for item in results)
+    assert all(item["metadata"]["modified"] is False for item in results)
+    assert "cannot contain '.' or '..'" in results[0]["metadata"]["error"]
+    assert "must stay inside the workspace" in results[1]["metadata"]["error"]
+    assert "configured resource directory" in results[2]["metadata"]["error"]
+    assert not model.calls
+    assert outside.read_bytes() == image_bytes()
+
+
+@pytest.mark.parametrize("routed", [False, True], ids=["image", "unified-router"])
+async def test_image_rejects_resource_symlink_outside_workspace(routed, auto_resource_env, tmp_path):
+    """An escaping resource symlink fails without weakening other containment tests."""
+    env = auto_resource_env
+    outside = write_binary(tmp_path / "outside.png", image_bytes())
+    external_link = env.workspace / "resource/2026-01-01/external.png"
+    external_link.parent.mkdir(parents=True, exist_ok=True)
+    try:
+        external_link.symlink_to(outside)
+    except OSError as exc:
+        pytest.skip(f"symlinks unavailable: {exc}")
+
+    model = FakeVisionModel(caption_json("unsafe", "Unsafe", "Must not be read."))
+    response = await env.run(
+        env.processor(model, routed=routed),
+        [{"change": "added", "path": str(external_link)}],
+    )
+
+    result = response.metadata["results"][0]
+    assert response.success is False
+    assert result["metadata"]["action"] == "failed"
+    assert result["metadata"]["modified"] is False
+    assert "must stay inside the workspace" in result["metadata"]["error"]
+    assert not model.calls
+    assert outside.read_bytes() == image_bytes()
+
+
+@pytest.mark.parametrize("routed", [False, True], ids=["image", "unified-router"])
+async def test_image_internal_symlink_keeps_logical_provenance(routed, auto_resource_env):
+    """An internal symlink is read safely while ownership follows the watched alias."""
+    env = auto_resource_env
+    target = env.write_binary("resource/2026-01-01/original.png", image_bytes())
+    link = target.with_name("alias.png")
+    try:
+        link.symlink_to(target.name)
+    except OSError as exc:
+        pytest.skip(f"symlinks unavailable: {exc}")
+
+    model = FakeVisionModel(caption_json("linked-image", "Linked", "An internal linked image."))
+    step = env.processor(model, routed=routed)
+    response = await env.run(step, [{"change": "added", "path": str(link)}])
+
+    note_path = env.workspace / "daily/2026-01-01/linked-image.md"
+    post = frontmatter.loads(note_path.read_text(encoding="utf-8"))
+    assert response.success is True
+    assert post.metadata["source_resource"] == "[[resource/2026-01-01/alias.png]]"
+    assert "![[resource/2026-01-01/alias.png]]" in post.content
+    assert len(model.calls) == 1
+
+    link.unlink()
+    deleted = await env.run(step, [{"change": "deleted", "path": str(link)}])
+    assert deleted.success is True
+    assert deleted.metadata["results"][0]["metadata"]["action"] == "deleted"
+    assert not note_path.exists()
+    assert target.is_file()
+
+
+@pytest.mark.parametrize("routed", [False, True], ids=["image", "unified-router"])
+@pytest.mark.parametrize("existing_owner", ["different-source", "no-source"])
+@pytest.mark.parametrize("change", ["modified", "deleted"])
+async def test_image_preserves_unowned_same_stem_note(routed, existing_owner, change, auto_resource_env):
+    """Upsert and delete never claim a same-stem note without exact ownership."""
+    env = auto_resource_env
+    source = env.workspace / "resource/2026-01-01/img.png"
+    if change == "modified":
+        write_binary(source, image_bytes())
+    same_stem = env.workspace / "daily/2026-01-01/img.md"
+    if existing_owner == "different-source":
+        write_note(same_stem, "[[resource/2026-01-01/other.png]]", body="unrelated image note")
+    else:
+        same_stem.parent.mkdir(parents=True, exist_ok=True)
+        same_stem.write_text(
+            "---\nname: img\ndescription: user-owned note\n---\nuser-owned bytes\n",
+            encoding="utf-8",
+        )
+    before = same_stem.read_bytes()
+
+    model = FakeVisionModel(caption_json("generated-caption", "Generated", "Generated caption."))
+    response = await env.run(env.processor(model, routed=routed), [{"change": change, "path": str(source)}])
+
+    assert response.success is True
+    assert same_stem.read_bytes() == before
+    if change == "deleted":
+        result = response.metadata["results"][0]["metadata"]
+        assert result["action"] == "skipped"
+        assert result["reason"] == "resource_note_not_found"
+        assert result["modified"] is False
+    else:
+        generated = env.workspace / "daily/2026-01-01/generated-caption.md"
+        post = frontmatter.loads(generated.read_text(encoding="utf-8"))
+        assert post.metadata["source_resource"] == "[[resource/2026-01-01/img.png]]"
+        assert "Generated caption." in post.content
+
+
+@pytest.mark.parametrize("routed", [False, True], ids=["image", "unified-router"])
+@pytest.mark.parametrize("plain_text", ["   ", "```json\n\n```"], ids=["whitespace", "empty-json-fence"])
+async def test_blank_plain_caption_does_not_create_or_overwrite_note(routed, plain_text, auto_resource_env):
+    """An empty structured result plus blank plain fallback leaves notes untouched."""
+    env = auto_resource_env
+    new_source = env.write_binary("resource/2026-01-01/blank-new.png", image_bytes())
+    old_source = env.write_binary("resource/2026-01-01/blank-old.png", image_bytes())
+    old_note = env.write_note(
+        "daily/2026-01-01/preserved.md",
+        "[[resource/2026-01-01/blank-old.png]]",
+        body="caption that must survive",
+    )
+    before = old_note.read_bytes()
+    model = StructuredVisionModel(content={}, plain_text=plain_text)
+    step = env.processor(model, routed=routed)
+
+    added = await env.run(step, [{"change": "added", "path": str(new_source)}])
+    modified = await env.run(step, [{"change": "modified", "path": str(old_source)}])
+
+    for response in (added, modified):
+        result = response.metadata["results"][0]["metadata"]
+        assert response.success is False
+        assert result["action"] == "failed"
+        assert result["modified"] is False
+        assert "no usable caption" in result["error"]
+    assert not (env.workspace / "daily/2026-01-01/blank-new.md").exists()
+    assert old_note.read_bytes() == before
+    assert len(model.structured_calls) == len(model.plain_calls) == 2

tests/unit/test_background_steps.py

@@ -189,6 +189,14 @@ def test_match_file_suffix():
     print("✓ test_match_file_suffix passed")
 
 
+def test_match_file_suffix_is_case_insensitive():
+    """Configured suffixes match uppercase file extensions."""
+    rules = [WatchRule(path=Path("/workspace/resource"), suffixes=["jpg", ".png"])]
+    assert match_file("/workspace/resource/photo.JPG", rules)
+    assert match_file("/workspace/resource/sub/diagram.PNG", rules)
+    assert not match_file("/workspace/resource/photo.GIF", rules)
+
+
 def test_match_file_no_suffix_filter():
     """Empty suffixes list means all files match."""
     rules = [WatchRule(path=Path("/workspace/resource"), suffixes=[])]
@@ -222,6 +230,18 @@ def test_collect_existing_filters():
     print("✓ test_collect_existing_filters passed")
 
 
+def test_collect_existing_matches_uppercase_suffixes():
+    """The initial scan includes files whose extension casing differs from the rule."""
+    with tempfile.TemporaryDirectory() as tmpdir:
+        resource = Path(tmpdir) / "resource"
+        uppercase = write_file(resource / "photo.JPG")
+        write_file(resource / "ignore.GIF")
+
+        result = collect_existing([WatchRule(path=resource, suffixes=["jpg"])], recursive=True)
+
+        assert set(result) == {str(uppercase.absolute())}
+
+
 # ---------------------------------------------------------------------------
 # InitChangesStep
 # ---------------------------------------------------------------------------
@@ -1173,6 +1193,21 @@ def test_watch_changes_filter_matches_rules():
     print("✓ test_watch_changes_filter_matches_rules passed")
 
 
+def test_watch_changes_filter_matches_uppercase_suffixes():
+    """The live watcher accepts uppercase extensions configured in lowercase."""
+    with tempfile.TemporaryDirectory() as tmpdir:
+        workspace = Path(tmpdir)
+        (workspace / "resource").mkdir()
+        app_ctx = _make_app_context(workspace)
+
+        step = WatchChangesStep(app_context=app_ctx)
+        step.context = RuntimeContext(watch_dirs=["resource_dir"], watch_suffixes=["jpg"])
+        step._rules = step._get_watch_rules()
+
+        assert step._filter(Change.added, str(workspace / "resource/photo.JPG"))
+        assert not step._filter(Change.added, str(workspace / "resource/photo.PNG"))
+
+
 def test_watch_changes_dispatch_steps_list():
     """dispatch_steps config is stored by BaseStep."""
     step = WatchChangesStep(dispatch_steps=["update_catalog_step", "auto_resource_step"])
@@ -1337,6 +1372,155 @@ def test_auto_resource_handles_file_removed_before_stat():
     asyncio.run(run())
 
 
+def test_auto_resource_rejects_paths_outside_resource_scope_before_agent_call():
+    """Traversal, outside absolute paths, and escaping symlinks fail closed."""
+
+    async def run():
+        with (
+            tempfile.TemporaryDirectory() as tmpdir,
+            tempfile.TemporaryDirectory() as outside_dir,
+            temp_chdir(tmpdir),
+        ):
+            workspace = Path.cwd()
+            outside = write_file(Path(outside_dir) / "outside.txt", "secret")
+            workspace_outside = write_file(workspace / "daily" / "outside.txt", "workspace secret")
+            link = workspace / "resource" / "2026-01-01" / "escape.txt"
+            link.parent.mkdir(parents=True, exist_ok=True)
+            link.symlink_to(outside)
+            wrapper = _FakeAgentWrapper()
+            step = AutoTextResourceStep(app_context=_make_app_context(workspace), agent_wrapper=wrapper)
+
+            resp = await step(
+                RuntimeContext(
+                    changes=[
+                        {"change": "added", "path": "resource/2026-01-01/../../../outside.txt"},
+                        {"change": "modified", "path": str(outside)},
+                        {"change": "added", "path": str(workspace_outside)},
+                        {"change": "added", "path": str(link)},
+                    ],
+                ),
+            )
+
+            assert resp.success is False
+            assert wrapper.inputs == ""
+            assert len(resp.metadata["results"]) == 4
+            assert all(result["metadata"]["action"] == "failed" for result in resp.metadata["results"])
+            assert all(result["metadata"]["modified"] is False for result in resp.metadata["results"])
+            assert outside.read_text(encoding="utf-8") == "secret"
+            assert workspace_outside.read_text(encoding="utf-8") == "workspace secret"
+
+    asyncio.run(run())
+
+
+def test_auto_resource_rejects_traversal_delete_without_touching_note():
+    """A malicious deleted path cannot reach or remove a daily note."""
+
+    async def run():
+        with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
+            workspace = Path.cwd()
+            note = write_file(
+                workspace / "daily" / "2026-01-01" / "outside.md",
+                "---\nname: outside\n"
+                "source_resource: '[[resource/2026-01-01/../../../outside.txt]]'\n---\nkeep me\n",
+            )
+            step = AutoTextResourceStep(app_context=_make_app_context(workspace))
+
+            resp = await step(
+                RuntimeContext(
+                    changes=[
+                        {"change": "deleted", "path": "resource/2026-01-01/../../../outside.txt"},
+                    ],
+                ),
+            )
+
+            result = resp.metadata["results"][0]
+            assert resp.success is False
+            assert result["metadata"]["action"] == "failed"
+            assert result["metadata"]["modified"] is False
+            assert note.read_text(encoding="utf-8").endswith("keep me\n")
+
+    asyncio.run(run())
+
+
+def test_auto_resource_internal_symlink_keeps_logical_source_identity():
+    """A safe internal symlink is read by target while provenance keeps the link path."""
+
+    async def run():
+        with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
+            workspace = Path.cwd()
+            target = write_file(workspace / "resource" / "2026-01-01" / "target.txt", "visible")
+            link = workspace / "resource" / "2026-01-01" / "link.txt"
+            link.symlink_to(target)
+            captured = {}
+            step = AutoTextResourceStep(app_context=_make_app_context(workspace))
+
+            async def fake_upsert(file_path, date_str, note_stem, added, source_path):
+                captured.update(
+                    {
+                        "file_path": file_path,
+                        "date_str": date_str,
+                        "note_stem": note_stem,
+                        "added": added,
+                        "source_path": source_path,
+                    },
+                )
+                step.context.response.success = True
+                step.context.response.answer = "ok"
+
+            step._handle_upsert = fake_upsert
+            resp = await step(RuntimeContext(changes=[{"change": "added", "path": str(link)}]))
+
+            assert resp.success is True
+            assert captured == {
+                "file_path": "resource/2026-01-01/link.txt",
+                "date_str": "2026-01-01",
+                "note_stem": "link",
+                "added": True,
+                "source_path": target.resolve(),
+            }
+
+    asyncio.run(run())
+
+
+def test_auto_resource_accepts_absolute_resource_dir_inside_workspace():
+    """An absolute in-workspace resource_dir keeps a workspace-relative source identity."""
+
+    async def run():
+        with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
+            workspace = Path.cwd()
+            resource_dir = workspace / "assets"
+            source = write_file(resource_dir / "2026-01-01" / "report.txt", "visible")
+            step = AutoTextResourceStep(app_context=_make_app_context(workspace, resource_dir=str(resource_dir)))
+            captured = {}
+
+            async def fake_upsert(file_path, date_str, note_stem, added, source_path):
+                captured.update(
+                    {
+                        "file_path": file_path,
+                        "date_str": date_str,
+                        "note_stem": note_stem,
+                        "added": added,
+                        "source_path": source_path,
+                    },
+                )
+                step.context.response.success = True
+                step.context.response.answer = "ok"
+
+            step._handle_upsert = fake_upsert
+            response = await step(RuntimeContext(changes=[{"change": "added", "path": str(source)}]))
+
+            assert response.success is True
+            assert captured == {
+                "file_path": "assets/2026-01-01/report.txt",
+                "date_str": "2026-01-01",
+                "note_stem": "report",
+                "added": True,
+                "source_path": source.resolve(),
+            }
+
+    asyncio.run(run())
+
+
 def test_auto_resource_accepts_loose_root_resource():
     """Root-level resource files use today's date without moving the source."""
 
@@ -1350,9 +1534,15 @@ def test_auto_resource_accepts_loose_root_resource():
 
             step = AutoTextResourceStep(app_context=app_ctx)
 
-            async def fake_upsert(file_path, date_str, note_stem, created):
+            async def fake_upsert(file_path, date_str, note_stem, created, source_path):
                 captured.update(
-                    {"file_path": file_path, "date_str": date_str, "note_stem": note_stem, "created": created},
+                    {
+                        "file_path": file_path,
+                        "date_str": date_str,
+                        "note_stem": note_stem,
+                        "created": created,
+                        "source_path": source_path,
+                    },
                 )
                 step.context.response.success = True
                 step.context.response.answer = "ok"
@@ -1369,6 +1559,7 @@ def test_auto_resource_accepts_loose_root_resource():
                 "date_str": today,
                 "note_stem": "report",
                 "created": True,
+                "source_path": source.resolve(),
             }
         print("✓ test_auto_resource_accepts_loose_root_resource passed")
 
@@ -1389,9 +1580,15 @@ def test_auto_resource_loose_root_resource_keeps_existing_dated_resource():
 
             step = AutoTextResourceStep(app_context=app_ctx)
 
-            async def fake_upsert(file_path, date_str, note_stem, created):
+            async def fake_upsert(file_path, date_str, note_stem, created, source_path):
                 captured.update(
-                    {"file_path": file_path, "date_str": date_str, "note_stem": note_stem, "created": created},
+                    {
+                        "file_path": file_path,
+                        "date_str": date_str,
+                        "note_stem": note_stem,
+                        "created": created,
+                        "source_path": source_path,
+                    },
                 )
                 step.context.response.success = True
                 step.context.response.answer = "ok"
@@ -1408,6 +1605,7 @@ def test_auto_resource_loose_root_resource_keeps_existing_dated_resource():
                 "date_str": today,
                 "note_stem": "report",
                 "created": True,
+                "source_path": source.resolve(),
             }
         print("✓ test_auto_resource_loose_root_resource_keeps_existing_dated_resource passed")
 
@@ -1451,6 +1649,98 @@ def test_auto_resource_modified_missing_note_uses_create_tools():
     asyncio.run(run())
 
 
+def test_auto_resource_create_preserves_unowned_same_stem_note():
+    """A no-source same-stem note is user-owned and never used as the staging path."""
+
+    async def run():
+        with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
+            workspace = Path.cwd()
+            app_ctx = _make_app_context(workspace)
+            fs = LocalFileStore(name="test_store", embedding_store="")
+            wrapper = _FakeAgentWrapper()
+            await fs.start()
+            _install_file_jobs(app_ctx, fs)
+            try:
+                source = write_file(workspace / "resource" / "2026-01-01" / "report.txt", "resource body")
+                user_note = write_file(
+                    workspace / "daily" / "2026-01-01" / "report.md",
+                    "---\nname: report\ndescription: private user note\n---\nkeep this body\n",
+                )
+                original = user_note.read_bytes()
+
+                def write_allocated_target(inputs, _kwargs):
+                    target_line = next(
+                        line for line in str(inputs).splitlines() if line.startswith("Target note path: ")
+                    )
+                    target_path = target_line.removeprefix("Target note path: ").strip()
+                    assert target_path != "daily/2026-01-01/report.md"
+                    write_file(
+                        workspace / target_path,
+                        "---\nname: generated-topic\ndescription: resource summary\n"
+                        "source_resource: '[[resource/2026-01-01/report.txt]]'\n---\nsummary\n",
+                    )
+
+                wrapper.on_reply = write_allocated_target
+                step = AutoTextResourceStep(app_context=app_ctx, file_store=fs, agent_wrapper=wrapper)
+                resp = await step(RuntimeContext(changes=[{"change": "added", "path": str(source)}]))
+
+                result_meta = resp.metadata["results"][0]["metadata"]
+                assert resp.success is True
+                assert result_meta["created"] is True
+                assert result_meta["path"] == "daily/2026-01-01/generated-topic.md"
+                assert user_note.read_bytes() == original
+                assert (workspace / result_meta["path"]).is_file()
+            finally:
+                await fs.close()
+
+    asyncio.run(run())
+
+
+def test_auto_resource_reports_modified_when_post_write_lookup_fails():
+    """A text note written before post-processing failure remains reported as modified."""
+
+    async def run():
+        with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
+            workspace = Path.cwd()
+            app_ctx = _make_app_context(workspace)
+            file_store = LocalFileStore(name="test_store", embedding_store="")
+            wrapper = _FakeAgentWrapper()
+            await file_store.start()
+            _install_file_jobs(app_ctx, file_store)
+            try:
+                source = write_file(workspace / "resource/2026-01-01/report.txt", "resource body")
+                wrapper.on_reply = lambda *_: write_file(
+                    workspace / "daily/2026-01-01/report.md",
+                    "---\nname: report\nsource_resource: '[[resource/2026-01-01/report.txt]]'\n---\nsummary\n",
+                )
+                step = AutoTextResourceStep(app_context=app_ctx, file_store=file_store, agent_wrapper=wrapper)
+                list_resource_note = step._list_resource_note
+                calls = 0
+
+                async def fail_second_lookup(day, file_path):
+                    nonlocal calls
+                    calls += 1
+                    if calls == 2:
+                        raise RuntimeError("daily_list failed after agent write")
+                    return await list_resource_note(day, file_path)
+
+                step._list_resource_note = fail_second_lookup
+                response = await step(RuntimeContext(changes=[{"change": "added", "path": str(source)}]))
+
+                result = response.metadata["results"][0]
+                assert response.success is False
+                assert result["metadata"]["action"] == "failed"
+                assert result["metadata"]["path"] == "daily/2026-01-01/report.md"
+                assert result["metadata"]["created"] is True
+                assert result["metadata"]["modified"] is True
+                assert "daily_list failed after agent write" in result["metadata"]["error"]
+                assert (workspace / "daily/2026-01-01/report.md").is_file()
+            finally:
+                await file_store.close()
+
+    asyncio.run(run())
+
+
 def test_auto_resource_sanitizes_invalid_generated_name():
     """Invalid LLM-suggested names are sanitized before renaming."""
 
@@ -1637,8 +1927,8 @@ def test_auto_resource_reports_unmodified_when_agent_skips_existing_note():
     asyncio.run(run())
 
 
-def test_auto_resource_deletes_loose_root_resource_note_for_today():
-    """Deleting a loose root resource deletes today's same-stem note."""
+def test_auto_resource_preserves_unowned_loose_root_same_stem_note():
+    """Deleting a loose resource never claims an unowned same-stem user note."""
 
     async def run():
         with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
@@ -1656,12 +1946,15 @@ def test_auto_resource_deletes_loose_root_resource_note_for_today():
 
                 assert resp.success is True
                 assert resp.metadata["results"][0]["path"] == "resource/report.txt"
-                assert resp.metadata["results"][0]["metadata"]["modified"] is True
-                assert resp.metadata["modified"] is True
-                assert not note_path.exists()
+                result_meta = resp.metadata["results"][0]["metadata"]
+                assert result_meta["action"] == "skipped"
+                assert result_meta["reason"] == "resource_note_not_found"
+                assert result_meta["modified"] is False
+                assert resp.metadata["modified"] is False
+                assert note_path.read_text(encoding="utf-8") == "---\nname: report\n---\nbody\n"
             finally:
                 await fs.close()
-        print("✓ test_auto_resource_deletes_loose_root_resource_note_for_today passed")
+        print("✓ test_auto_resource_preserves_unowned_loose_root_same_stem_note passed")
 
     asyncio.run(run())
 
@@ -1939,7 +2232,7 @@ if __name__ == "__main__":
     test_auto_resource_update_keeps_existing_renamed_path()
     test_auto_memory_uses_message_day_for_historical_create()
     test_auto_memory_rejects_invalid_explicit_date_before_saving_session()
-    test_auto_resource_deletes_loose_root_resource_note_for_today()
+    test_auto_resource_preserves_unowned_loose_root_same_stem_note()
     test_auto_resource_deletes_renamed_note_by_source_resource()
     test_auto_resource_result_hook_is_optional_and_isolated()
     # LogChangesStep