fix(cascade): reconcile vanished source paths as deletes

src/everos/memory/cascade/worker.py

@@ -330,7 +330,26 @@ async def _process_one(self, row: MdChangeState) -> str | None:
                 if row.change_type == "deleted":
                     outcome = await handler.handle_deleted(row.md_path)
                 else:
-                    outcome = await handler.handle_added_or_modified(row.md_path)
+                    try:
+                        outcome = await handler.handle_added_or_modified(row.md_path)
+                    except FileNotFoundError as exc:
+                        absolute = handler._deps.memory_root.root / row.md_path
+                        if absolute.exists():
+                            raise RecoverableError(
+                                "handler dependency disappeared while source md "
+                                "still exists"
+                            ) from exc
+                        # Watcher events can be stale by the time the worker
+                        # reads the path (for example, delete followed by a
+                        # fast replace). Reconcile the durable stores to the
+                        # filesystem state instead of leaving the old row
+                        # indexed as a permanent failure.
+                        logger.info(
+                            "cascade_worker_missing_path_reconciled_as_delete",
+                            md_path=row.md_path,
+                            kind=row.kind,
+                        )
+                        outcome = await handler.handle_deleted(row.md_path)
             except RecoverableError as exc:
                 last_error = f"{type(exc).__name__}: {exc}"
                 logger.warning(

tests/unit/test_memory/test_cascade/test_worker.py

@@ -26,6 +26,7 @@
 import time
 import unittest.mock as mock
 from dataclasses import dataclass
+from pathlib import Path
 
 import pytest
 
@@ -103,6 +104,19 @@ async def handle_added_or_modified(self, md_path: str) -> HandlerOutcome:
         raise RuntimeError("unexpected boom")
 
 
+class _MissingFileHandler(_OkHandler):
+    def __init__(self, root: Path) -> None:
+        self._deps = mock.Mock(memory_root=mock.Mock(root=root))
+        self.deleted_paths: list[str] = []
+
+    async def handle_added_or_modified(self, md_path: str) -> HandlerOutcome:
+        raise FileNotFoundError(md_path)
+
+    async def handle_deleted(self, md_path: str) -> HandlerOutcome:
+        self.deleted_paths.append(md_path)
+        return await super().handle_deleted(md_path)
+
+
 @pytest.fixture
 def patched_repo(monkeypatch: pytest.MonkeyPatch) -> _FakeRepo:
     """Drop a fake repo onto the module the worker imports."""
@@ -157,6 +171,40 @@ async def test_bare_exception_marked_permanent(patched_repo: _FakeRepo) -> None:
     assert retryable is False
 
 
+async def test_missing_file_during_upsert_is_reconciled_as_delete(
+    patched_repo: _FakeRepo,
+    tmp_path: Path,
+) -> None:
+    """A stale add/modify event must not leave the prior row indexed."""
+    patched_repo.batch = [_Row(md_path="vanished.md", change_type="modified")]
+    handler = _MissingFileHandler(tmp_path)
+    w = CascadeWorker({"episode": handler}, retry_backoff_seconds=0)
+
+    await w.drain_once()
+
+    assert handler.deleted_paths == ["vanished.md"]
+    assert patched_repo.done == ["vanished.md"]
+    assert patched_repo.failed == []
+
+
+async def test_missing_dependency_does_not_delete_an_existing_source(
+    patched_repo: _FakeRepo,
+    tmp_path: Path,
+) -> None:
+    """Only disappearance of the source md is treated as a delete."""
+    (tmp_path / "present.md").write_text("still here", encoding="utf-8")
+    patched_repo.batch = [_Row(md_path="present.md", change_type="modified")]
+    handler = _MissingFileHandler(tmp_path)
+    w = CascadeWorker({"episode": handler}, retry_backoff_seconds=0)
+
+    await w.drain_once()
+
+    assert handler.deleted_paths == []
+    assert patched_repo.done == []
+    assert patched_repo.failed[0][0] == "present.md"
+    assert patched_repo.failed[0][1] is True
+
+
 async def test_unknown_kind_marks_permanent_without_handler(
     patched_repo: _FakeRepo,
 ) -> None: