From 8601d2d3221219617cbf0a50f150131050588abe Mon Sep 17 00:00:00 2001 From: wang-qisen Date: Wed, 9 Sep 2026 21:41:13 +0800 Subject: [PATCH] refactor(auto_resource): reuse initial ownership without refreshes --- reme/steps/evolve/base_auto_resource.py | 144 ++++++++---------- tests/unit/test_auto_resource_batch_lookup.py | 114 +++++++------- 2 files changed, 123 insertions(+), 135 deletions(-) diff --git a/reme/steps/evolve/base_auto_resource.py b/reme/steps/evolve/base_auto_resource.py index 04b0ae4c..0ec7f4dc 100644 --- a/reme/steps/evolve/base_auto_resource.py +++ b/reme/steps/evolve/base_auto_resource.py @@ -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, diff --git a/tests/unit/test_auto_resource_batch_lookup.py b/tests/unit/test_auto_resource_batch_lookup.py index 7884783a..58fa780d 100644 --- a/tests/unit/test_auto_resource_batch_lookup.py +++ b/tests/unit/test_auto_resource_batch_lookup.py @@ -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"])