diff --git a/reme/steps/evolve/base_auto_resource.py b/reme/steps/evolve/base_auto_resource.py
index afef583d..04b0ae4c 100644
--- a/reme/steps/evolve/base_auto_resource.py
+++ b/reme/steps/evolve/base_auto_resource.py
@@ -1,15 +1,11 @@
 """Shared lifecycle and helpers for automatic resource processors."""
 
-import asyncio
-import errno
 import hashlib
-import os
 import re
-import stat
 from abc import abstractmethod
 from collections.abc import Mapping
 from contextlib import contextmanager
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from pathlib import Path, PurePosixPath
 from typing import Any
 
@@ -26,79 +22,34 @@ _SOURCE_RESOURCE_KEY = "source_resource"
 _DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$")
 _UNSAFE_FILENAME_CHARS = re.compile(r'[<>:"/\\|?*\x00-\x1f]+')
 _LOOKUP_BATCH_KEY = "_auto_resource_lookup_batch"
-_LookupKey = tuple[Path, str, int, int]
 
 
-@dataclass(frozen=True)
-class _ResourceDaySnapshot:
-    """Metadata only: all direct Markdown files, not just currently owned notes."""
-
-    directory: Path
-    files: dict[str, tuple[Path | None, int, int, int]]
-
-
-def _snapshot_note_day(workspace: Path, directory: Path) -> _ResourceDaySnapshot | None:
-    """Inspect file metadata without reading bodies or parsing frontmatter."""
-    try:
-        files = {}
-        with os.scandir(directory) as entries:
-            for entry in entries:
-                if Path(entry.name).suffix != ".md":
-                    continue
-                target = None
-                if entry.is_symlink():
-                    try:
-                        target, error = resolve_path(workspace, entry.path)
-                    except RuntimeError:
-                        # Path.resolve reports symlink loops this way on 3.11/3.12.
-                        continue
-                    except OSError as exc:
-                        if exc.errno == errno.ELOOP:
-                            continue
-                        raise
-                    if error or target is None:
-                        raise ValueError(f"invalid resource note path {entry.path!r}: {error}")
-                try:
-                    value = entry.stat()
-                    if stat.S_ISREG(value.st_mode):
-                        # ctime also catches edits which restore mtime on POSIX.
-                        files[entry.name] = (target, value.st_mtime_ns, value.st_size, value.st_ctime_ns)
-                except FileNotFoundError:
-                    # A removed file or a dangling internal link is not a note.
-                    continue
-                except OSError as exc:
-                    # Python 3.13 can report loops at stat rather than resolve.
-                    if exc.errno == errno.ELOOP:
-                        continue
-                    raise
-        return _ResourceDaySnapshot(directory, files)
-    except (FileNotFoundError, NotADirectoryError):
-        return None
-    except OSError as exc:
-        # Cyclic date directories are not readable history either.
-        if exc.errno == errno.ELOOP:
-            return None
-        raise
-
-
-@dataclass(frozen=True)
-class _ResourceDayLookup:
-    """One day's first owner per source, paired with its pre-read metadata."""
-
-    snapshot: _ResourceDaySnapshot
-    owners: dict[str, str]
+def _resource_note_owners(notes: list[dict]) -> dict[str, str]:
+    """Preserve daily_list's first-match ordering for each exact source link."""
+    owners = {}
+    for note in notes:
+        source = str(note.get(_SOURCE_RESOURCE_KEY, "")).strip()
+        if source:
+            owners.setdefault(source, str(note["path"]))
+    return owners
 
 
+@dataclass
 class _ResourceLookupBatch:
-    """Keep lookup data and its refresh lock within one resource invocation."""
+    """Cache historical ownership only for this sequential resource invocation.
 
-    def __init__(self):
-        self.indexes: dict[_LookupKey, dict[str, _ResourceDayLookup]] = {}
-        self.lock = asyncio.Lock()
+    Refresh our own changes; external edits are read by the next invocation.
+    None means history has not been built, while an empty dict is valid history.
+    """
 
-    def close(self) -> None:
-        """Drop ownership data; never carry it to another batch."""
-        self.indexes.clear()
+    days: dict[str, dict[str, str]] | None = None
+    dirty_days: set[str] = field(default_factory=set)
+
+    def replace_day(self, day: str, notes: list[dict]) -> None:
+        """Reuse a successful daily_list read without publishing partial history."""
+        if self.days is not None:
+            self.days[day] = _resource_note_owners(notes)
+            self.dirty_days.discard(day)
 
 
 @contextmanager
@@ -114,7 +65,8 @@ def _resource_lookup_scope(context: RuntimeContext, *, reuse: bool = False):
     try:
         yield batch
     finally:
-        batch.close()
+        batch.days = None
+        batch.dirty_days.clear()
         if existed:
             context[_LOOKUP_BATCH_KEY] = previous
         else:
@@ -338,68 +290,54 @@ class BaseAutoResourceStep(BaseStep):
         list_response = await self.run_job("daily_list", date=day)
         if not list_response.success:
             raise RuntimeError(f"daily_list failed: {list_response.answer}")
-        return list_response.metadata.get("notes") or []
+        notes = list_response.metadata.get("notes") or []
+        batch = self.context.get(_LOOKUP_BATCH_KEY)
+        if isinstance(batch, _ResourceLookupBatch):
+            batch.replace_day(day, notes)
+        return notes
 
-    def _resource_note_snapshot(self) -> dict[str, _ResourceDaySnapshot]:
-        """Inspect daily-note metadata, including files without resource ownership."""
+    def _daily_note_days(self) -> list[str]:
+        """Return safe, deterministic daily subdirectories available for lookup."""
         workspace = self.workspace_path.resolve()
         daily_dir = str(self.config_value("daily_dir"))
         daily_root, error = resolve_path(workspace, daily_dir)
         if error or daily_root is None:
             raise ValueError(f"invalid daily_dir {daily_dir!r}: {error or 'cannot resolve path'}")
-        snapshots = {}
-        try:
-            with os.scandir(daily_root) as entries:
-                for entry in entries:
-                    if not _DATE_RE.fullmatch(entry.name):
-                        continue
-                    directory = Path(entry.path)
-                    if entry.is_symlink():
-                        try:
-                            directory, path_error = resolve_path(workspace, entry.path)
-                        except (OSError, RuntimeError):
-                            continue
-                        if path_error or directory is None:
-                            continue
-                    value = _snapshot_note_day(workspace, directory)
-                    if value is not None:
-                        snapshots[entry.name] = value
-        except (FileNotFoundError, NotADirectoryError):
-            return {}
-        return snapshots
+        if not daily_root.is_dir():
+            return []
+        days = []
+        for entry in daily_root.iterdir():
+            if not _DATE_RE.fullmatch(entry.name):
+                continue
+            try:
+                resolved, path_error = resolve_path(workspace, f"{daily_dir}/{entry.name}")
+                if not path_error and resolved is not None and resolved.is_dir():
+                    days.append(entry.name)
+            except (OSError, RuntimeError):
+                # Broken or cyclic links must not prevent lookup in other days.
+                continue
+        return sorted(days)
+
+    def _invalidate_resource_day(self, day: str) -> None:
+        """Reload only a day changed by this batch, if history was already built."""
+        batch = self.context.get(_LOOKUP_BATCH_KEY)
+        if isinstance(batch, _ResourceLookupBatch) and batch.days is not None:
+            batch.dirty_days.add(day)
 
     async def _find_loose_resource_day(self, file_path: str) -> str | None:
         """Resolve ownership through the common batch index, for any resource modality."""
         assert self.context is not None
         with _resource_lookup_scope(self.context, reuse=True) as batch:
-            # Overridden stores/configurations must not share lookup contents.
-            key = (
-                self.workspace_path.resolve(),
-                str(self.config_value("daily_dir")),
-                id(self.file_store),
-                id(self.app_context),
-            )
-            async with batch.lock:
-                snapshots = self._resource_note_snapshot()
-                days = batch.indexes.get(key, {}).copy()
-                for day in sorted(snapshots.keys() | days.keys()):
-                    if day not in snapshots:
-                        days.pop(day)
-                    elif day not in days or days[day].snapshot != snapshots[day]:
-                        owners = {}
-                        for note in await self._list_daily_notes(day):
-                            owner = str(note.get(_SOURCE_RESOURCE_KEY, "")).strip()
-                            if owner and owner not in owners:
-                                owners[owner] = str(note["path"])
-                        # Save the pre-read version so later resources detect writes
-                        # during this read, without retrying the current lookup.
-                        days[day] = _ResourceDayLookup(snapshots[day], owners)
-                # Failed reads leave the previous cache available for a later retry.
-                batch.indexes[key] = days
-                source = self._source_resource_link(file_path)
-                matches = sorted(
-                    (day, cached.owners[source]) for day, cached in days.items() if source in cached.owners
-                )
+            if batch.days is None:
+                days = {}
+                for day in self._daily_note_days():
+                    days[day] = _resource_note_owners(await self._list_daily_notes(day))
+                # Publish only after the entire initial scan succeeds.
+                batch.days = days
+            for day in sorted(batch.dirty_days):
+                await self._list_daily_notes(day)
+            source = self._source_resource_link(file_path)
+            matches = sorted((day, owners[source]) for day, owners in batch.days.items() if source in owners)
         if len(matches) > 1:
             paths = ", ".join(path for _, path in matches)
             raise RuntimeError(f"Multiple daily resource notes claim {file_path}: {paths}")
@@ -678,16 +616,23 @@ class BaseAutoResourceStep(BaseStep):
         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,
-                source_path,
-            )
+        try:
+            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,
+                    source_path,
+                )
+        except Exception:
+            # A processor or post-processing job may have written before failing.
+            self._invalidate_resource_day(date_str)
+            raise
+        if self.context.response.metadata.get("modified"):
+            self._invalidate_resource_day(date_str)
         return {
             "success": self.context.response.success,
             "path": file_path,
diff --git a/reme/steps/file_io/_daily_index.py b/reme/steps/file_io/_daily_index.py
index f5151eeb..72a8863e 100644
--- a/reme/steps/file_io/_daily_index.py
+++ b/reme/steps/file_io/_daily_index.py
@@ -1,6 +1,7 @@
 """Daily-note helpers: session_id validation + day-index rebuild."""
 
 import datetime
+import errno
 from pathlib import Path
 
 import frontmatter
@@ -73,16 +74,41 @@ def scan_notes(workspace_dir: Path, date: str, daily_dir: str) -> list[dict]:
 
         {"path": str, "metadata": dict}
     """
-    date_dir, err = resolve_path(workspace_dir, f"{daily_dir}/{date}")
-    if err or date_dir is None:
-        logger.info(f"scan_notes skipped invalid daily path date={date!r} daily_dir={daily_dir!r} error={err!r}")
-        return []
-    if not date_dir.is_dir():
+    try:
+        date_dir, err = resolve_path(workspace_dir, f"{daily_dir}/{date}")
+        if err or date_dir is None:
+            logger.info(f"scan_notes skipped invalid daily path date={date!r} daily_dir={daily_dir!r} error={err!r}")
+            return []
+        if not date_dir.is_dir():
+            return []
+        entries = sorted(date_dir.iterdir())
+    except RuntimeError:
+        # Path.resolve reports symlink loops this way before Python 3.13.
         return []
+    except OSError as exc:
+        if exc.errno == errno.ELOOP:
+            return []
+        raise
     out: list[dict] = []
-    for md_path in sorted(p for p in date_dir.iterdir() if p.is_file() and p.suffix == ".md"):
+    for md_path in entries:
+        if md_path.suffix != ".md":
+            continue
         try:
-            post = frontmatter.loads(md_path.read_text(encoding="utf-8"))
+            target = md_path
+            if md_path.is_symlink():
+                target, err = resolve_path(workspace_dir, str(md_path))
+                if err or target is None:
+                    continue
+            if not target.is_file():
+                continue
+        except RuntimeError:
+            continue
+        except OSError as exc:
+            if exc.errno == errno.ELOOP:
+                continue
+            raise
+        try:
+            post = frontmatter.loads(target.read_text(encoding="utf-8"))
         except Exception:
             continue
         out.append(
diff --git a/tests/unit/test_auto_resource_batch_lookup.py b/tests/unit/test_auto_resource_batch_lookup.py
index d3d2ec11..7884783a 100644
--- a/tests/unit/test_auto_resource_batch_lookup.py
+++ b/tests/unit/test_auto_resource_batch_lookup.py
@@ -1,18 +1,22 @@
-"""Batch-scoped historical lookup keeps resource ownership safe without repeated scans."""
+"""Reuse history within a resource batch and maintain its own note mutations."""
 
+import asyncio
+import os
 from collections import Counter
 from unittest.mock import patch
 
+import frontmatter
 import pytest
 
-from reme.components.file_store import LocalFileStore
+from reme.components import R
 from reme.components.runtime_context import RuntimeContext
 from reme.schema import Response
-from reme.steps.evolve.auto_image_resource import AutoImageResourceStep
-from reme.steps.evolve.base_auto_resource import BaseAutoResourceStep, _resource_lookup_scope
-from reme.steps.file_io import DailyListStep
+from reme.steps.evolve import base_auto_resource as resource_module
+from reme.steps.evolve.auto_resource import AutoResourceStep
+from reme.steps.evolve.auto_text_resource import AutoTextResourceStep
+from reme.steps.evolve.base_auto_resource import BaseAutoResourceStep
 
-from .auto_resource_test_support import FakeVisionModel, caption_json, image_bytes, make_app_context, write_note
+from .auto_resource_test_support import FakeVisionModel, caption_json, image_bytes
 
 pytest_plugins = ("unit.auto_resource_test_plugin",)
 pytestmark = pytest.mark.asyncio
@@ -21,67 +25,124 @@ _HISTORY_DAYS = ("2025-12-01", "2025-12-02", "2025-12-03")
 
 
 def _seed_history(env) -> None:
-    """Create untouched days that should not be re-read for each resource."""
+    """Create older days untouched by the resource changes under test."""
     for day in _HISTORY_DAYS:
         env.write_note(f"daily/{day}/archive.md", f"[[resource/archive-{day}.txt]]")
 
 
-def _count_daily_lists(env, monkeypatch) -> Counter:
-    """Count real job calls without replacing file parsing or result contracts."""
+def _observe_history(env, monkeypatch) -> Counter:
+    """Observe directory walks, parsing, and daily_list, not every possible standalone stat call."""
     calls = Counter()
+    daily_root = (env.workspace / "daily").resolve()
+    history_text = {
+        path.read_text(encoding="utf-8"): day
+        for day in _HISTORY_DAYS
+        if (path := daily_root / day / "archive.md").is_file()
+    }
+    watched_directories = {
+        str(daily_root): "history_enumerations",
+        **{str(daily_root / day): ("enumerate", day) for day in _HISTORY_DAYS},
+    }
+    original_listdir = os.listdir
+    original_scandir = os.scandir
+    original_loads = frontmatter.loads
     daily_list = env.app_context.jobs["daily_list"]
 
+    def record_enumeration(path):
+        if isinstance(path, (str, bytes, os.PathLike)):
+            key = watched_directories.get(os.fsdecode(path))
+            if key is not None:
+                calls[key] += 1
+
+    def counted_listdir(path="."):
+        record_enumeration(path)
+        return original_listdir(path)
+
+    def counted_scandir(path="."):
+        record_enumeration(path)
+        return original_scandir(path)
+
+    def counted_loads(text, *args, **kwargs):
+        if isinstance(text, str) and text in history_text:
+            calls[("parse", history_text[text])] += 1
+        return original_loads(text, *args, **kwargs)
+
     async def counted_daily_list(**kwargs):
-        calls[kwargs["date"]] += 1
+        calls[("daily_list", kwargs["date"])] += 1
         return await daily_list(**kwargs)
 
+    # Path.iterdir uses listdir or scandir depending on the Python version.
+    # Return the original iterator unchanged, without wrapping DirEntry objects.
+    monkeypatch.setattr(os, "listdir", counted_listdir)
+    monkeypatch.setattr(os, "scandir", counted_scandir)
+    monkeypatch.setattr(frontmatter, "loads", counted_loads)
     monkeypatch.setitem(env.app_context.jobs, "daily_list", counted_daily_list)
     return calls
 
 
+def _assert_history_reads(calls, count=1) -> None:
+    assert calls["history_enumerations"] == count
+    for day in _HISTORY_DAYS:
+        assert calls[("enumerate", day)] == count
+        assert calls[("daily_list", day)] == count
+        assert calls[("parse", day)] == count
+
+
 def _model() -> FakeVisionModel:
-    """Use deterministic model output while retaining the real note lifecycle."""
     return FakeVisionModel(caption_json("caption", "A sample image", "A red square."))
 
 
 @pytest.mark.parametrize("routed", [False, True], ids=["image", "unified-router"])
 @pytest.mark.parametrize("resource_count", [1, 8])
-async def test_loose_batch_scans_unchanged_history_once(routed, resource_count, auto_resource_env, monkeypatch):
-    """Increasing the number of loose resources does not multiply historical scans."""
+@pytest.mark.parametrize("write_notes", [False, True], ids=["missing-deletions", "real-updates"])
+async def test_loose_batch_reads_history_once(
+    routed,
+    resource_count,
+    write_notes,
+    auto_resource_env,
+    monkeypatch,
+):
+    """More resources, including actual note writes, must not cause another history walk."""
     env = auto_resource_env
     _seed_history(env)
-    calls = _count_daily_lists(env, monkeypatch)
     changes = []
+    notes = []
     for index in range(resource_count):
-        source = env.write_binary(f"resource/photo-{index}.png", image_bytes())
-        changes.append({"change": "added", "path": str(source)})
+        source = f"resource/photo-{index}.png"
+        if write_notes:
+            env.write_binary(source, image_bytes())
+            notes.append(env.write_note(f"daily/2026-01-01/photo-{index}.md", f"[[{source}]]"))
+        changes.append({"change": "modified" if write_notes else "deleted", "path": source})
+    before = {note: note.read_bytes() for note in notes}
+    calls = _observe_history(env, monkeypatch)
 
     with patch.object(BaseAutoResourceStep, "_today", return_value="2026-01-01"):
         response = await env.run(env.processor(_model(), routed=routed), changes)
 
     assert response.success is True
     assert response.metadata["processed"] == resource_count
-    assert all(result["metadata"]["created"] for result in response.metadata["results"])
-    assert {day: calls[day] for day in _HISTORY_DAYS} == dict.fromkeys(_HISTORY_DAYS, 1)
-    paths = [result["metadata"]["path"] for result in response.metadata["results"]]
-    assert len(set(paths)) == resource_count
-    assert all((env.workspace / path).is_file() for path in paths)
+    results = response.metadata["results"]
+    assert [result["path"] for result in results] == [item["path"] for item in changes]
+    assert all(result["metadata"]["action"] == ("modified" if write_notes else "skipped") for result in results)
+    _assert_history_reads(calls)
+    if write_notes:
+        assert all(note.read_bytes() != previous for note, previous in before.items())
+        assert all(result["metadata"]["modified"] for result in results)
 
 
-async def test_mixed_processors_share_one_historical_scan(auto_resource_env, monkeypatch):
-    """Text and image sub-batches share lookup work and retain original result order."""
+async def test_mixed_processors_share_history_and_keep_result_order(auto_resource_env, monkeypatch):
+    """Both processors reuse the batch lookup while returning results in input order."""
     env = auto_resource_env
     _seed_history(env)
     text_note = env.write_note("daily/2026-01-01/report.md", "[[resource/report.txt]]")
     image_note = env.write_note("daily/2026-01-01/photo.md", "[[resource/photo.png]]")
-    calls = _count_daily_lists(env, monkeypatch)
+    calls = _observe_history(env, monkeypatch)
     changes = [
         {"change": "deleted", "path": "resource/report.txt"},
         {"change": "deleted", "path": "resource/photo.png"},
     ]
 
-    with patch.object(BaseAutoResourceStep, "_today", return_value="2026-01-02"):
-        response = await env.run(env.processor(_model(), routed=True), changes)
+    response = await env.run(env.processor(_model(), routed=True), changes)
 
     assert response.success is True
     results = response.metadata["results"]
@@ -89,16 +150,40 @@ async def test_mixed_processors_share_one_historical_scan(auto_resource_env, mon
     assert all(result["metadata"]["action"] == "deleted" for result in results)
     assert not text_note.exists()
     assert not image_note.exists()
-    assert {day: calls[day] for day in _HISTORY_DAYS} == dict.fromkeys(_HISTORY_DAYS, 1)
+    _assert_history_reads(calls)
+
+
+@pytest.mark.parametrize("routed", [False, True], ids=["text-step", "text-only-router"])
+@pytest.mark.parametrize("dated", [False, True], ids=["loose-root", "dated-resource"])
+async def test_text_only_batch_uses_history_without_image_processor(routed, dated, auto_resource_env, monkeypatch):
+    """A text-only pipeline neither needs an image processor nor scans history for dated paths."""
+    env = auto_resource_env
+    _seed_history(env)
+    env.app_context.registry = R.copy()
+    prefix = "resource/2026-01-01" if dated else "resource"
+    notes = [env.write_note(f"daily/2026-01-01/report-{i}.md", f"[[{prefix}/report-{i}.txt]]") for i in range(3)]
+    calls = _observe_history(env, monkeypatch)
+    options = {"app_context": env.app_context, "file_store": env.file_store}
+    step = (
+        AutoResourceStep(**options, dispatch_steps=["auto_text_resource_step"])
+        if routed
+        else AutoTextResourceStep(**options)
+    )
+
+    response = await env.run(step, [{"change": "deleted", "path": f"{prefix}/report-{i}.txt"} for i in range(3)])
+
+    assert response.success is True
+    assert all(not note.exists() for note in notes)
+    _assert_history_reads(calls, 0 if dated else 1)
 
 
 @pytest.mark.parametrize("routed", [False, True], ids=["image", "unified-router"])
-async def test_dated_batch_does_not_scan_history(routed, auto_resource_env, monkeypatch):
-    """Dated paths retain their direct-day lookup and do not build a global index."""
+async def test_dated_image_does_not_scan_history(routed, auto_resource_env, monkeypatch):
+    """An explicitly dated image only needs its own day's lifecycle."""
     env = auto_resource_env
     _seed_history(env)
-    source = env.write_binary("resource/2026-01-01/photo.png", image_bytes())
-    calls = _count_daily_lists(env, monkeypatch)
+    source = env.write_binary("resource/2026-01-01/caption.png", image_bytes())
+    calls = _observe_history(env, monkeypatch)
 
     response = await env.run(
         env.processor(_model(), routed=routed),
@@ -106,27 +191,22 @@ async def test_dated_batch_does_not_scan_history(routed, auto_resource_env, monk
     )
 
     assert response.success is True
-    assert response.metadata["results"][0]["metadata"]["path"] == "daily/2026-01-01/caption.md"
-    assert all(calls[day] == 0 for day in _HISTORY_DAYS)
+    path = response.metadata["results"][0]["metadata"]["path"]
+    assert path == "daily/2026-01-01/caption.md"
+    assert (env.workspace / path).is_file()
+    _assert_history_reads(calls, 0)
 
 
 @pytest.mark.parametrize("routed", [False, True], ids=["image", "unified-router"])
-@pytest.mark.parametrize("lookup_value", [None, "user-owned input"], ids=["absent-key", "existing-key"])
-async def test_reused_step_and_context_rebuild_ownership_next_batch(
-    routed,
-    lookup_value,
-    auto_resource_env,
-    monkeypatch,
-):
-    """A watcher-style reused context sees externally changed ownership on its next call."""
+async def test_reused_step_and_context_rebuild_history_next_invocation(routed, auto_resource_env, monkeypatch):
+    """External edits between invocations are visible; batch state never survives the call."""
     env = auto_resource_env
     _seed_history(env)
     original = env.write_note("daily/2026-01-01/original.md", "[[resource/photo.png]]")
-    calls = _count_daily_lists(env, monkeypatch)
+    calls = _observe_history(env, monkeypatch)
     step = env.processor(_model(), routed=routed)
-    context = RuntimeContext(changes=[{"change": "deleted", "path": "resource/photo.png"}], user_value="preserve")
-    if lookup_value is not None:
-        context["_auto_resource_lookup_batch"] = lookup_value
+    changes = [{"change": "deleted", "path": "resource/photo.png"}]
+    context = RuntimeContext(changes=changes, user_value="preserve")
     context_keys = set(context.data)
 
     first = await step(context)
@@ -134,43 +214,46 @@ async def test_reused_step_and_context_rebuild_ownership_next_batch(
     assert first.metadata["results"][0]["metadata"]["path"] == "daily/2026-01-01/original.md"
     assert not original.exists()
     assert set(context.data) == context_keys
+    assert context["changes"] is changes
     assert context["user_value"] == "preserve"
-    assert context.get("_auto_resource_lookup_batch") == lookup_value
 
-    new_owner = env.write_note("daily/2026-01-02/recreated.md", "[[resource/photo.png]]")
+    recreated = env.write_note("daily/2026-01-02/recreated.md", "[[resource/photo.png]]")
     second = await step(context)
 
     assert second.success is True
     assert second.metadata["results"][0]["metadata"]["path"] == "daily/2026-01-02/recreated.md"
     assert second.metadata["results"][0]["metadata"]["action"] == "deleted"
-    assert not new_owner.exists()
+    assert not recreated.exists()
     assert set(context.data) == context_keys
+    assert context["changes"] is changes
     assert context["user_value"] == "preserve"
-    assert context.get("_auto_resource_lookup_batch") == lookup_value
-    assert {day: calls[day] for day in _HISTORY_DAYS} == dict.fromkeys(_HISTORY_DAYS, 2)
+    _assert_history_reads(calls, 2)
 
 
 @pytest.mark.parametrize("routed", [False, True], ids=["image", "unified-router"])
-async def test_repeated_resource_events_keep_current_ownership_across_midnight(routed, auto_resource_env, monkeypatch):
-    """New writes become discoverable and deletion permits a new day's card in the same batch."""
+async def test_create_update_delete_recreate_maintains_ownership_across_midnight(
+    routed,
+    auto_resource_env,
+    monkeypatch,
+):
+    """A card keeps its original day until deletion permits a fresh card on today's date."""
     env = auto_resource_env
     _seed_history(env)
-    env.write_binary("resource/photo.png", image_bytes())
-    calls = _count_daily_lists(env, monkeypatch)
+    env.write_binary("resource/caption.png", image_bytes())
+    calls = _observe_history(env, monkeypatch)
     clock = {"day": "2026-01-01"}
-    # Advance the clock at the shared lifecycle boundary, after the actual write.
-    original_refresh = BaseAutoResourceStep._refresh_day_index  # pylint: disable=protected-access
+    refresh = resource_module.refresh_day_index
 
-    async def refresh_and_advance(step, day):
-        result = await original_refresh(step, day)
+    async def refresh_and_advance(*args, **kwargs):
+        result = await refresh(*args, **kwargs)
         clock["day"] = "2026-01-02"
         return result
 
     monkeypatch.setattr(BaseAutoResourceStep, "_today", lambda _step: clock["day"])
-    monkeypatch.setattr(BaseAutoResourceStep, "_refresh_day_index", refresh_and_advance)
+    monkeypatch.setattr(resource_module, "refresh_day_index", refresh_and_advance)
     response = await env.run(
         env.processor(_model(), routed=routed),
-        [{"change": change, "path": "resource/photo.png"} for change in ("added", "modified", "deleted", "added")],
+        [{"change": change, "path": "resource/caption.png"} for change in ("added", "modified", "deleted", "added")],
     )
 
     assert response.success is True
@@ -184,48 +267,47 @@ async def test_repeated_resource_events_keep_current_ownership_across_midnight(r
     ]
     assert not (env.workspace / "daily/2026-01-01/caption.md").exists()
     assert (env.workspace / "daily/2026-01-02/caption.md").is_file()
-    assert {day: calls[day] for day in _HISTORY_DAYS} == dict.fromkeys(_HISTORY_DAYS, 1)
+    _assert_history_reads(calls)
 
 
 @pytest.mark.parametrize("routed", [False, True], ids=["image", "unified-router"])
 @pytest.mark.parametrize("first_change", ["added", "deleted"])
-async def test_partial_index_failure_does_not_stale_batch_ownership(
+async def test_partial_index_failure_refreshes_only_affected_day(
     routed,
     first_change,
     auto_resource_env,
     monkeypatch,
 ):
-    """Post-write and post-delete failures preserve disk ownership for the next resource event."""
+    """A failure after writing or deleting must not hide that mutation from the next item."""
     env = auto_resource_env
     _seed_history(env)
-    env.write_binary("resource/photo.png", image_bytes())
+    env.write_binary("resource/caption.png", image_bytes())
     if first_change == "deleted":
-        env.write_note("daily/2026-01-01/caption.md", "[[resource/photo.png]]")
-    calls = _count_daily_lists(env, monkeypatch)
+        env.write_note("daily/2026-01-01/caption.md", "[[resource/caption.png]]")
+    calls = _observe_history(env, monkeypatch)
     clock = {"day": "2026-01-01"}
-    # Inject a failure after disk mutation without replacing note processing.
-    original_refresh = BaseAutoResourceStep._refresh_day_index  # pylint: disable=protected-access
+    refresh = resource_module.refresh_day_index
     refresh_calls = 0
 
-    async def fail_first_refresh(step, day):
+    async def fail_first_refresh(*args, **kwargs):
         nonlocal refresh_calls
         refresh_calls += 1
         if refresh_calls == 1:
             clock["day"] = "2026-01-02"
             raise RuntimeError("day index unavailable after disk change")
-        return await original_refresh(step, day)
+        return await refresh(*args, **kwargs)
 
     monkeypatch.setattr(BaseAutoResourceStep, "_today", lambda _step: clock["day"])
-    monkeypatch.setattr(BaseAutoResourceStep, "_refresh_day_index", fail_first_refresh)
+    monkeypatch.setattr(resource_module, "refresh_day_index", fail_first_refresh)
     next_change = "modified" if first_change == "added" else "added"
     response = await env.run(
         env.processor(_model(), routed=routed),
-        [{"change": change, "path": "resource/photo.png"} for change in (first_change, next_change)],
+        [{"change": change, "path": "resource/caption.png"} for change in (first_change, next_change)],
     )
 
+    failed, recovered = response.metadata["results"]
     assert response.success is False
     assert response.metadata["modified"] is True
-    failed, recovered = response.metadata["results"]
     assert failed["success"] is False
     assert failed["metadata"]["modified"] is True
     assert "day index unavailable" in failed["metadata"]["error"]
@@ -234,21 +316,70 @@ async def test_partial_index_failure_does_not_stale_batch_ownership(
     assert recovered["metadata"]["path"] == f"daily/{day}/caption.md"
     assert recovered["metadata"]["created"] is (first_change == "deleted")
     assert (env.workspace / f"daily/{day}/caption.md").is_file()
+    _assert_history_reads(calls)
     assert len(list((env.workspace / "daily").glob("*/caption.md"))) == 1
-    assert {day: calls[day] for day in _HISTORY_DAYS} == dict.fromkeys(_HISTORY_DAYS, 1)
+
+
+async def test_failed_dirty_day_read_remains_retryable(auto_resource_env, monkeypatch):
+    """A failed post-write reload cannot clear the dirty flag or force a new day's card."""
+    env = auto_resource_env
+    _seed_history(env)
+    env.write_binary("resource/caption.png", image_bytes())
+    calls = _observe_history(env, monkeypatch)
+    daily_list = env.app_context.jobs["daily_list"]
+    refresh = resource_module.refresh_day_index
+    clock = {"day": "2026-01-01"}
+    refresh_failed = False
+    read_failed = False
+
+    async def fail_first_refresh(*args, **kwargs):
+        nonlocal refresh_failed
+        if not refresh_failed:
+            refresh_failed = True
+            clock["day"] = "2026-01-02"
+            raise RuntimeError("index refresh failed")
+        return await refresh(*args, **kwargs)
+
+    async def fail_first_dirty_read(**kwargs):
+        nonlocal read_failed
+        if refresh_failed and kwargs["date"] == "2026-01-01" and not read_failed:
+            read_failed = True
+            return Response(success=False, answer="dirty day unavailable")
+        return await daily_list(**kwargs)
+
+    monkeypatch.setattr(BaseAutoResourceStep, "_today", lambda _step: clock["day"])
+    monkeypatch.setattr(resource_module, "refresh_day_index", fail_first_refresh)
+    monkeypatch.setitem(env.app_context.jobs, "daily_list", fail_first_dirty_read)
+    response = await env.run(
+        env.processor(_model(), routed=True),
+        [{"change": change, "path": "resource/caption.png"} for change in ("added", "modified", "modified")],
+    )
+
+    written, interrupted, recovered = response.metadata["results"]
+    assert response.success is False
+    assert written["success"] is False
+    assert written["metadata"]["modified"] is True
+    assert interrupted["success"] is False
+    assert interrupted["metadata"]["modified"] is False
+    assert "dirty day unavailable" in interrupted["metadata"]["error"]
+    assert recovered["success"] is True
+    assert recovered["metadata"]["created"] is False
+    assert recovered["metadata"]["path"] == "daily/2026-01-01/caption.md"
+    assert (env.workspace / "daily/2026-01-01/caption.md").is_file()
+    assert not (env.workspace / "daily/2026-01-02/caption.md").exists()
+    _assert_history_reads(calls)
 
 
 @pytest.mark.parametrize("routed", [False, True], ids=["image", "unified-router"])
-async def test_cross_day_duplicate_owner_only_fails_ambiguous_resource(routed, auto_resource_env, monkeypatch):
-    """Ambiguity is kept per source instead of making the shared index unusable for every item."""
+async def test_cross_day_duplicate_only_fails_ambiguous_resource(routed, auto_resource_env, monkeypatch):
+    """Ambiguous ownership prevents mutation without blocking another valid resource."""
     env = auto_resource_env
     _seed_history(env)
     first = env.write_note("daily/2026-01-01/first.md", "[[resource/duplicate.png]]")
     second = env.write_note("daily/2026-01-02/second.md", "[[resource/duplicate.png]]")
     valid = env.write_note("daily/2026-01-02/valid.md", "[[resource/valid.png]]")
     before = {note: note.read_bytes() for note in (first, second)}
-    calls = _count_daily_lists(env, monkeypatch)
-
+    calls = _observe_history(env, monkeypatch)
     response = await env.run(
         env.processor(_model(), routed=routed),
         [
@@ -266,152 +397,57 @@ async def test_cross_day_duplicate_owner_only_fails_ambiguous_resource(routed, a
     assert deleted["metadata"]["action"] == "deleted"
     assert not valid.exists()
     assert all(note.read_bytes() == contents for note, contents in before.items())
-    assert {day: calls[day] for day in _HISTORY_DAYS} == dict.fromkeys(_HISTORY_DAYS, 1)
+    _assert_history_reads(calls)
 
 
 @pytest.mark.parametrize("routed", [False, True], ids=["image", "unified-router"])
-async def test_same_day_first_owner_semantics_survive_repeated_deletion(routed, auto_resource_env, monkeypatch):
-    """Same-day duplicates retain the existing first-match behavior, including after one is removed."""
+async def test_same_day_duplicates_keep_first_match_after_deletion(routed, auto_resource_env, monkeypatch):
+    """Deleting the first same-day match exposes the next one, preserving existing behavior."""
     env = auto_resource_env
     _seed_history(env)
     first = env.write_note("daily/2026-01-01/first.md", "[[resource/photo.png]]")
     second = env.write_note("daily/2026-01-01/second.md", "[[resource/photo.png]]")
-    calls = _count_daily_lists(env, monkeypatch)
-    changes = [{"change": "deleted", "path": "resource/photo.png"} for _ in range(3)]
-
+    calls = _observe_history(env, monkeypatch)
     with patch.object(BaseAutoResourceStep, "_today", return_value="2026-01-02"):
-        response = await env.run(env.processor(_model(), routed=routed), changes)
+        response = await env.run(
+            env.processor(_model(), routed=routed),
+            [{"change": "deleted", "path": "resource/photo.png"} for _ in range(3)],
+        )
 
     assert response.success is True
     metadata = [result["metadata"] for result in response.metadata["results"]]
     assert [item["action"] for item in metadata] == ["deleted", "deleted", "skipped"]
-    assert [item["path"] for item in metadata[:2]] == [
-        "daily/2026-01-01/first.md",
-        "daily/2026-01-01/second.md",
-    ]
+    assert [item["path"] for item in metadata[:2]] == ["daily/2026-01-01/first.md", "daily/2026-01-01/second.md"]
     assert not first.exists()
     assert not second.exists()
-    assert {day: calls[day] for day in _HISTORY_DAYS} == dict.fromkeys(_HISTORY_DAYS, 1)
+    _assert_history_reads(calls)
 
 
 @pytest.mark.parametrize("routed", [False, True], ids=["image", "unified-router"])
-@pytest.mark.parametrize("next_change", ["deleted", "modified"])
-@pytest.mark.parametrize("initially_missing_target", [False, True], ids=["existing-target", "missing-target"])
-async def test_shared_daily_directory_aliases_refresh_after_write(
-    routed,
-    next_change,
-    initially_missing_target,
-    auto_resource_env,
-    monkeypatch,
-):
-    """Writing through one safe alias invalidates every date exposing the same physical notes."""
-    env = auto_resource_env
-    _seed_history(env)
-    physical_day = env.workspace / ("daily/2026-01-01" if initially_missing_target else "archive-day")
-    if not initially_missing_target:
-        physical_day.mkdir()
-    alias_days = ("2026-01-02",) if initially_missing_target else ("2026-01-01", "2026-01-02")
-    for day in alias_days:
-        try:
-            (env.workspace / "daily" / day).symlink_to(physical_day, target_is_directory=True)
-        except OSError as exc:
-            pytest.skip(f"symlinks unavailable: {exc}")
-    env.write_binary("resource/photo.png", image_bytes())
-    calls = _count_daily_lists(env, monkeypatch)
-    model = _model()
-
-    with patch.object(BaseAutoResourceStep, "_today", return_value="2026-01-01"):
-        response = await env.run(
-            env.processor(model, routed=routed),
-            [{"change": change, "path": "resource/photo.png"} for change in ("added", next_change)],
-        )
-
-    created, ambiguous = response.metadata["results"]
-    assert response.success is False
-    assert created["success"] is True
-    assert created["metadata"]["created"] is True
-    assert ambiguous["success"] is False
-    assert ambiguous["metadata"]["modified"] is False
-    assert "Multiple daily resource notes claim resource/photo.png" in ambiguous["metadata"]["error"]
-    assert "daily/2026-01-01/caption.md" in ambiguous["metadata"]["error"]
-    assert "daily/2026-01-02/caption.md" in ambiguous["metadata"]["error"]
-    assert (physical_day / "caption.md").is_file()
-    assert len(model.calls) == 1
-    assert {day: calls[day] for day in _HISTORY_DAYS} == dict.fromkeys(_HISTORY_DAYS, 1)
-
-
-@pytest.mark.parametrize("routed", [False, True], ids=["image", "unified-router"])
-async def test_failed_dirty_day_refresh_retries_without_changing_ownership(routed, auto_resource_env, monkeypatch):
-    """A failed targeted refresh stays dirty; the following event recovers the first-day card."""
-    env = auto_resource_env
-    _seed_history(env)
-    env.write_binary("resource/photo.png", image_bytes())
-    calls = _count_daily_lists(env, monkeypatch)
-    daily_list = env.app_context.jobs["daily_list"]
-    # Fail only the historical read after successful note finalization.
-    original_refresh = BaseAutoResourceStep._refresh_day_index  # pylint: disable=protected-access
-    clock = {"day": "2026-01-01"}
-    failed = False
-
-    async def refresh_and_advance(step, day):
-        result = await original_refresh(step, day)
-        clock["day"] = "2026-01-02"
-        return result
-
-    async def fail_first_dirty_read(**kwargs):
-        nonlocal failed
-        if clock["day"] == "2026-01-02" and kwargs["date"] == "2026-01-01" and not failed:
-            failed = True
-            return Response(success=False, answer="temporary dirty-day read failure")
-        return await daily_list(**kwargs)
-
-    monkeypatch.setattr(BaseAutoResourceStep, "_today", lambda _step: clock["day"])
-    monkeypatch.setattr(BaseAutoResourceStep, "_refresh_day_index", refresh_and_advance)
-    monkeypatch.setitem(env.app_context.jobs, "daily_list", fail_first_dirty_read)
-    response = await env.run(
-        env.processor(_model(), routed=routed),
-        [{"change": change, "path": "resource/photo.png"} for change in ("added", "modified", "modified")],
-    )
-
-    created, interrupted, recovered = response.metadata["results"]
-    assert response.success is False
-    assert created["success"] is True
-    assert interrupted["success"] is False
-    assert interrupted["metadata"]["modified"] is False
-    assert "temporary dirty-day read failure" in interrupted["metadata"]["error"]
-    assert recovered["success"] is True
-    assert recovered["metadata"]["created"] is False
-    assert created["metadata"]["path"] == recovered["metadata"]["path"] == "daily/2026-01-01/caption.md"
-    assert (env.workspace / "daily/2026-01-01/caption.md").is_file()
-    assert not (env.workspace / "daily/2026-01-02/caption.md").exists()
-    assert {day: calls[day] for day in _HISTORY_DAYS} == dict.fromkeys(_HISTORY_DAYS, 1)
-
-
-@pytest.mark.parametrize("routed", [False, True], ids=["image", "unified-router"])
-async def test_interrupted_history_scan_is_not_reused_as_complete(routed, auto_resource_env, monkeypatch):
-    """An item after transient daily_list failure rebuilds history and finds previously unvisited ownership."""
+async def test_initial_read_failure_does_not_publish_partial_history(routed, auto_resource_env, monkeypatch):
+    """A transient history read failure makes the next item retry the initial scan."""
     env = auto_resource_env
     _seed_history(env)
     owned_note = env.write_note("daily/2026-01-01/original.md", "[[resource/photo.png]]")
-    calls = Counter()
+    calls = _observe_history(env, monkeypatch)
     daily_list = env.app_context.jobs["daily_list"]
+    failed_once = False
 
     async def fail_once_mid_scan(**kwargs):
-        day = kwargs["date"]
-        calls[day] += 1
-        if day == _HISTORY_DAYS[1] and calls[day] == 1:
+        nonlocal failed_once
+        if kwargs["date"] == _HISTORY_DAYS[1] and not failed_once:
+            failed_once = True
             return Response(success=False, answer="temporary history read failure")
         return await daily_list(**kwargs)
 
     monkeypatch.setitem(env.app_context.jobs, "daily_list", fail_once_mid_scan)
-    with patch.object(BaseAutoResourceStep, "_today", return_value="2026-01-02"):
-        response = await env.run(
-            env.processor(_model(), routed=routed),
-            [
-                {"change": "deleted", "path": "resource/missing.png"},
-                {"change": "deleted", "path": "resource/photo.png"},
-            ],
-        )
+    response = await env.run(
+        env.processor(_model(), routed=routed),
+        [
+            {"change": "deleted", "path": "resource/missing.png"},
+            {"change": "deleted", "path": "resource/photo.png"},
+        ],
+    )
 
     failed, recovered = response.metadata["results"]
     assert response.success is False
@@ -419,23 +455,39 @@ async def test_interrupted_history_scan_is_not_reused_as_complete(routed, auto_r
     assert failed["metadata"]["modified"] is False
     assert "temporary history read failure" in failed["metadata"]["error"]
     assert recovered["success"] is True
-    assert recovered["metadata"]["action"] == "deleted"
     assert recovered["metadata"]["path"] == "daily/2026-01-01/original.md"
+    assert recovered["metadata"]["action"] == "deleted"
     assert not owned_note.exists()
-    assert [calls[day] for day in _HISTORY_DAYS] == [2, 2, 1]
+    assert calls["history_enumerations"] == 2
+    assert [calls[("parse", day)] for day in _HISTORY_DAYS] == [2, 1, 1]
 
 
-async def test_router_dispatch_exception_cleans_lookup_scope(auto_resource_env, monkeypatch):
-    """An exception after a processed sub-batch cannot leak a cached index into a reused context."""
+async def test_empty_history_is_initialized_once(auto_resource_env, monkeypatch):
+    """An empty ownership table is a successful scan, not an uninitialized cache."""
+    env = auto_resource_env
+    (env.workspace / "daily").mkdir(exist_ok=True)
+    calls = _observe_history(env, monkeypatch)
+
+    response = await env.run(
+        env.processor(_model(), routed=True),
+        [{"change": "deleted", "path": f"resource/missing-{index}.png"} for index in range(8)],
+    )
+
+    assert response.success is True
+    assert all(result["metadata"]["action"] == "skipped" for result in response.metadata["results"])
+    assert calls["history_enumerations"] == 1
+
+
+async def test_router_exception_restores_context_and_next_call_rebuilds(auto_resource_env, monkeypatch):
+    """Even an exception after dispatch cannot leak batch state into a reused context."""
     env = auto_resource_env
     _seed_history(env)
     original = env.write_note("daily/2026-01-01/original.md", "[[resource/photo.png]]")
-    calls = _count_daily_lists(env, monkeypatch)
+    calls = _observe_history(env, monkeypatch)
     step = env.processor(_model(), routed=True)
     changes = [{"change": "deleted", "path": "resource/photo.png"}]
     context = RuntimeContext(changes=changes, user_value="preserve")
     context_keys = set(context.data)
-    # Interrupt the router after a real processor invocation to test scope cleanup.
     dispatch = step._dispatch_processor  # pylint: disable=protected-access
 
     async def dispatch_then_fail(*args, **kwargs):
@@ -457,78 +509,55 @@ async def test_router_dispatch_exception_cleans_lookup_scope(auto_resource_env,
     assert response.metadata["results"][0]["metadata"]["path"] == "daily/2026-01-02/recreated.md"
     assert not recreated.exists()
     assert set(context.data) == context_keys
-    assert {day: calls[day] for day in _HISTORY_DAYS} == dict.fromkeys(_HISTORY_DAYS, 2)
+    _assert_history_reads(calls, 2)
 
 
-async def _lookup_day(step, context, source="resource/photo.png"):
-    """Exercise the lookup helper inside an explicitly shared router-like scope."""
-    step.context = context
-    return await step._find_loose_resource_day(source)  # pylint: disable=protected-access
-
-
-@pytest.mark.parametrize("domain", ["workspace", "daily_dir", "file_store"])
-async def test_shared_scope_separates_lookup_domains(domain, auto_resource_env, monkeypatch):
-    """A processor cannot reuse another workspace, daily layout, or store's cached history."""
-    env = auto_resource_env
-    env.write_note("daily/2026-01-01/original.md", "[[resource/photo.png]]")
-    calls = _count_daily_lists(env, monkeypatch)
-    first = env.processor(_model())
-    context = RuntimeContext()
-
-    with _resource_lookup_scope(context):
-        assert await _lookup_day(first, context) == "2026-01-01"
-        if domain == "workspace":
-            other_workspace = env.workspace / "other-workspace"
-            write_note(other_workspace / "daily/2026-01-02/other.md", "[[resource/photo.png]]")
-            other_app = make_app_context(other_workspace)
-            other_store = LocalFileStore(name="read-only-other", app_context=other_app, embedding_store="")
-
-            async def other_daily_list(**kwargs):
-                step = DailyListStep(app_context=other_app, file_store=other_store)
-                await step(**kwargs)
-                return step.context.response
-
-            other_app.jobs = {"daily_list": other_daily_list}
-            second = AutoImageResourceStep(app_context=other_app, file_store=other_store)
-        elif domain == "daily_dir":
-            env.write_note("other-daily/2026-01-02/other.md", "[[resource/photo.png]]")
-            monkeypatch.setattr(env.app_context.app_config, "daily_dir", "other-daily")
-            second = env.processor(_model())
-        else:
-            other_store = LocalFileStore(name="read-only-other", app_context=env.app_context, embedding_store="")
-            second = AutoImageResourceStep(app_context=env.app_context, file_store=other_store)
-
-        expected = "2026-01-01" if domain == "file_store" else "2026-01-02"
-        assert await _lookup_day(second, context) == expected
-        if domain == "file_store":
-            assert calls["2026-01-01"] == 2
-
-    assert not context.data
-
-
-@pytest.mark.parametrize("change", ["added", "deleted"])
-async def test_mutations_invalidate_other_warmed_store_lookup(change, auto_resource_env, monkeypatch):
-    """Separate store indexes of one physical workspace all observe this batch's note mutations."""
+@pytest.mark.parametrize("routed", [False, True], ids=["image", "unified-router"])
+async def test_cancelled_invocation_restores_context_and_rebuilds(routed, auto_resource_env, monkeypatch):
+    """Cancel at the model boundary after lookup, without mocking the resource processing path."""
     env = auto_resource_env
     _seed_history(env)
     env.write_binary("resource/photo.png", image_bytes())
-    if change == "deleted":
-        env.write_note("daily/2026-01-01/caption.md", "[[resource/photo.png]]")
-    calls = _count_daily_lists(env, monkeypatch)
-    other_store = LocalFileStore(name="read-only-observer", app_context=env.app_context, embedding_store="")
-    observer = AutoImageResourceStep(app_context=env.app_context, file_store=other_store)
-    writer = env.processor(_model())
-    context = RuntimeContext(changes=[{"change": change, "path": "resource/photo.png"}])
-    previous = "2026-01-01" if change == "deleted" else None
+    original = env.write_note("daily/2026-01-01/original.md", "[[resource/photo.png]]")
+    before = original.read_bytes()
+    calls = _observe_history(env, monkeypatch)
+    entered = asyncio.Event()
 
-    with _resource_lookup_scope(context):
-        assert await _lookup_day(observer, context) == previous
-        assert await _lookup_day(writer, context) == previous
-        with patch.object(BaseAutoResourceStep, "_today", return_value="2026-01-01"):
-            response = await writer(context)
-        assert response.success is True
-        assert response.metadata["results"][0]["metadata"]["action"] == change
-        expected = None if change == "deleted" else "2026-01-01"
-        assert await _lookup_day(observer, context) == expected
+    # The inherited structured method deliberately raises to exercise plain-call fallback.
+    class WaitingModel(FakeVisionModel):  # pylint: disable=abstract-method
+        """Keep the image request pending until the invocation is cancelled."""
 
-    assert {day: calls[day] for day in _HISTORY_DAYS} == dict.fromkeys(_HISTORY_DAYS, 2)
+        async def __call__(self, messages, **kwargs):
+            del messages, kwargs
+            entered.set()
+            await asyncio.Event().wait()
+
+    step = env.processor(WaitingModel("unused"), routed=routed)
+    changes = [{"change": "modified", "path": "resource/photo.png"}]
+    context = RuntimeContext(changes=changes, user_value="preserve")
+    context_keys = set(context.data)
+    task = asyncio.create_task(step(context))
+    try:
+        await asyncio.wait_for(entered.wait(), timeout=5)
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+    finally:
+        if not task.done():
+            task.cancel()
+        await asyncio.gather(task, return_exceptions=True)
+
+    assert original.read_bytes() == before
+    assert set(context.data) == context_keys
+    assert context["changes"] is changes
+    assert context["user_value"] == "preserve"
+    original.unlink()
+    recreated = env.write_note("daily/2026-01-02/recreated.md", "[[resource/photo.png]]")
+    context["changes"] = [{"change": "deleted", "path": "resource/photo.png"}]
+    response = await step(context)
+
+    assert response.success is True
+    assert response.metadata["results"][0]["metadata"]["path"] == "daily/2026-01-02/recreated.md"
+    assert not recreated.exists()
+    assert set(context.data) == context_keys
+    _assert_history_reads(calls, 2)
diff --git a/tests/unit/test_auto_resource_concurrent_lookup.py b/tests/unit/test_auto_resource_concurrent_lookup.py
deleted file mode 100644
index 2c02455b..00000000
--- a/tests/unit/test_auto_resource_concurrent_lookup.py
+++ /dev/null
@@ -1,324 +0,0 @@
-"""Concurrent resource invocations must not reuse another batch's stale ownership."""
-
-import asyncio
-from contextlib import asynccontextmanager
-
-import pytest
-
-from reme.components.runtime_context import RuntimeContext
-
-from .auto_resource_test_support import FakeVisionModel, caption_json, image_bytes
-
-pytest_plugins = ("unit.auto_resource_test_plugin",)
-pytestmark = pytest.mark.asyncio
-
-
-def _processor(env, day, model=None):
-    """Use real file jobs and a deterministic model with no external requests."""
-    model = model or FakeVisionModel(caption_json("caption", "A sample image", "A red square."))
-    step = env.processor(model)
-    step._today = lambda: day  # pylint: disable=protected-access
-    return step
-
-
-async def _stop_task(task):
-    """Never leave a blocked consumer behind when an assertion fails."""
-    if not task.done():
-        task.cancel()
-    await asyncio.gather(task, return_exceptions=True)
-
-
-def _assert_no_active_batches(env):
-    """Lookup state stays invocation-scoped, with no Application cache or subscriptions."""
-    assert env.app_context.metadata == {}
-
-
-@asynccontextmanager
-async def _paused_between_resources(env, monkeypatch, consumer, *, change="deleted"):
-    """Pause a two-item batch after its gate lookup; yield an explicit resume operation."""
-    ready, resume = asyncio.Event(), asyncio.Event()
-    original_delete = consumer._handle_delete  # pylint: disable=protected-access
-
-    async def pause_gate(file_path, date_str, note_stem):
-        if file_path == "resource/gate.png":
-            ready.set()
-            await resume.wait()
-        await original_delete(file_path, date_str, note_stem)
-
-    monkeypatch.setattr(consumer, "_handle_delete", pause_gate)
-    task = asyncio.create_task(
-        env.run(
-            consumer,
-            [
-                {"change": "deleted", "path": "resource/gate.png"},
-                {"change": change, "path": "resource/photo.png"},
-            ],
-        ),
-    )
-
-    async def resume_consumer():
-        resume.set()
-        return await asyncio.wait_for(task, timeout=5)
-
-    try:
-        await asyncio.wait_for(ready.wait(), timeout=5)
-        yield resume_consumer
-    finally:
-        resume.set()
-        await _stop_task(task)
-
-
-@pytest.mark.parametrize("consumer_change", ["modified", "deleted"])
-@pytest.mark.parametrize("existing_day", [False, True], ids=["new-day", "existing-day"])
-async def test_another_batch_new_owner_is_seen_by_later_resource(
-    consumer_change,
-    existing_day,
-    auto_resource_env,
-    monkeypatch,
-):
-    """After B creates a card, A's later event must use B's day, not A's today."""
-    env = auto_resource_env
-    env.app_context.metadata = {}
-    env.write_note("daily/2025-12-01/archive.md", "[[resource/archive.txt]]")
-    if existing_day:
-        env.write_note("daily/2026-01-01/archive.md", "[[resource/older.txt]]")
-    env.write_binary("resource/photo.png", image_bytes())
-    consumer = _processor(env, "2026-01-02")
-    producer = _processor(env, "2026-01-01")
-
-    async with _paused_between_resources(env, monkeypatch, consumer, change=consumer_change) as resume:
-        created = await env.run(producer, [{"change": "added", "path": "resource/photo.png"}])
-        assert created.success is True
-        card_path = created.metadata["results"][0]["metadata"]["path"]
-        assert card_path.startswith("daily/2026-01-01/")
-        response = await resume()
-        result = response.metadata["results"][-1]
-        assert result["success"] is True
-        assert result["metadata"]["action"] == consumer_change
-        assert result["metadata"]["path"] == card_path
-        assert (env.workspace / card_path).exists() is (consumer_change != "deleted")
-        assert not list((env.workspace / "daily/2026-01-02").glob("*.md"))
-    _assert_no_active_batches(env)
-
-
-@pytest.mark.parametrize("read_phase", ["history", "dirty-day"])
-async def test_concurrent_write_during_lookup_is_seen_by_next_resource(
-    read_phase,
-    auto_resource_env,
-    monkeypatch,
-):
-    """A stale read keeps its pre-read fingerprint, so the next resource refreshes it."""
-    env = auto_resource_env
-    env.app_context.metadata = {}
-    env.write_note("daily/2026-01-01/gate.md", "[[resource/gate.png]]")
-    env.write_binary("resource/photo.png", image_bytes())
-    consumer = _processor(env, "2026-01-02")
-    producer = _processor(env, "2026-01-01")
-    ready, resume = asyncio.Event(), asyncio.Event()
-    daily_list = env.app_context.jobs["daily_list"]
-    consumer_reads = 0
-    # Deleting gate takes a history read and a targeted read; its dirty refresh is third.
-    blocked_read = 1 if read_phase == "history" else 3
-    task = None
-
-    async def pause_stale_response(**kwargs):
-        nonlocal consumer_reads
-        response = await daily_list(**kwargs)
-        if asyncio.current_task() is task and kwargs["date"] == "2026-01-01":
-            consumer_reads += 1
-            if consumer_reads == blocked_read:
-                ready.set()
-                await resume.wait()
-        return response
-
-    monkeypatch.setitem(env.app_context.jobs, "daily_list", pause_stale_response)
-    changes = [
-        {"change": "deleted", "path": "resource/missing.png"},
-        {"change": "deleted", "path": "resource/photo.png"},
-    ]
-    if read_phase == "dirty-day":
-        changes.insert(0, {"change": "deleted", "path": "resource/gate.png"})
-    task = asyncio.create_task(env.run(consumer, changes))
-    try:
-        await asyncio.wait_for(ready.wait(), timeout=5)
-        created = await env.run(producer, [{"change": "added", "path": "resource/photo.png"}])
-        assert created.success is True
-        card_path = created.metadata["results"][0]["metadata"]["path"]
-        resume.set()
-        response = await asyncio.wait_for(task, timeout=5)
-        assert response.metadata["results"][-2]["metadata"]["action"] == "skipped"
-        result = response.metadata["results"][-1]
-        assert result["success"] is True
-        assert result["metadata"]["action"] == "deleted"
-        assert result["metadata"]["path"] == card_path
-        assert not (env.workspace / card_path).exists()
-    finally:
-        resume.set()
-        await _stop_task(task)
-    _assert_no_active_batches(env)
-
-
-@pytest.mark.parametrize("cancel_phase", ["history", "processor"])
-async def test_cancelled_batch_does_not_leak_lookup_into_reused_context(
-    cancel_phase,
-    auto_resource_env,
-    monkeypatch,
-):
-    """Cancellation during a scan or a processor releases state before the next call."""
-    env = auto_resource_env
-    env.app_context.metadata = {}
-    env.write_note("daily/2026-01-01/archive.md", "[[resource/archive.txt]]")
-    step = _processor(env, "2026-01-02")
-    context = RuntimeContext(changes=[{"change": "deleted", "path": "resource/photo.png"}], marker="keep")
-    keys = set(context.data)
-    ready = asyncio.Event()
-    never = asyncio.Event()
-    daily_list = env.app_context.jobs["daily_list"]
-    original_delete = step._handle_delete  # pylint: disable=protected-access
-
-    async def block_history(**kwargs):
-        response = await daily_list(**kwargs)
-        ready.set()
-        await never.wait()
-        return response
-
-    async def block_processor(*args):
-        ready.set()
-        await never.wait()
-        await original_delete(*args)
-
-    with monkeypatch.context() as patcher:
-        if cancel_phase == "history":
-            patcher.setitem(env.app_context.jobs, "daily_list", block_history)
-        else:
-            patcher.setattr(step, "_handle_delete", block_processor)
-        task = asyncio.create_task(step(context))
-        try:
-            await asyncio.wait_for(ready.wait(), timeout=5)
-            task.cancel()
-            with pytest.raises(asyncio.CancelledError):
-                await task
-        finally:
-            await _stop_task(task)
-    assert set(context.data) == keys
-    assert context["marker"] == "keep"
-    _assert_no_active_batches(env)
-    card = env.write_note("daily/2026-01-01/photo.md", "[[resource/photo.png]]")
-    response = await step(context)
-    assert response.success is True
-    assert response.metadata["results"][0]["metadata"]["action"] == "deleted"
-    assert not card.exists()
-    assert set(context.data) == keys
-    _assert_no_active_batches(env)
-
-
-async def test_cancelled_writer_invalidates_other_batch_after_partial_write(auto_resource_env, monkeypatch):
-    """Cancelling after a card is saved still exposes its owner to an overlapping reader."""
-    env = auto_resource_env
-    env.app_context.metadata = {}
-    env.write_note("daily/2025-12-01/archive.md", "[[resource/archive.txt]]")
-    env.write_binary("resource/photo.png", image_bytes())
-    consumer = _processor(env, "2026-01-02")
-    producer = _processor(env, "2026-01-01")
-    written = asyncio.Event()
-
-    async def pause_after_write(_day):
-        written.set()
-        await asyncio.Event().wait()
-
-    monkeypatch.setattr(producer, "_refresh_day_index", pause_after_write)
-
-    async with _paused_between_resources(env, monkeypatch, consumer) as resume:
-        producer_task = asyncio.create_task(env.run(producer, [{"change": "added", "path": "resource/photo.png"}]))
-        try:
-            await asyncio.wait_for(written.wait(), timeout=5)
-            card = env.workspace / "daily/2026-01-01/caption.md"
-            assert card.exists()
-            producer_task.cancel()
-            with pytest.raises(asyncio.CancelledError):
-                await producer_task
-            response = await resume()
-            result = response.metadata["results"][-1]
-            assert result["success"] is True
-            assert result["metadata"]["action"] == "deleted"
-            assert result["metadata"]["path"] == "daily/2026-01-01/caption.md"
-            assert not card.exists()
-        finally:
-            await _stop_task(producer_task)
-    _assert_no_active_batches(env)
-
-
-async def test_another_batch_deletion_removes_cached_day(auto_resource_env, monkeypatch):
-    """After another invocation deletes a card, a new card uses this invocation's day."""
-    env = auto_resource_env
-    env.app_context.metadata = {}
-    old_card = env.write_note("daily/2026-01-01/photo.md", "[[resource/photo.png]]")
-    env.write_binary("resource/photo.png", image_bytes())
-    consumer = _processor(env, "2026-01-02")
-    producer = _processor(env, "2026-01-01")
-
-    async with _paused_between_resources(env, monkeypatch, consumer, change="added") as resume:
-        removed = await env.run(producer, [{"change": "deleted", "path": "resource/photo.png"}])
-        assert removed.success is True
-        assert not old_card.exists()
-        response = await resume()
-        result = response.metadata["results"][-1]
-        assert result["success"] is True
-        assert result["metadata"]["created"] is True
-        assert result["metadata"]["path"] == "daily/2026-01-02/caption.md"
-        assert not list((env.workspace / "daily/2026-01-01").glob("*.md"))
-    _assert_no_active_batches(env)
-
-
-@pytest.mark.parametrize("mutation", [False, True], ids=["no-op", "continuous-writes"])
-async def test_concurrent_activity_refreshes_only_on_next_resource(mutation, auto_resource_env, monkeypatch):
-    """Do not chase concurrent writes within a lookup; recheck them for the next resource."""
-    env = auto_resource_env
-    env.app_context.metadata = {}
-    card = env.write_note("daily/2026-01-01/photo.md", "[[resource/photo.png]]")
-    env.write_note("daily/2026-01-01/churn.md", "[[resource/churn.png]]")
-    env.write_binary("resource/churn.png", image_bytes())
-    consumer = _processor(env, "2026-01-02")
-    model = FakeVisionModel(caption_json("churn", "Changed image", "version 0"))
-    producer = _processor(env, "2026-01-01", model)
-    daily_list = env.app_context.jobs["daily_list"]
-    consumer_reads = 0
-    task = None
-
-    async def inject_concurrent_activity(**kwargs):
-        nonlocal consumer_reads
-        response = await daily_list(**kwargs)
-        if asyncio.current_task() is task and kwargs["date"] == "2026-01-01":
-            consumer_reads += 1
-            assert consumer_reads <= 3, "lookup retried instead of leaving freshness to the next resource"
-            if mutation:
-                model.text = caption_json("churn", "Changed image", f"version {consumer_reads}")
-                changes = [{"change": "modified", "path": "resource/churn.png"}]
-            else:
-                changes = [{"change": "deleted", "path": "resource/missing.png"}]
-            other = await asyncio.create_task(env.run(producer, changes))
-            assert other.success is True
-            assert other.metadata["modified"] is mutation
-        return response
-
-    monkeypatch.setitem(env.app_context.jobs, "daily_list", inject_concurrent_activity)
-    task = asyncio.create_task(
-        env.run(
-            consumer,
-            [
-                {"change": "deleted", "path": "resource/missing.png"},
-                {"change": "deleted", "path": "resource/photo.png"},
-            ],
-        ),
-    )
-    try:
-        response = await asyncio.wait_for(task, timeout=5)
-        assert response.success is True
-        assert [item["metadata"]["action"] for item in response.metadata["results"]] == ["skipped", "deleted"]
-        # One history scan and a targeted check; only real writes require a
-        # changed-day refresh when the second resource starts its lookup.
-        assert consumer_reads == (3 if mutation else 2)
-        assert not card.exists()
-    finally:
-        await _stop_task(task)
-    _assert_no_active_batches(env)
diff --git a/tests/unit/test_auto_resource_external_lookup.py b/tests/unit/test_auto_resource_external_lookup.py
deleted file mode 100644
index f10f6434..00000000
--- a/tests/unit/test_auto_resource_external_lookup.py
+++ /dev/null
@@ -1,280 +0,0 @@
-"""Ordinary writers must remain visible to a live resource ownership lookup."""
-
-import asyncio
-import os
-import sys
-
-import pytest
-
-from reme.components.runtime_context import RuntimeContext
-
-from .auto_resource_test_support import FakeVisionModel, caption_json, image_bytes
-
-pytest_plugins = ("unit.auto_resource_test_plugin",)
-pytestmark = pytest.mark.asyncio
-
-_SOURCE = "resource/photo.png"
-_ORIGINAL_CARD = "daily/2026-01-01/photo.md"
-
-
-def _note_text(source):
-    """Build a normal note without calling the resource processor."""
-    provenance = f"source_resource: '[[{source}]]'\n" if source else ""
-    return f"---\nname: photo\ndescription: external note\n{provenance}---\nExternal note body.\n"
-
-
-async def _after_history_edit(env, monkeypatch, edit, *, change="deleted"):
-    """Pause after a harmless first lookup; complete an external write before the next item."""
-    env.app_context.metadata = {}
-    env.write_note("daily/2025-12-01/archive.md", "[[resource/archive.txt]]")
-    env.write_binary(_SOURCE, image_bytes())
-    model = FakeVisionModel(caption_json("photo", "Updated image", "A red square."))
-    step = env.processor(model)
-    step._today = lambda: "2026-01-02"  # pylint: disable=protected-access
-    ready, resume = asyncio.Event(), asyncio.Event()
-    original_delete = step._handle_delete  # pylint: disable=protected-access
-
-    async def pause_gate(file_path, date_str, note_stem):
-        if file_path == "resource/gate.png":
-            ready.set()
-            await resume.wait()
-        await original_delete(file_path, date_str, note_stem)
-
-    monkeypatch.setattr(step, "_handle_delete", pause_gate)
-    context = RuntimeContext(
-        changes=[
-            {"change": "deleted", "path": "resource/gate.png"},
-            {"change": change, "path": _SOURCE},
-        ],
-        marker="preserve",
-    )
-    original_keys = set(context.data)
-    task = asyncio.create_task(step(context))
-    try:
-        await asyncio.wait_for(ready.wait(), timeout=5)
-        await edit()
-        resume.set()
-        response = await asyncio.wait_for(task, timeout=5)
-    finally:
-        resume.set()
-        if not task.done():
-            task.cancel()
-        await asyncio.gather(task, return_exceptions=True)
-
-    assert set(context.data) == original_keys
-    assert context["marker"] == "preserve"
-    results = response.metadata["results"]
-    assert [item["path"] for item in results] == ["resource/gate.png", _SOURCE]
-    assert results[0]["metadata"]["action"] == "skipped"
-    return response, results[1]
-
-
-@pytest.mark.parametrize("change", ["modified", "deleted"])
-@pytest.mark.parametrize("existing_day", [False, True], ids=["new-day", "existing-day"])
-async def test_ordinary_write_job_adds_owner_during_resource_batch(
-    change,
-    existing_day,
-    auto_resource_env,
-    monkeypatch,
-):
-    """A normal write job can create the original card after history was cached."""
-    env = auto_resource_env
-    if existing_day:
-        env.write_note("daily/2026-01-01/unrelated.md", "[[resource/other.txt]]")
-
-    async def write_card():
-        response = await env.app_context.jobs["write"](
-            path=_ORIGINAL_CARD,
-            name="photo",
-            description="Written through the public write job",
-            content="An externally created image card.",
-            metadata={"source_resource": f"[[{_SOURCE}]]"},
-        )
-        assert response.success is True
-
-    response, result = await _after_history_edit(env, monkeypatch, write_card, change=change)
-
-    assert response.success is True
-    assert result["metadata"]["action"] == change
-    assert result["metadata"]["path"] == _ORIGINAL_CARD
-    assert result["metadata"]["index"]["date"] == "2026-01-01"
-    assert (env.workspace / _ORIGINAL_CARD).exists() is (change == "modified")
-    if change == "modified":
-        assert result["metadata"]["created"] is False
-    assert not list((env.workspace / "daily/2026-01-02").glob("*.md"))
-
-
-@pytest.mark.parametrize("restore_mtime", [False, True], ids=["add-provenance", "same-size-restored-mtime"])
-async def test_in_place_frontmatter_change_is_seen_without_directory_change(
-    restore_mtime,
-    auto_resource_env,
-    monkeypatch,
-):
-    """Ownership edits need file metadata checks, including ctime when mtime is restored."""
-    env = auto_resource_env
-    original = _note_text("resource/other.png" if restore_mtime else None)
-    card = env.write_binary(_ORIGINAL_CARD, original.encode())
-
-    async def edit_card():
-        before = card.stat()
-        directory_before = card.parent.stat()
-        replacement = _note_text(_SOURCE).encode()
-        if restore_mtime:
-            assert len(replacement) == before.st_size
-        card.write_bytes(replacement)
-        if restore_mtime:
-            os.utime(card, ns=(before.st_atime_ns, before.st_mtime_ns))
-            after = card.stat()
-            assert after.st_size == before.st_size
-            assert after.st_mtime_ns == before.st_mtime_ns
-            if after.st_ctime_ns == before.st_ctime_ns:
-                pytest.skip("Filesystem does not expose this edit through a distinct ctime")
-        assert card.parent.stat().st_mtime_ns == directory_before.st_mtime_ns
-
-    response, result = await _after_history_edit(env, monkeypatch, edit_card)
-
-    assert response.success is True
-    assert result["metadata"]["action"] == "deleted"
-    assert result["metadata"]["path"] == _ORIGINAL_CARD
-    assert not card.exists()
-
-
-@pytest.mark.parametrize("new_source", [None, "resource/other.png"], ids=["remove-owner", "change-owner"])
-async def test_external_owner_removal_stops_reusing_previous_date(new_source, auto_resource_env, monkeypatch):
-    """A card no longer owned by this resource is preserved; a replacement uses today."""
-    env = auto_resource_env
-    card = env.write_binary(_ORIGINAL_CARD, _note_text(_SOURCE).encode())
-    replacement = _note_text(new_source).encode()
-
-    async def edit_card():
-        card.write_bytes(replacement)
-
-    response, result = await _after_history_edit(env, monkeypatch, edit_card, change="modified")
-
-    assert response.success is True
-    assert result["metadata"]["created"] is True
-    assert result["metadata"]["path"] == "daily/2026-01-02/photo.md"
-    assert result["metadata"]["index"]["date"] == "2026-01-02"
-    assert card.read_bytes() == replacement
-
-
-@pytest.mark.parametrize("restore_mtime", [False, True], ids=["normal-save", "preserved-mtime"])
-async def test_atomic_replacement_refreshes_owner(restore_mtime, auto_resource_env, monkeypatch):
-    """Replacing a same-size file must not preserve its previous ownership in the cache."""
-    env = auto_resource_env
-    card = env.write_binary(_ORIGINAL_CARD, _note_text("resource/other.png").encode())
-    replacement = card.with_suffix(".tmp")
-
-    async def replace_card():
-        before = card.stat()
-        content = _note_text(_SOURCE).encode()
-        assert len(content) == before.st_size
-        replacement.write_bytes(content)
-        if restore_mtime:
-            os.utime(replacement, ns=(before.st_atime_ns, before.st_mtime_ns))
-        replacement.replace(card)
-        after = card.stat()
-        assert after.st_size == before.st_size
-        if restore_mtime:
-            assert after.st_mtime_ns == before.st_mtime_ns
-        if (after.st_mtime_ns, after.st_ctime_ns) == (before.st_mtime_ns, before.st_ctime_ns):
-            pytest.skip("Filesystem does not expose this replacement through a distinct mtime or ctime")
-
-    response, result = await _after_history_edit(env, monkeypatch, replace_card)
-
-    assert response.success is True
-    assert result["metadata"]["action"] == "deleted"
-    assert result["metadata"]["path"] == _ORIGINAL_CARD
-    assert not card.exists()
-    assert not replacement.exists()
-
-
-@pytest.mark.parametrize("operation", ["move", "delete"])
-async def test_external_move_or_delete_updates_cached_owner(operation, auto_resource_env, monkeypatch):
-    """Manual moves and deletion cannot leave a cached historical owner behind."""
-    env = auto_resource_env
-    card = env.write_binary(_ORIGINAL_CARD, _note_text(_SOURCE).encode())
-    destination = env.workspace / "daily/2026-01-03/moved.md"
-
-    async def edit_card():
-        if operation == "move":
-            destination.parent.mkdir(parents=True)
-            card.rename(destination)
-        else:
-            card.unlink()
-
-    change = "deleted" if operation == "move" else "modified"
-    response, result = await _after_history_edit(env, monkeypatch, edit_card, change=change)
-
-    assert response.success is True
-    assert result["metadata"]["action"] == change
-    expected = "daily/2026-01-03/moved.md" if operation == "move" else "daily/2026-01-02/photo.md"
-    assert result["metadata"]["path"] == expected
-    assert not card.exists()
-    if operation == "move":
-        assert not destination.exists()
-    else:
-        assert result["metadata"]["created"] is True
-
-
-async def test_external_second_owner_fails_closed_before_mutating_cards(auto_resource_env, monkeypatch):
-    """A second exact owner added between resources must retain the duplicate-owner safeguard."""
-    env = auto_resource_env
-    card = env.write_binary(_ORIGINAL_CARD, _note_text(_SOURCE).encode())
-    before = card.read_bytes()
-    duplicate = env.workspace / "daily/2026-01-03/duplicate.md"
-
-    async def add_duplicate():
-        duplicate.parent.mkdir(parents=True)
-        duplicate.write_bytes(before)
-
-    response, result = await _after_history_edit(env, monkeypatch, add_duplicate)
-
-    assert response.success is False
-    assert result["success"] is False
-    assert result["metadata"]["action"] == "failed"
-    assert result["metadata"]["modified"] is False
-    assert "Multiple daily resource notes claim resource/photo.png" in result["metadata"]["error"]
-    assert _ORIGINAL_CARD in result["metadata"]["error"]
-    assert "daily/2026-01-03/duplicate.md" in result["metadata"]["error"]
-    assert card.read_bytes() == duplicate.read_bytes() == before
-
-
-async def test_separate_process_write_is_visible_inside_resource_batch(auto_resource_env, monkeypatch):
-    """A writer without this Application or event subscriptions can establish ownership."""
-    env = auto_resource_env
-    card = env.workspace / _ORIGINAL_CARD
-
-    async def write_in_child():
-        code = (
-            "from pathlib import Path\n"
-            "import sys\n"
-            "path = Path(sys.argv[1])\n"
-            "path.parent.mkdir(parents=True, exist_ok=True)\n"
-            "path.write_text(sys.argv[2], encoding='utf-8')\n"
-        )
-        process = await asyncio.create_subprocess_exec(
-            sys.executable,
-            "-c",
-            code,
-            str(card),
-            _note_text(_SOURCE),
-            cwd=env.workspace,
-            stdout=asyncio.subprocess.PIPE,
-            stderr=asyncio.subprocess.PIPE,
-        )
-        try:
-            _stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=5)
-        finally:
-            if process.returncode is None:
-                process.kill()
-                await process.wait()
-        assert process.returncode == 0, stderr.decode()
-        assert card.is_file()
-
-    response, result = await _after_history_edit(env, monkeypatch, write_in_child)
-
-    assert response.success is True
-    assert result["metadata"]["action"] == "deleted"
-    assert result["metadata"]["path"] == _ORIGINAL_CARD
-    assert not card.exists()
diff --git a/tests/unit/test_auto_resource_shared_lookup.py b/tests/unit/test_auto_resource_shared_lookup.py
deleted file mode 100644
index 578c3f4d..00000000
--- a/tests/unit/test_auto_resource_shared_lookup.py
+++ /dev/null
@@ -1,104 +0,0 @@
-"""Every resource modality consumes the same Base-managed ownership lookup."""
-
-from collections import Counter
-from unittest.mock import patch
-
-import pytest
-
-from reme.components import R
-from reme.steps.evolve.auto_resource import AutoResourceStep
-from reme.steps.evolve.auto_text_resource import AutoTextResourceStep
-from reme.steps.evolve.base_auto_resource import BaseAutoResourceStep
-
-from .auto_resource_test_support import FakeAudioResourceStep, FakeVisionModel
-
-pytest_plugins = ("unit.auto_resource_test_plugin",)
-pytestmark = pytest.mark.asyncio
-
-_HISTORY_DAYS = ("2025-12-01", "2025-12-02", "2025-12-03")
-
-
-def _count_history(env, monkeypatch):
-    """Count the real daily_list jobs on history untouched by resource processing."""
-    for day in _HISTORY_DAYS:
-        env.write_note(f"daily/{day}/archive.md", f"[[resource/archive-{day}.txt]]")
-    calls = Counter()
-    daily_list = env.app_context.jobs["daily_list"]
-
-    async def counted(**kwargs):
-        calls[kwargs["date"]] += 1
-        return await daily_list(**kwargs)
-
-    monkeypatch.setitem(env.app_context.jobs, "daily_list", counted)
-    env.app_context.metadata = {}
-    return calls
-
-
-@pytest.mark.parametrize("audio_first", [True, False], ids=["audio-first", "image-first"])
-@pytest.mark.parametrize("invalid_audio", [False, True], ids=["valid-batch", "one-invalid-resource"])
-async def test_third_modality_shares_history_regardless_of_processor_order(
-    audio_first,
-    invalid_audio,
-    auto_resource_env,
-    monkeypatch,
-):
-    """A new Step inherits the common lookup without any image-specific dependency."""
-    env = auto_resource_env
-    calls = _count_history(env, monkeypatch)
-    notes = [
-        env.write_note(f"daily/2026-01-01/{stem}.md", f"[[resource/{stem}.{suffix}]]")
-        for stem, suffix in (("report", "txt"), ("photo", "png"), ("recording", "wav"))
-    ]
-    env.app_context.registry = R.copy()
-    env.app_context.registry.add("test_audio_resource_step", FakeAudioResourceStep, owner=__name__)
-    specific = ["test_audio_resource_step", "auto_image_resource_step"]
-    if not audio_first:
-        specific.reverse()
-    model = FakeVisionModel("must not be called for deletion")
-    router = AutoResourceStep(
-        app_context=env.app_context,
-        file_store=env.file_store,
-        as_llm=model,
-        dispatch_steps=[*specific, "auto_text_resource_step"],
-    )
-    changes = [
-        {"change": "deleted", "path": f"resource/{stem}.{suffix}"}
-        for stem, suffix in (("report", "txt"), ("photo", "png"), ("recording", "wav"))
-    ]
-    if invalid_audio:
-        changes.insert(0, {"change": "deleted", "path": "resource/../private.wav"})
-    with patch.object(BaseAutoResourceStep, "_today", return_value="2026-01-02"):
-        response = await env.run(router, changes)
-
-    results = response.metadata["results"]
-    assert response.success is not invalid_audio
-    assert [result["path"] for result in results] == [change["path"] for change in changes]
-    assert all(result["metadata"]["action"] == "deleted" for result in results[int(invalid_audio) :])
-    assert all(not note.exists() for note in notes)
-    assert {day: calls[day] for day in _HISTORY_DAYS} == dict.fromkeys(_HISTORY_DAYS, 1)
-    assert not model.calls
-    if invalid_audio:
-        assert results[0]["metadata"]["action"] == "failed"
-        assert results[0]["metadata"]["modified"] is False
-
-
-@pytest.mark.parametrize("routed", [False, True], ids=["text-step", "text-only-router"])
-@pytest.mark.parametrize("dated", [False, True], ids=["loose-root", "dated-resource"])
-async def test_text_only_batch_uses_common_lookup_without_image_step(routed, dated, auto_resource_env, monkeypatch):
-    """Text does not depend on an image processor priming the cache; dated files stay direct."""
-    env = auto_resource_env
-    calls = _count_history(env, monkeypatch)
-    env.app_context.registry = R.copy()
-    prefix = "resource/2026-01-01" if dated else "resource"
-    notes = [env.write_note(f"daily/2026-01-01/report-{i}.md", f"[[{prefix}/report-{i}.txt]]") for i in range(3)]
-    options = {"app_context": env.app_context, "file_store": env.file_store}
-    step = (
-        AutoResourceStep(**options, dispatch_steps=["auto_text_resource_step"])
-        if routed
-        else AutoTextResourceStep(**options)
-    )
-    response = await env.run(step, [{"change": "deleted", "path": f"{prefix}/report-{i}.txt"} for i in range(3)])
-
-    assert response.success is True
-    assert all(not note.exists() for note in notes)
-    assert {day: calls[day] for day in _HISTORY_DAYS} == dict.fromkeys(_HISTORY_DAYS, 0 if dated else 1)
diff --git a/tests/unit/test_auto_resource_snapshot_lookup.py b/tests/unit/test_auto_resource_snapshot_lookup.py
deleted file mode 100644
index 9818918a..00000000
--- a/tests/unit/test_auto_resource_snapshot_lookup.py
+++ /dev/null
@@ -1,427 +0,0 @@
-"""Filesystem snapshots keep ownership lookups fresh without re-parsing stable history."""
-
-import asyncio
-import errno
-from collections import Counter
-from contextlib import nullcontext
-from types import SimpleNamespace
-from unittest.mock import Mock
-
-import pytest
-
-from reme.schema import Response
-from reme.steps.evolve import base_auto_resource
-
-from .auto_resource_test_support import FakeVisionModel, caption_json, write_note
-
-pytest_plugins = ("unit.auto_resource_test_plugin",)
-pytestmark = pytest.mark.asyncio
-
-
-def _processor(env):
-    """Use real file jobs with no model requests on these deletion paths."""
-    model = FakeVisionModel(caption_json("caption", "A sample image", "A red square."))
-    step = env.processor(model)
-    step._today = lambda: "2026-01-03"  # pylint: disable=protected-access
-    return step
-
-
-async def _stop_task(task):
-    """Clean up an event-blocked consumer even when an assertion fails."""
-    if not task.done():
-        task.cancel()
-    await asyncio.gather(task, return_exceptions=True)
-
-
-def _symlink(link, target, *, directory=False):
-    """Skip only environments that cannot create test symlinks."""
-    try:
-        link.symlink_to(target, target_is_directory=directory)
-    except NotImplementedError as exc:
-        pytest.skip(f"Symlinks unavailable: {exc}")
-    except OSError as exc:
-        if exc.errno in {errno.EPERM, errno.EACCES, errno.ENOSYS, errno.EOPNOTSUPP}:
-            pytest.skip(f"Symlinks unavailable: {exc}")
-        raise
-
-
-@pytest.mark.parametrize("read_phase", ["history", "dirty-day", "new-day"])
-async def test_external_change_during_lookup_is_seen_by_next_resource(read_phase, auto_resource_env, monkeypatch):
-    """Keep the pre-read version so an external write is detected by the next resource."""
-    env = auto_resource_env
-    env.app_context.metadata = {}
-    env.write_note("daily/2026-01-01/gate.md", "[[resource/gate.png]]")
-    consumer = _processor(env)
-    daily_list = env.app_context.jobs["daily_list"]
-    ready, resume = asyncio.Event(), asyncio.Event()
-    reads = 0
-    # Deleting gate takes a history read and a targeted read; its dirty refresh is third.
-    blocked_read = 3 if read_phase == "dirty-day" else 1
-
-    async def pause_stale_response(**kwargs):
-        nonlocal reads
-        response = await daily_list(**kwargs)
-        if kwargs["date"] == "2026-01-01":
-            reads += 1
-            if reads == blocked_read:
-                ready.set()
-                await resume.wait()
-        return response
-
-    monkeypatch.setitem(env.app_context.jobs, "daily_list", pause_stale_response)
-    changes = [
-        {"change": "deleted", "path": "resource/missing.png"},
-        {"change": "deleted", "path": "resource/photo.png"},
-    ]
-    if read_phase == "dirty-day":
-        changes.insert(0, {"change": "deleted", "path": "resource/gate.png"})
-    task = asyncio.create_task(env.run(consumer, changes))
-    try:
-        await asyncio.wait_for(ready.wait(), timeout=5)
-        day = "2026-01-02" if read_phase == "new-day" else "2026-01-01"
-        card = env.write_note(f"daily/{day}/photo.md", "[[resource/photo.png]]")
-        resume.set()
-        response = await asyncio.wait_for(task, timeout=5)
-        assert response.metadata["results"][-2]["metadata"]["action"] == "skipped"
-        result = response.metadata["results"][-1]
-        assert result["success"] is True
-        assert result["metadata"]["action"] == "deleted"
-        assert result["metadata"]["path"] == f"daily/{day}/photo.md"
-        assert not card.exists()
-    finally:
-        resume.set()
-        await _stop_task(task)
-
-
-@pytest.mark.parametrize("alias_change", ["retarget", "missing-target"])
-@pytest.mark.parametrize("alias_count", [1, 2], ids=["single-owner", "duplicate-owners"])
-async def test_external_alias_change_refreshes_all_logical_dates(
-    alias_change,
-    alias_count,
-    auto_resource_env,
-    monkeypatch,
-):
-    """Alias identities and newly available targets are checked on later resources."""
-    env = auto_resource_env
-    env.app_context.metadata = {}
-    old_target = env.workspace / "storage/old"
-    new_target = env.workspace / "storage/new"
-    env.write_note("storage/old/archive.md", "[[resource/archive.txt]]")
-    if alias_change == "retarget":
-        card = env.write_note("storage/new/photo.md", "[[resource/photo.png]]")
-    else:
-        card = new_target / "photo.md"
-    daily = env.workspace / "daily"
-    daily.mkdir()
-    aliases = [daily / f"2026-01-0{index}" for index in range(1, alias_count + 1)]
-    for alias in aliases:
-        _symlink(alias, old_target if alias_change == "retarget" else new_target, directory=True)
-    consumer = _processor(env)
-    ready, resume = asyncio.Event(), asyncio.Event()
-    original_delete = consumer._handle_delete  # pylint: disable=protected-access
-
-    async def pause_after_lookup(file_path, date_str, note_stem):
-        if file_path == "resource/gate.png":
-            ready.set()
-            await resume.wait()
-        await original_delete(file_path, date_str, note_stem)
-
-    monkeypatch.setattr(consumer, "_handle_delete", pause_after_lookup)
-    task = asyncio.create_task(
-        env.run(
-            consumer,
-            [
-                {"change": "deleted", "path": "resource/gate.png"},
-                {"change": "deleted", "path": "resource/photo.png"},
-            ],
-        ),
-    )
-    try:
-        await asyncio.wait_for(ready.wait(), timeout=5)
-        if alias_change == "retarget":
-            for alias in aliases:
-                alias.unlink()
-                _symlink(alias, new_target, directory=True)
-        else:
-            card = env.write_note("storage/new/photo.md", "[[resource/photo.png]]")
-        resume.set()
-        response = await asyncio.wait_for(task, timeout=5)
-        result = response.metadata["results"][-1]
-        if alias_count == 1:
-            assert result["success"] is True
-            assert result["metadata"]["action"] == "deleted"
-            assert result["metadata"]["path"] == "daily/2026-01-01/photo.md"
-            assert not card.exists()
-        else:
-            assert result["success"] is False
-            assert "Multiple daily resource notes claim" in result["metadata"]["error"]
-            assert card.exists()
-        assert (old_target / "archive.md").exists()
-    finally:
-        resume.set()
-        await _stop_task(task)
-
-
-@pytest.mark.parametrize("failure_phase", ["initial", "refresh"])
-async def test_snapshot_failure_remains_retryable(failure_phase, auto_resource_env, monkeypatch):
-    """A stat/scan failure fails one resource without committing an incomplete view."""
-    env = auto_resource_env
-    env.app_context.metadata = {}
-    env.write_note("daily/2026-01-01/gate.md", "[[resource/gate.png]]")
-    card = env.write_note("daily/2026-01-01/photo.md", "[[resource/photo.png]]")
-    consumer = _processor(env)
-    snapshot = consumer._resource_note_snapshot  # pylint: disable=protected-access
-    original_delete = consumer._handle_delete  # pylint: disable=protected-access
-    warmed = False
-    failed = False
-
-    def fail_snapshot_once(*args, **kwargs):
-        nonlocal failed
-        if not failed and (failure_phase == "initial" or warmed):
-            failed = True
-            raise OSError("temporary snapshot scan failure")
-        return snapshot(*args, **kwargs)
-
-    async def mark_warmed(file_path, date_str, note_stem):
-        nonlocal warmed
-        await original_delete(file_path, date_str, note_stem)
-        if file_path == "resource/gate.png":
-            warmed = True
-
-    monkeypatch.setattr(consumer, "_resource_note_snapshot", fail_snapshot_once)
-    monkeypatch.setattr(consumer, "_handle_delete", mark_warmed)
-    changes = [
-        {"change": "deleted", "path": "resource/photo.png"},
-        {"change": "deleted", "path": "resource/photo.png"},
-    ]
-    if failure_phase == "refresh":
-        changes.insert(0, {"change": "deleted", "path": "resource/gate.png"})
-    response = await env.run(consumer, changes)
-    results = response.metadata["results"]
-    assert results[-2]["success"] is False
-    assert "temporary snapshot scan failure" in results[-2]["metadata"]["error"]
-    assert results[-1]["success"] is True
-    assert results[-1]["metadata"]["action"] == "deleted"
-    assert not card.exists()
-
-
-async def test_failed_changed_day_read_does_not_accept_new_fingerprint(auto_resource_env, monkeypatch):
-    """A failed refresh must not mark an externally added owner as already indexed."""
-    env = auto_resource_env
-    env.app_context.metadata = {}
-    env.write_note("daily/2026-01-01/gate.md", "[[resource/gate.png]]")
-    card = env.write_note("daily/2026-01-01/photo.md", "[[resource/other.png]]")
-    consumer = _processor(env)
-    original_delete = consumer._handle_delete  # pylint: disable=protected-access
-    daily_list = env.app_context.jobs["daily_list"]
-    changed = False
-    failed = False
-
-    async def change_owner_after_gate(file_path, date_str, note_stem):
-        nonlocal changed
-        await original_delete(file_path, date_str, note_stem)
-        if file_path == "resource/gate.png":
-            env.write_note("daily/2026-01-01/photo.md", "[[resource/photo.png]]")
-            changed = True
-
-    async def fail_refresh_once(**kwargs):
-        nonlocal failed
-        if changed and not failed:
-            failed = True
-            return Response(success=False, answer="temporary changed-day read failure")
-        return await daily_list(**kwargs)
-
-    monkeypatch.setattr(consumer, "_handle_delete", change_owner_after_gate)
-    monkeypatch.setitem(env.app_context.jobs, "daily_list", fail_refresh_once)
-    response = await env.run(
-        consumer,
-        [
-            {"change": "deleted", "path": "resource/gate.png"},
-            {"change": "deleted", "path": "resource/photo.png"},
-            {"change": "deleted", "path": "resource/photo.png"},
-        ],
-    )
-    results = response.metadata["results"]
-    assert results[0]["success"] is True
-    assert results[1]["success"] is False
-    assert "temporary changed-day read failure" in results[1]["metadata"]["error"]
-    assert results[2]["success"] is True
-    assert results[2]["metadata"]["action"] == "deleted"
-    assert not card.exists()
-
-
-@pytest.mark.parametrize("irrelevant_edit", ["none", "non-markdown", "same-target-link"])
-async def test_unchanged_snapshots_do_not_reparse_historical_notes(irrelevant_edit, auto_resource_env, monkeypatch):
-    """Only changes to note contents or targets require another historical read."""
-    env = auto_resource_env
-    env.app_context.metadata = {}
-    days = ("2025-12-01", "2025-12-02")
-    for day in days:
-        for index in range(8):
-            env.write_note(f"daily/{day}/archive-{index}.md", f"[[resource/archive-{day}-{index}.txt]]")
-    step = _processor(env)
-    original_delete = step._handle_delete  # pylint: disable=protected-access
-    alias = env.workspace / f"daily/{days[0]}/alias.md"
-    target = env.workspace / "storage/archive.md"
-    if irrelevant_edit == "same-target-link":
-        env.write_note("storage/archive.md", "[[resource/archive.txt]]")
-        _symlink(alias, target)
-
-    async def edit_unrelated_metadata(file_path, date_str, note_stem):
-        await original_delete(file_path, date_str, note_stem)
-        if irrelevant_edit == "non-markdown":
-            (alias.parent / "unrelated.txt").write_text(file_path, encoding="utf-8")
-        elif irrelevant_edit == "same-target-link":
-            alias.unlink()
-            _symlink(alias, target)
-
-    monkeypatch.setattr(step, "_handle_delete", edit_unrelated_metadata)
-    daily_list = env.app_context.jobs["daily_list"]
-    calls = Counter()
-
-    async def counted_daily_list(**kwargs):
-        calls[kwargs["date"]] += 1
-        return await daily_list(**kwargs)
-
-    monkeypatch.setitem(env.app_context.jobs, "daily_list", counted_daily_list)
-    response = await env.run(
-        step,
-        [{"change": "deleted", "path": f"resource/missing-{index}.png"} for index in range(8)],
-    )
-    assert response.success is True
-    assert response.metadata["modified"] is False
-    assert response.metadata["processed"] == 8
-    assert {day: calls[day] for day in days} == dict.fromkeys(days, 1)
-
-
-async def test_note_link_retarget_refreshes_cached_ownership(auto_resource_env, monkeypatch):
-    """A different safe target is discovered by the next resource lookup."""
-    env = auto_resource_env
-    old_target = env.write_note("storage/old.md", "[[resource/other.png]]")
-    new_target = env.write_note("storage/new.md", "[[resource/photo.png]]")
-    day = env.workspace / "daily/2026-01-01"
-    day.mkdir(parents=True)
-    alias = day / "photo.md"
-    _symlink(alias, old_target)
-    step = _processor(env)
-    original_delete = step._handle_delete  # pylint: disable=protected-access
-
-    async def retarget_after_first_resource(file_path, date_str, note_stem):
-        await original_delete(file_path, date_str, note_stem)
-        if file_path == "resource/missing.png":
-            alias.unlink()
-            _symlink(alias, new_target)
-
-    monkeypatch.setattr(step, "_handle_delete", retarget_after_first_resource)
-    response = await env.run(
-        step,
-        [
-            {"change": "deleted", "path": "resource/missing.png"},
-            {"change": "deleted", "path": "resource/photo.png"},
-        ],
-    )
-    assert response.success is True
-    assert response.metadata["results"][0]["metadata"]["action"] == "skipped"
-    result = response.metadata["results"][1]["metadata"]
-    assert result["action"] == "deleted"
-    assert result["path"] == "daily/2026-01-01/photo.md"
-    assert old_target.exists()
-    assert new_target.exists()
-    assert not alias.is_symlink()
-
-
-@pytest.mark.parametrize("cycle", ["self", "mutual"])
-async def test_cyclic_note_links_do_not_block_unrelated_resource(cycle, auto_resource_env):
-    """Like daily_list, metadata scans skip cyclic links that are not readable files."""
-    env = auto_resource_env
-    card = env.write_note("daily/2026-01-01/photo.md", "[[resource/photo.png]]")
-    link = card.parent / "loop.md"
-    _symlink(link, "loop.md" if cycle == "self" else "other-loop.md")
-    if cycle == "mutual":
-        _symlink(card.parent / "other-loop.md", "loop.md")
-    response = await env.run(_processor(env), [{"change": "deleted", "path": "resource/photo.png"}])
-    assert response.success is True
-    assert response.metadata["results"][0]["metadata"]["action"] == "deleted"
-    assert not card.exists()
-    assert link.is_symlink()
-
-
-@pytest.mark.parametrize("cycle", ["self", "mutual"])
-async def test_cyclic_daily_links_do_not_block_unrelated_resource(cycle, auto_resource_env):
-    """Unresolvable date directories do not hide a normal historical owner."""
-    env = auto_resource_env
-    card = env.write_note("daily/2026-01-01/photo.md", "[[resource/photo.png]]")
-    link = card.parent.parent / "2025-12-01"
-    _symlink(link, "2025-12-01" if cycle == "self" else "2025-12-02", directory=True)
-    if cycle == "mutual":
-        _symlink(link.with_name("2025-12-02"), "2025-12-01", directory=True)
-    response = await env.run(_processor(env), [{"change": "deleted", "path": "resource/photo.png"}])
-    assert response.success is True
-    assert response.metadata["results"][0]["metadata"]["action"] == "deleted"
-    assert not card.exists()
-    assert link.is_symlink()
-
-
-@pytest.mark.parametrize("failure_site", ["scandir", "file-stat"])
-@pytest.mark.parametrize("error_number", [errno.ELOOP, errno.EACCES, errno.EIO], ids=["loop", "permission", "io"])
-async def test_snapshot_handles_only_loop_errors(failure_site, error_number, tmp_path, monkeypatch):
-    """Exercise late OS errors on every Python version, without bypassing safety failures."""
-    day = tmp_path / "daily" / "2026-01-01"
-    day.mkdir(parents=True)
-    (day / "valid.md").write_text("A normal note.", encoding="utf-8")
-    error = OSError(error_number, "snapshot probe failed")
-    with base_auto_resource.os.scandir(day) as entries:
-        valid_entries = list(entries)
-
-    with monkeypatch.context() as patch:
-        if failure_site == "scandir":
-            patch.setattr(base_auto_resource.os, "scandir", Mock(side_effect=error))
-        else:
-            unreadable = SimpleNamespace(
-                name="unreadable.md",
-                is_symlink=lambda: False,
-                stat=Mock(side_effect=error),
-            )
-            patch.setattr(
-                base_auto_resource.os,
-                "scandir",
-                lambda _: nullcontext(iter([unreadable, *valid_entries])),
-            )
-        if error_number != errno.ELOOP:
-            with pytest.raises(OSError) as caught:
-                base_auto_resource._snapshot_note_day(tmp_path, day)  # pylint: disable=protected-access
-            assert caught.value is error
-            return
-        snapshot = base_auto_resource._snapshot_note_day(tmp_path, day)  # pylint: disable=protected-access
-
-    if failure_site == "file-stat":
-        assert snapshot is not None
-        assert set(snapshot.files) == {"valid.md"}
-    else:
-        assert snapshot is None
-    assert (day / "valid.md").read_text(encoding="utf-8") == "A normal note."
-
-
-async def test_escaping_note_link_fails_before_daily_list(auto_resource_env, monkeypatch):
-    """Metadata inspection refuses an escaping Markdown target before parsing any notes."""
-    env = auto_resource_env
-    card = env.write_note("daily/2026-01-01/photo.md", "[[resource/photo.png]]")
-    outside = write_note(env.workspace.parent / "outside.md", "[[resource/other.png]]")
-    original = outside.read_bytes()
-    _symlink(card.parent / "escape.md", outside)
-    calls = 0
-    daily_list = env.app_context.jobs["daily_list"]
-
-    async def counted_daily_list(**kwargs):
-        nonlocal calls
-        calls += 1
-        return await daily_list(**kwargs)
-
-    monkeypatch.setitem(env.app_context.jobs, "daily_list", counted_daily_list)
-    response = await env.run(_processor(env), [{"change": "deleted", "path": "resource/photo.png"}])
-    result = response.metadata["results"][0]
-    assert result["success"] is False
-    assert "invalid resource note path" in result["metadata"]["error"]
-    assert calls == 0
-    assert card.exists()
-    assert outside.read_bytes() == original
diff --git a/tests/unit/test_daily_note_links.py b/tests/unit/test_daily_note_links.py
new file mode 100644
index 00000000..856c450b
--- /dev/null
+++ b/tests/unit/test_daily_note_links.py
@@ -0,0 +1,135 @@
+"""Daily-note scans preserve internal links without reading outside the workspace."""
+
+import errno
+from pathlib import Path
+
+import pytest
+
+from reme.steps.file_io import _daily_index
+
+DAY = "2026-01-01"
+
+
+def _symlink(link: Path, target: Path, *, directory: bool = False) -> None:
+    """Skip platforms where creating a test symlink is unavailable."""
+    try:
+        link.symlink_to(target, target_is_directory=directory)
+    except NotImplementedError as exc:
+        pytest.skip(f"Symlinks unavailable: {exc}")
+    except OSError as exc:
+        if exc.errno in {errno.EPERM, errno.EACCES, errno.ENOSYS, errno.EOPNOTSUPP}:
+            pytest.skip(f"Symlinks unavailable: {exc}")
+        raise
+
+
+@pytest.fixture(name="daily_workspace")
+def _daily_workspace(tmp_path):
+    """Keep note targets and an outside sibling under pytest's temporary directory."""
+    workspace = (tmp_path / "workspace").resolve()
+    day_dir = workspace / "daily" / DAY
+    day_dir.mkdir(parents=True)
+    (day_dir / "regular.md").write_text("---\nname: regular\n---\nBody\n", encoding="utf-8")
+    return workspace, day_dir
+
+
+@pytest.mark.parametrize("kind", ["internal", "escaping", "dangling", "self-loop", "mutual-loop"])
+def test_scan_note_links_keep_logical_paths_and_skip_unreadable_targets(kind, daily_workspace, monkeypatch):
+    """Unsafe or unreadable links must neither hide a valid note nor reach read_text."""
+    workspace, day_dir = daily_workspace
+    link = day_dir / "linked.md"
+    target = workspace / "target.md"
+    if kind == "escaping":
+        target = workspace.parent / "outside.md"
+    if kind in {"internal", "escaping"}:
+        target.write_text("---\nname: linked\nsource_resource: '[[resource/photo.png]]'\n---\nBody\n", encoding="utf-8")
+    elif kind == "self-loop":
+        target = link
+    elif kind == "mutual-loop":
+        target = day_dir / "other.md"
+        _symlink(link=target, target=link)
+    _symlink(link, target)
+
+    read_paths = []
+    original_read = Path.read_text
+
+    def record_read(path, *args, **kwargs):
+        read_paths.append(path)
+        return original_read(path, *args, **kwargs)
+
+    monkeypatch.setattr(Path, "read_text", record_read)
+    notes = _daily_index.scan_notes(workspace, DAY, "daily")
+
+    expected_paths = [f"daily/{DAY}/regular.md"]
+    if kind == "internal":
+        expected_paths.insert(0, f"daily/{DAY}/linked.md")
+        assert notes[0]["metadata"]["source_resource"] == "[[resource/photo.png]]"
+        assert target in read_paths
+    else:
+        assert read_paths == [day_dir / "regular.md"]
+    assert [note["path"] for note in notes] == expected_paths
+    assert link.is_symlink()
+
+
+@pytest.mark.parametrize("kind", ["internal", "escaping", "dangling", "self-loop", "mutual-loop"])
+def test_scan_date_directory_links(kind, daily_workspace, monkeypatch):
+    """A valid date alias retains its logical date; invalid aliases are not read."""
+    workspace, day_dir = daily_workspace
+    alias = day_dir.parent / "2026-01-02"
+    target = day_dir
+    if kind == "escaping":
+        target = workspace.parent / "outside-day"
+        target.mkdir()
+        (target / "outside.md").write_text("---\nname: outside\n---\nBody\n", encoding="utf-8")
+    elif kind == "dangling":
+        target = day_dir.parent / "missing"
+    elif kind == "self-loop":
+        target = alias
+    elif kind == "mutual-loop":
+        target = day_dir.parent / "2026-01-03"
+        _symlink(target, alias, directory=True)
+    _symlink(alias, target, directory=True)
+
+    read_paths = []
+    original_read = Path.read_text
+
+    def record_read(path, *args, **kwargs):
+        read_paths.append(path)
+        return original_read(path, *args, **kwargs)
+
+    monkeypatch.setattr(Path, "read_text", record_read)
+    notes = _daily_index.scan_notes(workspace, alias.name, "daily")
+
+    if kind == "internal":
+        assert notes == [{"path": "daily/2026-01-02/regular.md", "metadata": {"name": "regular"}}]
+        assert read_paths == [day_dir / "regular.md"]
+    else:
+        assert not notes
+        assert not read_paths
+    assert alias.is_symlink()
+
+
+@pytest.mark.parametrize(
+    "method,error_number",
+    [("is_file", errno.ELOOP), ("is_file", errno.EIO), ("iterdir", errno.EACCES)],
+)
+def test_scan_only_suppresses_loop_errors(method, error_number, daily_workspace, monkeypatch):
+    """Do not turn unrelated enumeration or stat failures into an empty history."""
+    workspace, day_dir = daily_workspace
+    note = day_dir / "regular.md"
+    error = OSError(error_number, "injected filesystem error")
+    original = getattr(Path, method)
+    target = note if method == "is_file" else day_dir
+
+    def fail_operation(path, *args, **kwargs):
+        if path == target:
+            raise error
+        return original(path, *args, **kwargs)
+
+    monkeypatch.setattr(Path, method, fail_operation)
+
+    if error_number == errno.ELOOP:
+        assert not _daily_index.scan_notes(workspace, DAY, "daily")
+    else:
+        with pytest.raises(OSError) as captured:
+            _daily_index.scan_notes(workspace, DAY, "daily")
+        assert captured.value.errno == error_number