refactor(auto_resource): streamline batch lookup refresh

reme/steps/evolve/auto_resource.py

@@ -168,16 +168,15 @@ class AutoResourceStep(BaseStep):
 
         route_counts = ", ".join(f"{spec['backend']}={len(batch)}" for spec, _, batch in routes)
         self.logger.info(f"[{self.name}] route changes={len(changes)} processors=({route_counts})")
-        # One common lookup scope covers every configured modality. Base builds
-        # history only when a loose-root resource needs it, regardless of order.
-        with _resource_lookup_scope(self.context):
-            try:
+        try:
+            # Base lazily builds history in a scope shared by every modality.
+            with _resource_lookup_scope(self.context):
                 for spec, _, indexed_changes in routes:
                     await self._dispatch_processor(spec, indexed_changes, result_slots)
-            finally:
-                # dispatch_steps merges the sub-batch into the shared context.
-                # Downstream steps and the result hook must see the original batch.
-                self.context["changes"] = changes
+        finally:
+            # dispatch_steps merges the sub-batch into the shared context.
+            # Downstream steps and the result hook must see the original batch.
+            self.context["changes"] = changes
 
         results = [item for item in result_slots if item is not None]
         success_count = sum(1 for item in results if item.get("success"))

reme/steps/evolve/base_auto_resource.py

@@ -7,7 +7,7 @@ import os
 import re
 import stat
 from abc import abstractmethod
-from collections.abc import Awaitable, Callable, Mapping
+from collections.abc import Mapping
 from contextlib import contextmanager
 from dataclasses import dataclass
 from pathlib import Path, PurePosixPath
@@ -29,26 +29,17 @@ _LOOKUP_BATCH_KEY = "_auto_resource_lookup_batch"
 _LookupKey = tuple[Path, str, int, int]
 
 
-def _stat_signature(value: os.stat_result) -> tuple[int, ...]:
-    """Detect replacement, in-place edits, and edits which restore mtime."""
-    return (value.st_dev, value.st_ino, value.st_mode, value.st_size, value.st_mtime_ns, value.st_ctime_ns)
-
-
 @dataclass(frozen=True)
 class _ResourceDaySnapshot:
     """Metadata only: all direct Markdown files, not just currently owned notes."""
 
     directory: Path
-    signature: tuple[int, ...]
-    files: dict[str, tuple]
+    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:
-        directory_stat = directory.stat()
-        if not stat.S_ISDIR(directory_stat.st_mode):
-            return None
         files = {}
         with os.scandir(directory) as entries:
             for entry in entries:
@@ -70,11 +61,8 @@ def _snapshot_note_day(workspace: Path, directory: Path) -> _ResourceDaySnapshot
                 try:
                     value = entry.stat()
                     if stat.S_ISREG(value.st_mode):
-                        files[entry.name] = (
-                            target,
-                            _stat_signature(value),
-                            _stat_signature(entry.stat(follow_symlinks=False)) if target is not None else None,
-                        )
+                        # 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
@@ -83,7 +71,7 @@ def _snapshot_note_day(workspace: Path, directory: Path) -> _ResourceDaySnapshot
                     if exc.errno == errno.ELOOP:
                         continue
                     raise
-        return _ResourceDaySnapshot(directory, _stat_signature(directory_stat), files)
+        return _ResourceDaySnapshot(directory, files)
     except (FileNotFoundError, NotADirectoryError):
         return None
     except OSError as exc:
@@ -102,7 +90,7 @@ class _ResourceDayLookup:
 
 
 class _ResourceLookupBatch:
-    """Own modality-independent index construction, refresh and invocation lifetime."""
+    """Keep lookup data and its refresh lock within one resource invocation."""
 
     def __init__(self):
         self.indexes: dict[_LookupKey, dict[str, _ResourceDayLookup]] = {}
@@ -112,33 +100,6 @@ class _ResourceLookupBatch:
         """Drop ownership data; never carry it to another batch."""
         self.indexes.clear()
 
-    async def find_matches(
-        self,
-        key: _LookupKey,
-        source: str,
-        snapshot: Callable[[], dict[str, _ResourceDaySnapshot]],
-        read_day: Callable[[str], Awaitable[list[dict]]],
-    ) -> list[tuple[str, str]]:
-        """Refresh changed days before lookup, without making each scan a transaction."""
-        async with self.lock:
-            before = snapshot()
-            days = self.indexes.get(key, {}).copy()
-            for day in sorted(before.keys() | days.keys()):
-                if day not in before:
-                    days.pop(day)
-                elif day not in days or days[day].snapshot != before[day]:
-                    owners = {}
-                    for note in await read_day(day):
-                        owner = str(note.get(_SOURCE_RESOURCE_KEY, "")).strip()
-                        if owner and owner not in owners:
-                            owners[owner] = str(note["path"])
-                    # Keep the pre-read version: a write during this read must
-                    # invalidate the cache at the next resource's lookup.
-                    days[day] = _ResourceDayLookup(before[day], owners)
-            # Failed reads never publish a partial refresh or a new version.
-            self.indexes[key] = days
-            return sorted((day, cached.owners[source]) for day, cached in days.items() if source in cached.owners)
-
 
 @contextmanager
 def _resource_lookup_scope(context: RuntimeContext, *, reuse: bool = False):
@@ -418,12 +379,27 @@ class BaseAutoResourceStep(BaseStep):
                 id(self.file_store),
                 id(self.app_context),
             )
-            matches = await batch.find_matches(
-                key,
-                self._source_resource_link(file_path),
-                self._resource_note_snapshot,
-                self._list_daily_notes,
-            )
+            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 len(matches) > 1:
             paths = ", ".join(path for _, path in matches)
             raise RuntimeError(f"Multiple daily resource notes claim {file_path}: {paths}")

tests/unit/test_auto_resource_concurrent_lookup.py

@@ -1,6 +1,7 @@
 """Concurrent resource invocations must not reuse another batch's stale ownership."""
 
 import asyncio
+from contextlib import asynccontextmanager
 
 import pytest
 
@@ -32,6 +33,41 @@ def _assert_no_active_batches(env):
     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(
@@ -49,40 +85,19 @@ async def test_another_batch_new_owner_is_seen_by_later_resource(
     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()
-    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)
-    context = RuntimeContext(
-        changes=[
-            {"change": "deleted", "path": "resource/gate.png"},
-            {"change": consumer_change, "path": "resource/photo.png"},
-        ],
-    )
-    task = asyncio.create_task(consumer(context))
-    try:
-        await asyncio.wait_for(ready.wait(), timeout=5)
+    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/")
-        resume.set()
-        response = await asyncio.wait_for(task, timeout=5)
+        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"))
-    finally:
-        resume.set()
-        await _stop_task(task)
     _assert_no_active_batches(env)
 
 
@@ -205,51 +220,30 @@ async def test_cancelled_writer_invalidates_other_batch_after_partial_write(auto
     env.write_binary("resource/photo.png", image_bytes())
     consumer = _processor(env, "2026-01-02")
     producer = _processor(env, "2026-01-01")
-    ready, resume, written = asyncio.Event(), asyncio.Event(), asyncio.Event()
-    original_delete = consumer._handle_delete  # pylint: disable=protected-access
-
-    async def pause_consumer(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)
+    written = asyncio.Event()
 
     async def pause_after_write(_day):
         written.set()
         await asyncio.Event().wait()
 
-    monkeypatch.setattr(consumer, "_handle_delete", pause_consumer)
     monkeypatch.setattr(producer, "_refresh_day_index", pause_after_write)
-    consumer_task = asyncio.create_task(
-        env.run(
-            consumer,
-            [
-                {"change": "deleted", "path": "resource/gate.png"},
-                {"change": "deleted", "path": "resource/photo.png"},
-            ],
-        ),
-    )
-    producer_task = None
-    try:
-        await asyncio.wait_for(ready.wait(), timeout=5)
+
+    async with _paused_between_resources(env, monkeypatch, consumer) as resume:
         producer_task = asyncio.create_task(env.run(producer, [{"change": "added", "path": "resource/photo.png"}]))
-        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
-        resume.set()
-        response = await asyncio.wait_for(consumer_task, timeout=5)
-        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:
-        resume.set()
-        await _stop_task(consumer_task)
-        if producer_task is not None:
+        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)
 
@@ -262,40 +256,17 @@ async def test_another_batch_deletion_removes_cached_day(auto_resource_env, monk
     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()
-    original_delete = consumer._handle_delete  # pylint: disable=protected-access
 
-    async def pause_consumer(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_consumer)
-    task = asyncio.create_task(
-        env.run(
-            consumer,
-            [
-                {"change": "deleted", "path": "resource/gate.png"},
-                {"change": "added", "path": "resource/photo.png"},
-            ],
-        ),
-    )
-    try:
-        await asyncio.wait_for(ready.wait(), timeout=5)
+    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()
-        resume.set()
-        response = await asyncio.wait_for(task, timeout=5)
+        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"))
-    finally:
-        resume.set()
-        await _stop_task(task)
     _assert_no_active_batches(env)
 
 

tests/unit/test_auto_resource_external_lookup.py

@@ -158,6 +158,37 @@ async def test_external_owner_removal_stops_reusing_previous_date(new_source, au
     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."""

tests/unit/test_auto_resource_snapshot_lookup.py

@@ -4,7 +4,6 @@ import asyncio
 import errno
 from collections import Counter
 from contextlib import nullcontext
-from pathlib import Path
 from types import SimpleNamespace
 from unittest.mock import Mock
 
@@ -251,14 +250,32 @@ async def test_failed_changed_day_read_does_not_accept_new_fingerprint(auto_reso
     assert not card.exists()
 
 
-async def test_unchanged_snapshots_do_not_reparse_historical_notes(auto_resource_env, monkeypatch):
-    """Every resource checks metadata, but unchanged historical days are parsed once."""
+@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()
 
@@ -268,7 +285,7 @@ async def test_unchanged_snapshots_do_not_reparse_historical_notes(auto_resource
 
     monkeypatch.setitem(env.app_context.jobs, "daily_list", counted_daily_list)
     response = await env.run(
-        _processor(env),
+        step,
         [{"change": "deleted", "path": f"resource/missing-{index}.png"} for index in range(8)],
     )
     assert response.success is True
@@ -277,6 +294,42 @@ async def test_unchanged_snapshots_do_not_reparse_historical_notes(auto_resource
     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."""
@@ -309,7 +362,7 @@ async def test_cyclic_daily_links_do_not_block_unrelated_resource(cycle, auto_re
     assert link.is_symlink()
 
 
-@pytest.mark.parametrize("failure_site", ["directory-stat", "scandir", "file-stat"])
+@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."""
@@ -317,19 +370,11 @@ async def test_snapshot_handles_only_loop_errors(failure_site, error_number, tmp
     day.mkdir(parents=True)
     (day / "valid.md").write_text("A normal note.", encoding="utf-8")
     error = OSError(error_number, "snapshot probe failed")
-    original_stat = Path.stat
     with base_auto_resource.os.scandir(day) as entries:
         valid_entries = list(entries)
 
-    def stat_with_failure(path, *args, **kwargs):
-        if path == day:
-            raise error
-        return original_stat(path, *args, **kwargs)
-
     with monkeypatch.context() as patch:
-        if failure_site == "directory-stat":
-            patch.setattr(Path, "stat", stat_with_failure)
-        elif failure_site == "scandir":
+        if failure_site == "scandir":
             patch.setattr(base_auto_resource.os, "scandir", Mock(side_effect=error))
         else:
             unreadable = SimpleNamespace(