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): reuse initial ownership without refreshes

Browse source
This commit is contained in:
wang-qisen 2026-09-09 21:41:13 +08:00
parent 1848428828
commit 8601d2d322
2 changed files with 123 additions and 135 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -21,56 +21,40 @@ from ._evolve import now
_SOURCE_RESOURCE_KEY = "source_resource"
_DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$")
_UNSAFE_FILENAME_CHARS = re.compile(r'[<>:"/\\|?*\x00-\x1f]+')
_LOOKUP_BATCH_KEY = "_auto_resource_lookup_batch"
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
_LOOKUP_TABLE_KEY = "_auto_resource_lookup_table"
@dataclass
class _ResourceLookupBatch:
"""Cache historical ownership only for this sequential resource invocation.
class _ResourceLookupTable:
"""Initial source → day/path ownership, shared by this invocation's processors.
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.
Build once without refreshes. Repeated sources use the original live lookup.
None means history has not been read; an empty dict is a valid result.
"""
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)
owners: dict[str, dict[str, str]] | None = None
seen_sources: set[str] = field(default_factory=set)
@contextmanager
def _resource_lookup_scope(context: RuntimeContext, *, reuse: bool = False):
"""Share a router's index with processors, never with the next watch batch."""
previous = context.get(_LOOKUP_BATCH_KEY)
if reuse and isinstance(previous, _ResourceLookupBatch):
previous = context.get(_LOOKUP_TABLE_KEY)
if reuse and isinstance(previous, _ResourceLookupTable):
yield previous
return
existed = _LOOKUP_BATCH_KEY in context
batch = _ResourceLookupBatch()
context[_LOOKUP_BATCH_KEY] = batch
existed = _LOOKUP_TABLE_KEY in context
table = _ResourceLookupTable()
context[_LOOKUP_TABLE_KEY] = table
try:
yield batch
yield table
finally:
batch.days = None
batch.dirty_days.clear()
table.owners = None
table.seen_sources.clear()
if existed:
context[_LOOKUP_BATCH_KEY] = previous
context[_LOOKUP_TABLE_KEY] = previous
else:
del context[_LOOKUP_BATCH_KEY]
del context[_LOOKUP_TABLE_KEY]
@dataclass(frozen=True)
@ -153,6 +137,7 @@ def _sanitize_note_name(raw: str, fallback: str) -> str:
class BaseAutoResourceStep(BaseStep):
"""Shared source-linked daily-note lifecycle for resource processors."""
_resource_lookup: _ResourceLookupTable | None = None
resource_fallback = False
resource_suffixes: frozenset[str] = frozenset()
router_inherit_keys = frozenset({"file_store", "language"})
@ -285,16 +270,21 @@ class BaseAutoResourceStep(BaseStep):
notes = await self._list_daily_notes(day)
return self._find_resource_note(notes, file_path)
async def _lookup_resource_note_path(self, day: str, file_path: str) -> str | None:
"""Reuse the initial path before processing a source, not after writing it."""
if self._resource_lookup is not None:
assert self._resource_lookup.owners is not None
source = self._source_resource_link(file_path)
return self._resource_lookup.owners.get(source, {}).get(day)
note = await self._list_resource_note(day, file_path)
return str(note["path"]) if note is not None else None
async def _list_daily_notes(self, day: str) -> list[dict]:
"""Use the configured daily_list job for both history and targeted checks."""
list_response = await self.run_job("daily_list", date=day)
if not list_response.success:
raise RuntimeError(f"daily_list failed: {list_response.answer}")
notes = list_response.metadata.get("notes") or []
batch = self.context.get(_LOOKUP_BATCH_KEY)
if isinstance(batch, _ResourceLookupBatch):
batch.replace_day(day, notes)
return notes
return list_response.metadata.get("notes") or []
def _daily_note_days(self) -> list[str]:
"""Return safe, deterministic daily subdirectories available for lookup."""
@ -318,26 +308,32 @@ class BaseAutoResourceStep(BaseStep):
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."""
"""Use initial ownership once per source; repeat events retain the original lookup."""
assert self.context is not None
with _resource_lookup_scope(self.context, reuse=True) as batch:
if batch.days is None:
days = {}
self._resource_lookup = None
with _resource_lookup_scope(self.context, reuse=True) as table:
if file_path in table.seen_sources:
matches = []
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)
note = await self._list_resource_note(day, file_path)
if note is not None:
matches.append((day, str(note["path"])))
else:
table.seen_sources.add(file_path)
if table.owners is None:
owners = {}
for day in self._daily_note_days():
for note in await self._list_daily_notes(day):
source = str(note.get(_SOURCE_RESOURCE_KEY, "")).strip()
if source:
# Preserve daily_list's first match within each day.
owners.setdefault(source, {}).setdefault(day, str(note["path"]))
# Publish only after the entire initial scan succeeds.
table.owners = owners
self._resource_lookup = table
source = self._source_resource_link(file_path)
matches = sorted(table.owners.get(source, {}).items())
if len(matches) > 1:
paths = ", ".join(path for _, path in matches)
raise RuntimeError(f"Multiple daily resource notes claim {file_path}: {paths}")
@ -345,9 +341,8 @@ class BaseAutoResourceStep(BaseStep):
async def _prepare_resource_note(self, day: str, file_path: str, note_stem: str) -> _ResourceNoteState:
"""Find the owned note or allocate a safe path before the first write."""
note = await self._list_resource_note(day, file_path)
if note is not None:
note_path = str(note["path"])
note_path = await self._lookup_resource_note_path(day, file_path)
if note_path is not None:
return _ResourceNoteState(path=note_path, created=False, before_bytes=self._note_bytes(note_path))
_, note_path = self._unique_daily_note_path(day, note_stem, file_path, current_path="")
@ -508,8 +503,8 @@ class BaseAutoResourceStep(BaseStep):
return note_path
async def _handle_delete(self, file_path: str, date_str: str, note_stem: str) -> None:
note = await self._list_resource_note(date_str, file_path)
if note is None:
note_rel = await self._lookup_resource_note_path(date_str, file_path)
if note_rel is None:
self.context.response.success = True
self.context.response.answer = f"No linked resource note to delete: {file_path}"
self.context.response.metadata.update(
@ -525,7 +520,6 @@ class BaseAutoResourceStep(BaseStep):
self.logger.info(f"[{self.name}] delete skipped; no owned note file_path={file_path}")
return
note_rel = str(note["path"])
note_abs = self.workspace_path / note_rel
note_existed = note_abs.is_file()
self.logger.info(f"[{self.name}] delete start note={note_rel}")
@ -564,6 +558,7 @@ class BaseAutoResourceStep(BaseStep):
async def _handle_change(self, file_path: str, raw_change) -> dict:
assert self.context is not None
self._resource_lookup = None
# Handlers write item-scoped fields into the shared response. Start each
# change with a fresh mapping so one result cannot inherit another's metadata.
self.context.response.metadata = {}
@ -616,23 +611,16 @@ 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}")
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)
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,
)
return {
"success": self.context.response.success,
"path": file_path,

114
tests/unit/test_auto_resource_batch_lookup.py
View file

@ -1,4 +1,4 @@
"""Reuse history within a resource batch and maintain its own note mutations."""
"""Reuse initial ownership for distinct resources and live lookup for repeated sources."""
import asyncio
import os
@ -128,6 +128,9 @@ async def test_loose_batch_reads_history_once(
if write_notes:
assert all(note.read_bytes() != previous for note, previous in before.items())
assert all(result["metadata"]["modified"] for result in results)
# All existing notes share one already-read day. Pre-write lookup must
# reuse it; normal day-index rebuilds are not daily_list calls.
assert calls[("daily_list", "2026-01-01")] == 1
async def test_mixed_processors_share_history_and_keep_result_order(auto_resource_env, monkeypatch):
@ -151,6 +154,7 @@ async def test_mixed_processors_share_history_and_keep_result_order(auto_resourc
assert not text_note.exists()
assert not image_note.exists()
_assert_history_reads(calls)
assert calls[("daily_list", "2026-01-01")] == 1
@pytest.mark.parametrize("routed", [False, True], ids=["text-step", "text-only-router"])
@ -175,26 +179,37 @@ async def test_text_only_batch_uses_history_without_image_processor(routed, date
assert response.success is True
assert all(not note.exists() for note in notes)
_assert_history_reads(calls, 0 if dated else 1)
assert calls[("daily_list", "2026-01-01")] == (3 if dated else 1)
@pytest.mark.parametrize("routed", [False, True], ids=["image", "unified-router"])
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."""
@pytest.mark.parametrize("dated", [False, True], ids=["loose-root", "dated-resource"])
async def test_new_image_uses_live_post_write_resolution(routed, dated, auto_resource_env, monkeypatch):
"""Only dated input needs a pre-write daily_list; both paths read the newly written note."""
env = auto_resource_env
_seed_history(env)
source = env.write_binary("resource/2026-01-01/caption.png", image_bytes())
prefix = "resource/2026-01-01" if dated else "resource"
source = env.write_binary(f"{prefix}/caption.png", image_bytes())
calls = _observe_history(env, monkeypatch)
plain_call = FakeVisionModel.__call__
response = await env.run(
env.processor(_model(), routed=routed),
[{"change": "added", "path": str(source)}],
)
async def observe_pre_write(model, messages, **kwargs):
assert calls[("daily_list", "2026-01-01")] == (1 if dated else 0)
return await plain_call(model, messages, **kwargs)
monkeypatch.setattr(FakeVisionModel, "__call__", observe_pre_write)
with patch.object(BaseAutoResourceStep, "_today", return_value="2026-01-01"):
response = await env.run(
env.processor(_model(), routed=routed),
[{"change": "added", "path": str(source)}],
)
assert response.success is True
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)
_assert_history_reads(calls, 0 if dated else 1)
assert calls[("daily_list", "2026-01-01")] == (2 if dated else 1)
@pytest.mark.parametrize("routed", [False, True], ids=["image", "unified-router"])
@ -267,18 +282,19 @@ async def test_create_update_delete_recreate_maintains_ownership_across_midnight
]
assert not (env.workspace / "daily/2026-01-01/caption.md").exists()
assert (env.workspace / "daily/2026-01-02/caption.md").is_file()
_assert_history_reads(calls)
# The three repeated events deliberately use the original uncached lookup.
_assert_history_reads(calls, 4)
@pytest.mark.parametrize("routed", [False, True], ids=["image", "unified-router"])
@pytest.mark.parametrize("first_change", ["added", "deleted"])
async def test_partial_index_failure_refreshes_only_affected_day(
async def test_repeated_source_retries_after_partial_write_or_delete(
routed,
first_change,
auto_resource_env,
monkeypatch,
):
"""A failure after writing or deleting must not hide that mutation from the next item."""
"""A repeated source uses live lookup even when its previous event failed after mutation."""
env = auto_resource_env
_seed_history(env)
env.write_binary("resource/caption.png", image_bytes())
@ -316,58 +332,42 @@ async def test_partial_index_failure_refreshes_only_affected_day(
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_history_reads(calls, 2)
assert len(list((env.workspace / "daily").glob("*/caption.md"))) == 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."""
@pytest.mark.parametrize("routed", [False, True], ids=["image", "unified-router"])
@pytest.mark.parametrize("mixed_paths", [False, True], ids=["relative-paths", "absolute-then-relative"])
async def test_repeated_existing_updates_keep_original_day_and_path(
routed,
mixed_paths,
auto_resource_env,
monkeypatch,
):
"""Repeated events use the same logical source even when absolute and relative inputs differ."""
env = auto_resource_env
_seed_history(env)
env.write_binary("resource/caption.png", image_bytes())
source = env.write_binary("resource/photo.png", image_bytes())
note = env.write_note("daily/2026-01-01/original.md", "[[resource/photo.png]]")
before = note.read_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
model = _model()
changes = [
{"change": "modified", "path": str(source) if mixed_paths else "resource/photo.png"},
{"change": "modified", "path": "resource/photo.png"},
]
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)
with patch.object(BaseAutoResourceStep, "_today", return_value="2026-01-02"):
response = await env.run(env.processor(model, routed=routed), changes)
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)
assert response.success is True
metadata = [result["metadata"] for result in response.metadata["results"]]
assert [item["path"] for item in metadata] == ["daily/2026-01-01/original.md"] * 2
assert all(item["action"] == "modified" and item["created"] is False for item in metadata)
assert len(model.calls) == 2
assert note.read_bytes() != before
assert not (env.workspace / "daily/2026-01-02").exists()
_assert_history_reads(calls, 2)
@pytest.mark.parametrize("routed", [False, True], ids=["image", "unified-router"])
@ -420,7 +420,7 @@ async def test_same_day_duplicates_keep_first_match_after_deletion(routed, auto_
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_history_reads(calls)
_assert_history_reads(calls, 3)
@pytest.mark.parametrize("routed", [False, True], ids=["image", "unified-router"])