fix(memory): retry an erase cleanup until it succeeds

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.

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

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

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}

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:

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,21 +1132,75 @@ 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:
                     made_of = {str(m) for m in json.loads(r["message_ids"] or "[]")}
                 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()

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
+);