diff --git a/changes/462.fixed.md b/changes/462.fixed.md
new file mode 100644
index 000000000..909acaab2
--- /dev/null
+++ b/changes/462.fixed.md
@@ -0,0 +1,3 @@
+**A memory's derived data no longer outlives it when a cleanup step fails once.** Before, if deleting a memory could not finish removing what it fed (a locked file, a transient database error), the leftovers stayed forever and nothing retried; now the unfinished cleanup is queued and replayed by memory upkeep — hourly as well as nightly — until it succeeds, or dropped with a warning after seven days so deleted text never lingers. A retry starts only once the memory row itself is confirmed gone, and a summary's vector is recorded before its row is deleted, so a failed step can never orphan it.
+
+Two sharp edges found in review are closed as well: a retry records each summary's id next to its rowid, so a new summary that SQLite hands the freed rowid while the retry is waiting is left alone instead of being deleted with its vector; and a delete that could not queue its cleanup for a retry answers "try again" (503) instead of "not found" (404), because the memory is still there. A crash between the delete and its queue mark no longer strands the cleanup forever — upkeep replays old unfinished queue rows once their memory is gone — and a failed retry no longer skips review expiry.
diff --git a/sentient/app.py b/sentient/app.py
index 0bc84a21e..08d4fd2de 100644
--- a/sentient/app.py
+++ b/sentient/app.py
@@ -324,10 +324,7 @@ async def delete_chat(self, session_id: str, *, reason: str | None = None) -> No
     def _after_memory_erased(self, erased: Any) -> None:
         """A memory was deleted (#152): its words leave USER.md and MEMORY.md (not a remaining memory's similar lines,
         #456), and the user model reads its insights again (some may be gone with it)."""
-        try:
-            self.workspace.scrub(erased.wordings, self.config.assistant.user_name, repeats=erased)
-        except OSError:
-            log.exception("removing a deleted memory from the workspace files failed")
+        self.workspace.scrub(erased.wordings, self.config.assistant.user_name, repeats=erased)
         model = getattr(self, "user_model", None)
         if model is not None:
             model._vectors = None
diff --git a/sentient/evolution/service.py b/sentient/evolution/service.py
index 9443f1f2d..fe1188571 100644
--- a/sentient/evolution/service.py
+++ b/sentient/evolution/service.py
@@ -209,8 +209,18 @@ async def _tick(self) -> None:
             await self.review_due(now)
         if self.app.memory is not None:
             if await self._due("memory.purge_last_run", timedelta(minutes=cfg.memory.purge_interval_minutes), now):
-                await self.app.memory.purge_expired()
-                await memory_review.expire(self.app, now)
+                try:
+                    await self.app.memory.purge_expired()
+                finally:
+                    # A purge failure must not block the pending-erase retries (#462):
+                    # the drain still runs — its own failure is logged rather than
+                    # raised, so review expiry still happens — then the purge error
+                    # keeps propagating.
+                    try:
+                        await self.app.memory.drain_pending_erases()
+                    except Exception:
+                        log.exception("pending-erase retry failed during upkeep")
+                    await memory_review.expire(self.app, now)
             if cfg.memory.summaries_enabled and await self._due(
                 "memory.summaries_last_run", timedelta(minutes=cfg.memory.summarize_interval_minutes), now
             ):
diff --git a/sentient/gateway/routes/memory.py b/sentient/gateway/routes/memory.py
index aae641939..3271e3a75 100644
--- a/sentient/gateway/routes/memory.py
+++ b/sentient/gateway/routes/memory.py
@@ -303,5 +303,10 @@ async def delete_memory(request: Request, memory_id: int):
     mem = _memory(request)
     deleted = await mem.erase(memory_id, notify=True)  # with everything derived from it (#152)
     if not deleted:
+        # A False that left the memory in place means the erase aborted before
+        # touching anything (its pending-cleanup row could not be recorded),
+        # which is a temporary failure, not a missing id (#462).
+        if await mem.get_fact(memory_id) is not None:
+            raise HTTPException(503, "couldn't delete this memory right now, try again")
         raise HTTPException(404, f"memory {memory_id} not found")
     return {"deleted": True}
diff --git a/sentient/memory/dreaming.py b/sentient/memory/dreaming.py
index 4acbc3813..9ef242bf7 100644
--- a/sentient/memory/dreaming.py
+++ b/sentient/memory/dreaming.py
@@ -304,7 +304,7 @@ async def _execute(self, dream_id: str) -> dict:
         cfg = self.cfg
         stats = {
             "facts_reviewed": 0, "merged": 0, "contradictions_resolved": 0, "promoted": 0, "expired": 0,
-            "insights_updated": 0, "languages_merged": 0,
+            "insights_updated": 0, "languages_merged": 0, "erase_retries": 0,
         }
         notes: dict[str, Any] = {"merged": [], "languages": [], "contradictions": [], "promoted": [], "insights": None}
         errors: list[str] = []
@@ -334,6 +334,10 @@ async def _execute(self, dream_id: str) -> dict:
                     stats["expired"] = await mem.purge_expired()
                 except Exception as exc:
                     errors.append(f"purge: {exc}")
+                try:
+                    stats["erase_retries"] = await mem.drain_pending_erases()
+                except Exception as exc:
+                    errors.append(f"erase retries: {exc}")
             if cfg.refresh_user_model and self.app.config.user_model.enabled:
                 try:
                     res = await self.app.user_model.refresh(trigger="dream")
@@ -659,6 +663,8 @@ def journal(self, stats: dict, notes: dict, errors: list[str]) -> str:
             did.append(f"kept {stats['promoted']} short-term memor{'ies' if stats['promoted'] != 1 else 'y'} for the long run")
         if stats["expired"]:
             did.append(f"let go of {stats['expired']} expired memor{'ies' if stats['expired'] != 1 else 'y'}")
+        if stats.get("erase_retries"):
+            did.append(f"retried cleanup for {stats['erase_retries']} erased memor{'ies' if stats['erase_retries'] != 1 else 'y'}")
         if stats["insights_updated"]:
             did.append(f"updated {stats['insights_updated']} things I understand about {name}")
         if did:
diff --git a/sentient/memory/facts.py b/sentient/memory/facts.py
index 1a0baef43..9c9d6f793 100644
--- a/sentient/memory/facts.py
+++ b/sentient/memory/facts.py
@@ -1040,22 +1040,90 @@ async def erase(self, fid: int, *, notify: bool = False) -> bool:
             " OR fact_id IN (SELECT id FROM facts WHERE merged_into = ? AND status = 'merged')",
             (fid, fid),
         )
+        rows = [row, *copies]
+        texts = list(dict.fromkeys(t for r in rows for t in (r["content"], r["previous_content"]) if t))
+        refs = [_load_ref(r["source_ref"]) for r in rows] + [_load_ref(r["ref"]) for r in links]
+        ids = [int(r["id"]) for r in rows]
+        live_refs = [r for r in refs if r]
+        # Record the intended cleanup before the irreversible deletion (#462).
+        pending = None
+        try:
+            cur = await self.store.execute(
+                "INSERT INTO pending_erase(fact_id, payload, at) VALUES(?,?,?)",
+                (fid, json.dumps({"ids": ids, "texts": texts, "refs": live_refs}), now_iso()),
+            )
+            pending = int(cur.lastrowid)
+        except Exception:
+            log.exception("could not record pending cleanup for memory %s; aborting erase", fid)
+            return False
         if not await self.forget(fid, notify=notify):
+            with contextlib.suppress(Exception):
+                await self.store.execute("DELETE FROM pending_erase WHERE id = ?", (pending,))
             return False
+        try:
+            # Only now may upkeep replay this cleanup: while the delete was still
+            # running, a concurrent drain could strip a live memory's derived data
+            # before forget() had really removed the memory (#462).
+            await self.store.execute("UPDATE pending_erase SET complete = 1 WHERE id = ?", (pending,))
+        except Exception:
+            log.warning("could not mark memory %s's cleanup as ready to retry", fid)
         if row["state"] == "superseded" and row["superseded_by"] is not None:
             await self.store.execute(
                 "UPDATE facts SET previous_content = NULL WHERE id = ? AND previous_content = ?",
                 (row["superseded_by"], row["content"]),
             )
-        rows = [row, *copies]
-        texts = list(dict.fromkeys(t for r in rows for t in (r["content"], r["previous_content"]) if t))
-        refs = [_load_ref(r["source_ref"]) for r in rows] + [_load_ref(r["ref"]) for r in links]
         try:
-            await self._cascade([int(r["id"]) for r in rows], texts, [r for r in refs if r])
-        except Exception:  # the memory itself is gone; a leftover is logged, never a failed delete
-            log.exception("cleaning up after memory %s failed", fid)
+            await self._cascade(ids, texts, live_refs, pending=pending)
+        except Exception:
+            log.exception("cleaning up after memory %s failed; will retry later", fid)
+            return True
+        with contextlib.suppress(Exception):
+            await self.store.execute("DELETE FROM pending_erase WHERE id = ?", (pending,))
         return True
 
+    async def drain_pending_erases(self) -> int:
+        """Cleanup steps an erase could not finish (#462), replayed by memory upkeep until they succeed, so
+        derived data never outlives its memory because one step failed once (a locked file, a transient
+        database error). Returns how many waits were cleared."""
+        await self.ensure_schema()
+        # Drop stale retries (>7 days) so erased text does not linger indefinitely.
+        cutoff = (datetime.now(UTC) - timedelta(days=7)).isoformat()
+        stale = await self.store.fetchall("SELECT id, fact_id FROM pending_erase WHERE at < ?", (cutoff,))
+        for row in stale:
+            log.warning(
+                "dropping stale pending erase for memory %s after 7 days; some derived data may remain",
+                row["fact_id"],
+            )
+        if stale:
+            await self.store.execute("DELETE FROM pending_erase WHERE at < ?", (cutoff,))
+        # A crash between the delete and its "complete" mark (a forget() that raised
+        # after the fact row went) strands the wait at complete = 0; once the memory
+        # no longer exists the delete really happened, so the wait is safe to replay.
+        # The one-hour age keeps a drain from racing an erase that is still running.
+        settled = (datetime.now(UTC) - timedelta(hours=1)).isoformat()
+        await self.store.execute(
+            "UPDATE pending_erase SET complete = 1 WHERE complete = 0 AND at < ?"
+            " AND fact_id NOT IN (SELECT id FROM facts)",
+            (settled,),
+        )
+        drained = 0
+        for row in await self.store.fetchall(
+            "SELECT id, fact_id, payload FROM pending_erase WHERE complete = 1 ORDER BY id"
+        ):
+            try:
+                data = json.loads(row["payload"])
+                await self._cascade(
+                    data["ids"], data["texts"], data["refs"],
+                    summary_vec_ids=data.get("summary_vec_ids") or (), pending=int(row["id"]),
+                )
+            except Exception:
+                log.exception("cleaning up after memory %s failed again", row["fact_id"])
+                continue
+            await self.store.execute("DELETE FROM pending_erase WHERE id = ?", (row["id"],))
+            drained += 1
+            log.info("finished cleaning up after memory %s on retry", row["fact_id"])
+        return drained
+
     async def _erased(self, texts: list[str], ids: list[int]) -> origins.Erased:
         """What ``texts`` said, against every memory still in use (#456): what derived data repeats them goes,
         unless a remaining memory says it as well or better (:class:`origins.Erased`)."""
@@ -1064,12 +1132,50 @@ async def _erased(self, texts: list[str], ids: list[int]) -> origins.Erased:
         remaining = [(int(r["id"]), str(r["content"])) for r in rows if int(r["id"]) not in gone]
         return origins.Erased(texts, remaining, self.config.assistant.user_name)
 
-    async def _cascade(self, ids: list[int], texts: list[str], refs: list[dict]) -> None:
+    async def _note_summary_vecs(self, pending: int, vec_ids: list) -> None:
+        """Merge summary vector row IDs into a pending erase's payload (#462), so a retry can still reach the
+        vectors once their summary rows are gone. Each rowid is stored with its summary's TEXT id, so a retry
+        can tell an orphaned vector apart from a new summary that reused the freed rowid."""
+        row = await self.store.fetchone("SELECT payload FROM pending_erase WHERE id = ?", (pending,))
+        if row is None:
+            return
+        payload = json.loads(row["payload"])
+        merged: dict[int, str | None] = {}
+        for v in payload.get("summary_vec_ids") or []:
+            if isinstance(v, (list, tuple)) and len(v) >= 2:
+                merged[int(v[0])] = str(v[1])
+            else:
+                merged[int(v)] = None
+        for vid, sid in vec_ids:
+            if sid is not None:
+                merged[int(vid)] = str(sid)
+            else:
+                merged.setdefault(int(vid), None)
+        payload["summary_vec_ids"] = [
+            [vid, sid] if sid is not None else vid for vid, sid in sorted(merged.items())
+        ]
+        await self.store.execute(
+            "UPDATE pending_erase SET payload = ? WHERE id = ?", (json.dumps(payload), pending)
+        )
+
+    async def _cascade(
+        self, ids: list[int], texts: list[str], refs: list[dict],
+        summary_vec_ids: tuple[int, ...] | list[int] = (), pending: int | None = None,
+    ) -> None:
         store = self.store
         repeats = await self._erased(texts, ids)
 
         # chat summaries made of the messages it came from, that repeat it, and their vectors
         said_in = {str(r["message_id"]) for r in refs if r.get("message_id")}
+        # rowid -> the summary's TEXT id, so a retry can tell the original apart from a
+        # new summary SQLite handed the freed rowid (#462); a bare int (a payload from
+        # before ids were recorded next to rowids) keeps the old delete-both semantics.
+        vec_ids: dict[int, str | None] = {}
+        for v in summary_vec_ids:
+            if isinstance(v, (list, tuple)) and len(v) >= 2:
+                vec_ids[int(v[0])] = str(v[1])
+            else:
+                vec_ids[int(v)] = None
         if said_in:
             for r in await store.fetchall("SELECT rowid, id, content, message_ids FROM summaries"):
                 try:
@@ -1077,8 +1183,24 @@ async def _cascade(self, ids: list[int], texts: list[str], refs: list[dict]) ->
                 except ValueError:
                     continue
                 if made_of & said_in and repeats(r["content"]):
-                    await store.execute("DELETE FROM summaries WHERE id = ?", (r["id"],))
-                    await self.episodic.vec.delete(int(r["rowid"]))
+                    vec_ids[int(r["rowid"])] = str(r["id"])
+        if vec_ids and pending is not None:
+            # Note the vector row IDs (with their summary ids) before any summary
+            # row goes, so a retry can still delete the vectors once their rows are
+            # gone (#462) — and can leave a reused rowid alone.
+            with contextlib.suppress(Exception):
+                await self._note_summary_vecs(pending, sorted(vec_ids.items()))
+        for vid, sid in sorted(vec_ids.items()):
+            current = await store.fetchone("SELECT id FROM summaries WHERE rowid = ?", (vid,))
+            if current is not None and sid is not None and current["id"] != sid:
+                # SQLite handed the freed rowid to a new summary: the row and the
+                # vector at it are the newcomer's, not part of this cleanup (#462).
+                continue
+            # The vector goes first: a failure here leaves the summary row in place,
+            # so the next attempt finds both again.
+            await self.episodic.vec.delete(vid)
+            if current is not None:
+                await store.execute("DELETE FROM summaries WHERE rowid = ?", (vid,))
         # the nightly journal's lines about it
         for r in await store.fetchall("SELECT id, journal_md FROM dreams WHERE journal_md != ''"):
             lines = str(r["journal_md"]).splitlines()
@@ -2033,7 +2155,7 @@ async def purge_expired(self) -> int:
             if r["state"] == "superseded":
                 await self.erase(int(r["id"]))
             else:
-                await self.forget(int(r["id"]))
+                await self.erase(int(r["id"]))
         if rows:
             self.publish("DELETE", None, None, reason="expired", count=len(rows))
         return len(rows) + await self.purge_superseded()
diff --git a/sentient/memory/schema.sql b/sentient/memory/schema.sql
index f4c50aa3f..8322234e4 100644
--- a/sentient/memory/schema.sql
+++ b/sentient/memory/schema.sql
@@ -89,3 +89,17 @@ CREATE TABLE IF NOT EXISTS memory_forgets (
     at         TEXT NOT NULL
 );
 CREATE INDEX IF NOT EXISTS idx_memory_forgets_at ON memory_forgets(at);
+
+-- Cleanup an erase could not finish (#462): the memory row is gone but one derived step failed once (a locked
+-- file, a transient database error). One row per unfinished erase, replayed by memory upkeep until it succeeds.
+-- ``payload`` is JSON {ids, texts, refs} exactly as erase() computed them.
+CREATE TABLE IF NOT EXISTS pending_erase (
+    id      INTEGER PRIMARY KEY,
+    fact_id INTEGER NOT NULL,
+    payload TEXT NOT NULL,
+    at      TEXT NOT NULL,
+    -- 0 until the memory row itself is confirmed gone: upkeep only replays
+    -- completed erases, so a concurrent drain can never strip a live memory's
+    -- derived data while its delete is still running (#462).
+    complete INTEGER NOT NULL DEFAULT 0
+);
diff --git a/tests/memory/test_erase_retry.py b/tests/memory/test_erase_retry.py
new file mode 100644
index 000000000..10f98d69e
--- /dev/null
+++ b/tests/memory/test_erase_retry.py
@@ -0,0 +1,316 @@
+"""A cleanup step that fails during an erase is retried until it succeeds (#462).
+
+`erase()` deletes the memory itself first and then removes everything derived
+from it. One failing step (a locked file, a transient database error) used to
+drop the rest: the derived data stayed forever and nothing ever tried again.
+The unfinished cleanup is now recorded in `pending_erase` and replayed by
+memory upkeep (`drain_pending_erases`) until it succeeds.
+"""
+
+from __future__ import annotations
+
+import json
+from datetime import UTC, datetime, timedelta
+
+import pytest
+
+from sentient.store.db import now_iso
+
+HONDA = "Maya drives a red Honda City"
+
+
+async def test_a_cleanup_step_that_fails_once_is_retried_by_the_next_upkeep(app, monkeypatch):
+    mem, store = app.memory, app.store
+    tg = await store.create_session(channel="telegram")
+    await store.add_message(tg, "user", "I drive a red Honda City to work")
+    first = await mem.remember(HONDA, use_llm=False, session_id=tg)
+    honda = int(first["id"])
+    summary = await mem.episodic.add_summary(
+        tg, "Maya drives a red Honda City to the office.", await store.recent_messages(tg, 10)
+    )
+
+    real_cascade = mem._cascade
+    calls = []
+
+    async def fails_once(*args, **kwargs):
+        calls.append(args)
+        if len(calls) == 1:
+            raise RuntimeError("database is locked")
+        return await real_cascade(*args, **kwargs)
+
+    monkeypatch.setattr(mem, "_cascade", fails_once)
+
+    # The delete itself still succeeds: the memory row goes, the failing cleanup is recorded.
+    assert await mem.erase(honda) is True
+    assert len(calls) == 1
+    assert await mem.get_fact(honda) is None
+    assert await store.fetchone("SELECT id FROM summaries WHERE id = ?", (summary["id"],))
+    pending = await store.fetchall("SELECT id, fact_id, payload FROM pending_erase")
+    assert len(pending) == 1 and int(pending[0]["fact_id"]) == honda
+    payload = json.loads(pending[0]["payload"])
+    assert honda in payload["ids"] and HONDA in payload["texts"]
+
+    # The next upkeep run drains the queue: the retry replays the same cleanup,
+    # the derived data is gone and the wait is cleared.
+    # We simulate the hourly upkeep path (purge_expired + drain_pending_erases).
+    await app.memory.purge_expired()
+    assert await app.memory.drain_pending_erases() == 1
+    assert len(calls) == 2
+    assert await store.fetchone("SELECT id FROM summaries WHERE id = ?", (summary["id"],)) is None
+    assert not await store.fetchall("SELECT id FROM pending_erase")
+
+
+async def test_a_cleanup_that_keeps_failing_is_kept_and_retried_next_time(app, monkeypatch):
+    mem, store = app.memory, app.store
+    tg = await store.create_session(channel="telegram")
+    await store.add_message(tg, "user", "I drive a red Honda City to work")
+    first = await mem.remember(HONDA, use_llm=False, session_id=tg)
+    honda = int(first["id"])
+
+    async def always_fails(*args, **kwargs):
+        raise RuntimeError("database is locked")
+
+    monkeypatch.setattr(mem, "_cascade", always_fails)
+    assert await mem.erase(honda) is True
+    assert await mem.drain_pending_erases() == 0
+    assert len(await store.fetchall("SELECT id FROM pending_erase")) == 1  # kept for the next run
+
+    monkeypatch.undo()
+    assert await mem.drain_pending_erases() == 1
+    assert not await store.fetchall("SELECT id FROM pending_erase")
+
+
+async def test_a_clean_erase_records_nothing(app):
+    mem, store = app.memory, app.store
+    tg = await store.create_session(channel="telegram")
+    await store.add_message(tg, "user", "I drive a red Honda City to work")
+    first = await mem.remember(HONDA, use_llm=False, session_id=tg)
+    honda = int(first["id"])
+    assert await mem.erase(honda) is True
+    assert not await store.fetchall("SELECT id FROM pending_erase")
+
+
+async def test_upkeep_does_not_strip_a_memory_still_being_deleted(app, monkeypatch):
+    """A drain that lands while the delete is still running must do nothing (#462)."""
+    mem, store = app.memory, app.store
+    tg = await store.create_session(channel="telegram")
+    await store.add_message(tg, "user", "I drive a red Honda City to work")
+    honda = int((await mem.remember(HONDA, use_llm=False, session_id=tg))["id"])
+    summary = await mem.episodic.add_summary(
+        tg, "Maya drives a red Honda City to the office.", await store.recent_messages(tg, 10)
+    )
+
+    real_forget = mem.forget
+
+    async def forget_slow(fid, *, notify=False):
+        # A concurrent upkeep tick lands while the delete is suspended.
+        assert await mem.drain_pending_erases() == 0
+        # The summary is still there: the wait is not complete yet.
+        assert await store.fetchone("SELECT id FROM summaries WHERE id = ?", (summary["id"],))
+        return await real_forget(fid, notify=notify)
+
+    monkeypatch.setattr(mem, "forget", forget_slow)
+    assert await mem.erase(honda) is True
+    # Only the finished erase cleaned up: the summary is gone now.
+    assert await store.fetchone("SELECT id FROM summaries WHERE id = ?", (summary["id"],)) is None
+    assert not await store.fetchall("SELECT id FROM pending_erase")
+
+
+async def test_a_summary_vector_whose_delete_failed_goes_on_the_retry(app, monkeypatch):
+    """A vector whose delete failed once still goes on the retry (#462): its row ID is
+    recorded in the pending payload before the summary row can disappear."""
+    mem, store = app.memory, app.store
+    tg = await store.create_session(channel="telegram")
+    await store.add_message(tg, "user", "I drive a red Honda City to work")
+    honda = int((await mem.remember(HONDA, use_llm=False, session_id=tg))["id"])
+    summary = await mem.episodic.add_summary(
+        tg, "Maya drives a red Honda City to the office.", await store.recent_messages(tg, 10)
+    )
+
+    real_delete = mem.episodic.vec.delete
+    calls = []
+
+    async def delete_once(rowid):
+        calls.append(rowid)
+        if len(calls) == 1:
+            raise RuntimeError("database is locked")
+        return await real_delete(rowid)
+
+    monkeypatch.setattr(mem.episodic.vec, "delete", delete_once)
+    assert await mem.erase(honda) is True  # the cleanup fails at the first vector delete
+    # The vector went first, so the summary row survived the failed step.
+    row = await store.fetchone("SELECT rowid FROM summaries WHERE id = ?", (summary["id"],))
+    assert row is not None
+    pending = await store.fetchone("SELECT payload FROM pending_erase")
+    noted = json.loads(pending["payload"])["summary_vec_ids"][0]
+    assert noted[0] == int(row["rowid"]) and noted[1] == summary["id"]
+
+    monkeypatch.undo()
+    assert await mem.drain_pending_erases() == 1
+    assert await store.fetchone("SELECT rowid FROM summaries WHERE id = ?", (summary["id"],)) is None
+    assert not await store.fetchall("SELECT rowid FROM summaries_vec WHERE rowid = ?", (int(row["rowid"]),))
+
+
+async def test_a_retry_deletes_vectors_whose_summary_row_is_already_gone(app):
+    """Retries use the vector IDs stored in the payload, so a vector still goes even
+    when its summary row disappeared by another route (#462)."""
+    mem, store = app.memory, app.store
+    tg = await store.create_session(channel="telegram")
+    await store.add_message(tg, "user", "I drive a red Honda City to work")
+    honda = int((await mem.remember(HONDA, use_llm=False, session_id=tg))["id"])
+    summary = await mem.episodic.add_summary(
+        tg, "Maya drives a red Honda City to the office.", await store.recent_messages(tg, 10)
+    )
+    # The summary row disappears by another route, orphaning its vector.
+    row = await store.fetchone("SELECT rowid FROM summaries WHERE id = ?", (summary["id"],))
+    assert row is not None
+    await store.execute("DELETE FROM summaries WHERE id = ?", (summary["id"],))
+    await store.execute(
+        "INSERT INTO pending_erase(fact_id, payload, at, complete) VALUES(?,?,?,1)",
+        (
+            honda,
+            json.dumps({"ids": [honda], "texts": [], "refs": [], "summary_vec_ids": [int(row["rowid"])]}),
+            now_iso(),
+        ),
+    )
+
+    assert await mem.drain_pending_erases() == 1
+    assert not await store.fetchall("SELECT rowid FROM summaries_vec WHERE rowid = ?", (int(row["rowid"]),))
+    assert not await store.fetchall("SELECT id FROM pending_erase")
+
+
+async def test_hourly_upkeep_retries_cleanups_even_when_the_purge_fails(app, monkeypatch):
+    """A purge failure must not block the pending-erase retries (#462)."""
+    mem, store = app.memory, app.store
+    app.config.evolution.review_enabled = False
+    app.config.memory.summaries_enabled = False
+    app.config.evolution.curator_enabled = False
+    app.config.evolution.user_profile_updates = False
+    tg = await store.create_session(channel="telegram")
+    await store.add_message(tg, "user", "I drive a red Honda City to work")
+    honda = int((await mem.remember(HONDA, use_llm=False, session_id=tg))["id"])
+    summary = await mem.episodic.add_summary(
+        tg, "Maya drives a red Honda City to the office.", await store.recent_messages(tg, 10)
+    )
+
+    async def always_fails(*args, **kwargs):
+        raise RuntimeError("database is locked")
+
+    monkeypatch.setattr(mem, "_cascade", always_fails)
+    assert await mem.erase(honda) is True  # the wait is recorded for the next upkeep
+    monkeypatch.undo()
+
+    async def purge_raises():
+        raise RuntimeError("database is locked")
+
+    monkeypatch.setattr(app.memory, "purge_expired", purge_raises)
+    with pytest.raises(RuntimeError):
+        await app.evolution._tick()
+    # The purge failed, but the retry still ran and cleared the wait.
+    assert not await store.fetchall("SELECT id FROM pending_erase")
+    assert await store.fetchone("SELECT id FROM summaries WHERE id = ?", (summary["id"],)) is None
+
+
+async def test_a_retry_leaves_a_summary_that_reused_the_freed_rowid(app, monkeypatch):
+    """SQLite can hand a freed rowid to a new summary while a retry is still waiting;
+    the retry must not delete the newcomer (#462)."""
+    mem, store = app.memory, app.store
+    tg = await store.create_session(channel="telegram")
+    await store.add_message(tg, "user", "I drive a red Honda City to work")
+    honda = int((await mem.remember(HONDA, use_llm=False, session_id=tg))["id"])
+    summary = await mem.episodic.add_summary(
+        tg, "Maya drives a red Honda City to the office.", await store.recent_messages(tg, 10)
+    )
+    row = await store.fetchone("SELECT rowid FROM summaries WHERE id = ?", (summary["id"],))
+    assert row is not None
+    freed = int(row["rowid"])
+
+    def scrub_fails(erased):
+        raise OSError("MEMORY.md is locked by another program")
+
+    monkeypatch.setattr(mem, "on_erase", scrub_fails)
+    assert await mem.erase(honda) is True  # the cleanup fails after the summaries step
+    # The summaries step already ran: the original summary and its vector are gone,
+    # and the wait keeps the rowid (with its summary id) for the next upkeep.
+    assert await store.fetchone("SELECT id FROM summaries WHERE id = ?", (summary["id"],)) is None
+    assert not await store.fetchall("SELECT rowid FROM summaries_vec WHERE rowid = ?", (freed,))
+    pending = json.loads((await store.fetchone("SELECT payload FROM pending_erase"))["payload"])
+    assert pending["summary_vec_ids"] == [[freed, summary["id"]]]
+    monkeypatch.undo()
+
+    # A new, unrelated summary takes the freed rowid while the retry is waiting.
+    tea = await store.create_session(channel="telegram")
+    await store.add_message(tea, "user", "I prefer green tea with breakfast")
+    newcomer = await mem.episodic.add_summary(
+        tea, "Maya drinks green tea.", await store.recent_messages(tea, 10)
+    )
+    got = await store.fetchone("SELECT rowid FROM summaries WHERE id = ?", (newcomer["id"],))
+    assert got is not None and int(got["rowid"]) == freed  # the rowid was indeed reused
+
+    assert await mem.drain_pending_erases() == 1  # the retry runs and clears the wait
+    # The newcomer survived, row and vector both.
+    assert await store.fetchone("SELECT id FROM summaries WHERE id = ?", (newcomer["id"],))
+    assert await store.fetchall("SELECT rowid FROM summaries_vec WHERE rowid = ?", (freed,))
+    assert not await store.fetchall("SELECT id FROM pending_erase")
+
+
+async def test_a_wait_stranded_by_a_crashed_delete_is_replayed_once_the_memory_is_gone(app, monkeypatch):
+    """A forget() that raised after the fact row went stranded its wait at complete = 0
+    forever; upkeep now promotes such waits once the memory no longer exists (#462)."""
+    mem, store = app.memory, app.store
+    tg = await store.create_session(channel="telegram")
+    await store.add_message(tg, "user", "I drive a red Honda City to work")
+    honda = int((await mem.remember(HONDA, use_llm=False, session_id=tg))["id"])
+    summary = await mem.episodic.add_summary(
+        tg, "Maya drives a red Honda City to the office.", await store.recent_messages(tg, 10)
+    )
+
+    real_forget = mem.forget
+
+    async def forget_crashes(fid, *, notify=False):
+        assert await real_forget(fid, notify=notify)
+        raise RuntimeError("vector delete failed after the row went")
+
+    monkeypatch.setattr(mem, "forget", forget_crashes)
+    with pytest.raises(RuntimeError):
+        await mem.erase(honda)
+    # The memory is gone, but its wait is stranded at complete = 0.
+    assert await mem.get_fact(honda) is None
+    stranded = await store.fetchone("SELECT id, complete FROM pending_erase")
+    assert stranded is not None and int(stranded["complete"]) == 0
+    monkeypatch.undo()
+
+    # A fresh stranded wait is left alone: the erase could still be running.
+    assert await mem.drain_pending_erases() == 0
+    assert await store.fetchone("SELECT id FROM summaries WHERE id = ?", (summary["id"],))
+    # An old stranded wait whose memory no longer exists is replayed and cleared.
+    await store.execute(
+        "UPDATE pending_erase SET at = ? WHERE id = ?",
+        ((datetime.now(UTC) - timedelta(hours=2)).isoformat(), int(stranded["id"])),
+    )
+    assert await mem.drain_pending_erases() == 1
+    assert await store.fetchone("SELECT id FROM summaries WHERE id = ?", (summary["id"],)) is None
+    assert not await store.fetchall("SELECT id FROM pending_erase")
+
+
+async def test_an_unrecordable_cleanup_aborts_the_erase_and_keeps_the_memory(app, monkeypatch):
+    """When the pending-cleanup row cannot be recorded, the erase aborts before
+    deleting anything, so the REST route can tell a kept memory (503) from a
+    missing one (404) by looking it up again (#462)."""
+    mem, store = app.memory, app.store
+    tg = await store.create_session(channel="telegram")
+    await store.add_message(tg, "user", "I drive a red Honda City to work")
+    honda = int((await mem.remember(HONDA, use_llm=False, session_id=tg))["id"])
+
+    real_execute = store.execute
+
+    async def execute_failing_on_pending_insert(sql, *args, **kwargs):
+        if sql.startswith("INSERT INTO pending_erase"):
+            raise RuntimeError("database is locked")
+        return await real_execute(sql, *args, **kwargs)
+
+    monkeypatch.setattr(store, "execute", execute_failing_on_pending_insert)
+    assert await mem.erase(honda) is False  # aborted: the memory must not go without its retry
+    monkeypatch.undo()
+    assert await mem.get_fact(honda) is not None  # still listed, so the route answers 503, not 404
+    assert not await store.fetchall("SELECT id FROM pending_erase")