scriptoriumadmin/ReMe
1
0
Fork 0
mirror of https://github.com/agentscope-ai/ReMe.git synced 2026-11-15 09:11:24 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

refactor(auto_resource): scope ownership cache to batch mutations

Browse source 
Replace filesystem snapshots with one lazy historical lookup shared by resource processors. Reload only days changed by this invocation and discard the cache when it ends.

Reuse canonical daily-note scans for symlink safety and Python 3.13 compatibility. Consolidate regression tests around the narrowed batch contract.

Validation: Python 3.11, 3.12 and 3.13 core/plugin suites each passed 1407 tests; pre-commit --all-files and offline lookup/deletion comparison passed.
This commit is contained in:
wang-qisen 2026-09-09 18:14:58 +08:00
parent 375d8c2d2a
commit 1848428828
8 changed files with 546 additions and 1546 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

211
reme/steps/evolve/base_auto_resource.py
View file

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

40
reme/steps/file_io/_daily_index.py
View file

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

571
tests/unit/test_auto_resource_batch_lookup.py
View file

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

324
tests/unit/test_auto_resource_concurrent_lookup.py
View file

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

280
tests/unit/test_auto_resource_external_lookup.py
View file

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

104
tests/unit/test_auto_resource_shared_lookup.py
View file

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

427
tests/unit/test_auto_resource_snapshot_lookup.py
View file

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

135
tests/unit/test_daily_note_links.py Normal file
View file

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