fix(persistence): harden chat and memory storage

README.md

@@ -402,6 +402,24 @@ What is kept, all beside the database files in that folder:
   `.corrupt-<id>` file and a valid backup looks like exactly that interrupted
   recovery and is restored, so to start over on purpose remove the backups and
   the `.corrupt-<id>` files along with it (they hold copies of your chats).
+- Deleting chats frees space inside `cortex_db.sqlite` but not on disk. At
+  launch, once that launch's backup has been written and verified, Cortex hands
+  the space back when more than a quarter of the file is free. A database
+  created by this release does it in small steps; an older one is rewritten
+  once (which needs roughly twice its size free on the data drive and the
+  temporary-files drive, and is skipped when there is not, when the database
+  holds over 1 GiB, or when it takes over 20 seconds). A skipped or interrupted
+  pass changes nothing. One that was skipped for lack of room or because
+  another program had the database open is tried again on the next launch; a
+  rewrite that failed or ran out of its 20 seconds is not tried again for seven
+  days (a small `cortex_db.sqlite.reclaim-backoff` file beside the database
+  holds only the time it failed; deleting it lifts the wait), so a very large
+  old database does not add those seconds to every launch.
+- Chats imported from the old JSON `chat_history` folder are moved to
+  `chat_history_migrated_<time>` and files that could not be read to
+  `chat_history/quarantine`. Once the `chat_history` folder is completely empty
+  it is renamed to `chat_history.retired` (never deleted, so it is yours to
+  remove); anything still in it is left where it is.
 
 Cortex needs write-ahead logging, so the data directory has to be on a local
 drive SQLite can use it on; some network, cloud-synced and removable drives

backend/cortex_backend/api/routers/chats.py

@@ -367,6 +367,7 @@ async def regenerate_chat(
                         reservation=reservation,
                         target_message_id=payload.message_id,
                         transcript=messages,
+                        transcript_revision=current_revision,
                     )
                 finally:
                     request.app.state.jobs.abort_reservation(

backend/cortex_backend/api/routes.py

@@ -13,10 +13,12 @@
 from pathlib import Path
 from threading import Event as ThreadEvent, Thread
 import asyncio
+import errno
 import hashlib
 import json
 import logging
 import re
+import sqlite3
 from typing import Any, NoReturn, cast
 from uuid import uuid4
 
@@ -311,17 +313,24 @@ async def _start_generation_job(
     reservation: JobReservation | None = None,
     target_message_id: str | None = None,
     transcript: Sequence[Mapping[str, Any]] | None = None,
+    transcript_revision: int | None = None,
 ) -> tuple[JobSnapshot, str | None]:
     """Atomically admit, prepare, and run one authoritative generation job.
 
     A regeneration passes ``transcript``, the messages its route already read to
-    validate the target, so the chat is not read again. A new turn passes
-    nothing: admission needs only the chat's overview, and preparation reads
-    the transcript once, because the model needs it as history. Either way the
-    generation service is handed its history and never loads it itself.
+    validate the target, and ``transcript_revision``, the revision that read
+    returned (a regeneration changes a message without changing how many there
+    are, so the revision cannot be derived from the transcript), so the chat is
+    not read again. A new turn passes neither: admission needs only the chat's
+    overview, and preparation reads the transcript once, because the model
+    needs it as history. Either way the generation service is handed its
+    history and never loads it itself.
     """
-    if (target_message_id is None) != (transcript is None):
-        raise ValueError("a regeneration needs its transcript and a new turn has none")
+    regenerating = target_message_id is not None
+    if regenerating != (transcript is not None) or regenerating != (transcript_revision is not None):
+        raise ValueError(
+            "a regeneration needs its transcript and its revision and a new turn has neither"
+        )
     jobs = request.app.state.jobs
     candidate_thread_id = payload.thread_id or uuid4().hex
     if reservation is None:
@@ -347,12 +356,16 @@ async def _start_generation_job(
     try:
         target_position = -1
         target_role = ""
-        if transcript is not None and target_message_id is not None:
+        if (
+            transcript is not None
+            and target_message_id is not None
+            and transcript_revision is not None
+        ):
             # The route read the chat and answered 404 if it was missing.
             target_position, target_role = _regeneration_target(
                 transcript, target_message_id
             )
-            admission_revision = len(transcript)
+            admission_revision = transcript_revision
         else:
             # Only the revision is needed to admit a new turn, so read the
             # overview rather than every message of what may be a long thread.
@@ -459,8 +472,8 @@ def prepare() -> Mapping[str, Any]:
                 # The route validated this target against the transcript it
                 # read. That still holds as long as nothing was appended since:
                 # a chat only grows, and this reserved job is the only writer
-                # that replaces a reply. The message count is the revision, so
-                # the overview answers it without reading the transcript again.
+                # that replaces a reply. The overview carries the revision, so
+                # it answers this without reading the transcript again.
                 overview = deps.chats.get_chat_overview(thread_id)
                 if overview is None:
                     raise ChatDomainError("Chat not found.", code="not_found")
@@ -477,7 +490,7 @@ def prepare() -> Mapping[str, Any]:
             # and that write is guarded by the revision it was read at.
             current_chat = deps.chats.get_chat(thread_id)
             existing = list(current_chat.get("messages", ())) if current_chat else []
-            if len(existing) != admission_revision:
+            if (chat_revision(current_chat) if current_chat else 0) != admission_revision:
                 raise ChatDomainError(
                     "This chat changed. Reload it before generating again.",
                     code="stale_revision",
@@ -1467,6 +1480,34 @@ def _raise_job_error(exc: Exception) -> NoReturn:
     ) from exc
 
 
+# Windows reports a full volume as ERROR_DISK_FULL or ERROR_HANDLE_DISK_FULL,
+# which Python only sometimes maps to ENOSPC.
+_WINDOWS_DISK_FULL = frozenset({39, 112})
+
+
+def _is_disk_full(exc: BaseException) -> bool:
+    """Whether a repository failure, or anything it was raised from, is a full disk.
+
+    A store wraps what went wrong (``PersistenceError.cause``, or ``raise ...
+    from``), so the chain is followed. SQLite says "database or disk is full"
+    for SQLITE_FULL, and the file-copy paths raise ENOSPC.
+    """
+    seen: set[int] = set()
+    current: BaseException | None = exc
+    while current is not None and id(current) not in seen:
+        seen.add(id(current))
+        if isinstance(current, OSError) and (
+            current.errno in (errno.ENOSPC, getattr(errno, "EDQUOT", errno.ENOSPC))
+            or getattr(current, "winerror", None) in _WINDOWS_DISK_FULL
+        ):
+            return True
+        if isinstance(current, sqlite3.OperationalError) and "disk is full" in str(current).lower():
+            return True
+        cause = getattr(current, "cause", None)
+        current = cause if isinstance(cause, BaseException) else current.__cause__
+    return False
+
+
 def _raise_repository_error(operation: str, exc: Exception) -> NoReturn:
     if isinstance(exc, HTTPException):
         raise exc
@@ -1476,6 +1517,13 @@ def _raise_repository_error(operation: str, exc: Exception) -> NoReturn:
     # reports "Could not list chats" can be matched to the failure.
     request_id = current_request_id()
     log_failure(logger, f"Cortex API {operation} failed", exc, request_id=request_id)
+    if _is_disk_full(exc):
+        # Not the server's fault and not a bug: nothing was saved because there
+        # is no room, and the user can fix that.
+        raise HTTPException(
+            status_code=status.HTTP_507_INSUFFICIENT_STORAGE,
+            detail=f"Could not {operation} because the disk is full. Free some disk space and try again.",
+        ) from exc
     raise HTTPException(
         status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
         detail=f"Could not {operation}." + (f" (Request ID: {request_id})" if request_id else ""),

backend/cortex_backend/repositories/chats.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 from copy import deepcopy
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 from threading import RLock
 from typing import Any, Protocol
 
@@ -104,6 +104,23 @@ def replace_message(
     ) -> None: ...
 
 
+def _typed_failure(exc: Exception) -> Exception:
+    """The repository error for an outcome the store reports by ``operation``.
+
+    The SQLite manager raises one ``PersistenceError`` type and names what
+    happened in ``operation``; callers and the in-memory double speak in typed
+    errors. Anything else comes back unchanged.
+    """
+    operation = getattr(exc, "operation", None)
+    if operation == "chat_revision_conflict":
+        return ChatRevisionConflict(str(exc))
+    if operation == "chat_not_found":
+        return ChatNotFound("Chat does not exist.")
+    if operation == "message_not_found":
+        return MessageNotFound("Message does not exist.")
+    return exc
+
+
 class LegacyDatabaseChatRepository:
     """Adapt the merged SQLite manager without importing the legacy module."""
 
@@ -161,17 +178,24 @@ def add_message(
         try:
             result = self._database.add_message(thread_id, role, content, **kwargs)
         except Exception as exc:
-            if getattr(exc, "operation", None) == "chat_revision_conflict":
-                raise ChatRevisionConflict(str(exc)) from exc
-            raise
+            typed = _typed_failure(exc)
+            if typed is exc:
+                raise
+            raise typed from exc
         if result is None:
             chat = self._database.load_chat(thread_id) or {}
             messages = chat.get("messages", [])
             return str(messages[-1].get("id", len(messages) - 1))
         return str(result)
 
     def rename_chat(self, thread_id: str, title: str) -> None:
-        self._database.update_chat_title(thread_id, title)
+        try:
+            self._database.update_chat_title(thread_id, title)
+        except Exception as exc:
+            typed = _typed_failure(exc)
+            if typed is exc:
+                raise
+            raise typed from exc
 
     def delete_chat(self, thread_id: str) -> None:
         self._database.delete_chat(thread_id)
@@ -220,9 +244,10 @@ def replace_message(
                 original_content=original_content,
             )
         except Exception as exc:
-            if getattr(exc, "operation", None) == "chat_revision_conflict":
-                raise ChatRevisionConflict(str(exc)) from exc
-            raise
+            typed = _typed_failure(exc)
+            if typed is exc:
+                raise
+            raise typed from exc
 
 
 class InMemoryChatRepository:
@@ -241,6 +266,8 @@ def __init__(self, chats: list[dict[str, Any]] | None = None):
                     message["id"] = self._new_message_id()
                 else:
                     self._advance_message_counter(message["id"])
+            # A seeded chat starts from its message count, as a stored one does.
+            copied.setdefault("revision", len(copied.get("messages", [])))
             self._chats[str(chat["id"])] = copied
 
     def _new_message_id(self) -> str:
@@ -289,7 +316,9 @@ def create_group(self, group_id: str, name: str) -> None:
             self._groups[group_id] = {
                 "id": group_id,
                 "name": name,
-                "position": len(self._groups),
+                # After the highest position, not the count: a delete leaves
+                # gaps, and the count would then reuse a position still taken.
+                "position": max((group["position"] for group in self._groups.values()), default=-1) + 1,
                 "collapsed": False,
                 "timestamp": self._timestamp(),
             }
@@ -340,7 +369,7 @@ def get_chat_overview(self, thread_id: str) -> dict[str, Any] | None:
                 "title": chat.get("title"),
                 "timestamp": chat.get("timestamp"),
                 "group_id": chat.get("group_id"),
-                "revision": len(chat.get("messages", ())),
+                "revision": chat["revision"],
             }
 
     def create_chat(self, thread_id: str, title: str) -> None:
@@ -352,6 +381,7 @@ def create_chat(self, thread_id: str, title: str) -> None:
                 "title": title,
                 "timestamp": self._timestamp(),
                 "group_id": None,
+                "revision": 0,
                 "messages": [],
             }
 
@@ -363,7 +393,7 @@ def _check_expected_revision(
             return
         if type(expected_revision) is not int or expected_revision < 0:
             raise ValueError("expected_revision must be a non-negative integer")
-        actual_revision = len(chat.get("messages", ()))
+        actual_revision = chat["revision"]
         if actual_revision != expected_revision:
             raise ChatRevisionConflict(
                 f"Chat revision changed (expected {expected_revision}, found {actual_revision})."
@@ -402,23 +432,27 @@ def add_message(
                     "role": role,
                     "content": content,
                     "timestamp": self._timestamp(),
-                    "sources": deepcopy(sources),
+                    # Empty means absent, as in the database, which stores no
+                    # column for an empty list or mapping.
+                    "sources": deepcopy(sources) or None,
                     "thoughts": _assistant_thoughts(role, thoughts),
-                    "attachments": deepcopy(attachments),
-                    "stats": deepcopy(stats) if role == "assistant" else None,
+                    "attachments": deepcopy(attachments) or None,
+                    "stats": (deepcopy(stats) or None) if role == "assistant" else None,
                     "original_content": original_content if role == "assistant" else None,
                 }
             )
             chat["timestamp"] = self._timestamp()
+            chat["revision"] += 1
             return message_id
 
     def rename_chat(self, thread_id: str, title: str) -> None:
         with self._lock:
             chat = self._chats.get(thread_id)
             if chat is None:
                 raise ChatNotFound("Chat does not exist.")
+            # Only the title: a rename is not activity, so the chat keeps its
+            # place in the recency-ordered list, as it does in the database.
             chat["title"] = title
-            chat["timestamp"] = self._timestamp()
 
     def delete_chat(self, thread_id: str) -> None:
         with self._lock:
@@ -436,19 +470,28 @@ def fork_chat(self, thread_id: str, message_id: str, new_thread_id: str) -> None
                 )
             except StopIteration as exc:
                 raise MessageNotFound("Message does not exist.") from exc
+            if new_thread_id in self._chats:
+                raise ChatRepositoryError("Chat already exists.")
             copied = deepcopy(source)
             copied["id"] = new_thread_id
             copied["title"] = f"Fork of {source.get('title') or 'Untitled Chat'}"
             copied["timestamp"] = self._timestamp()
+            # A fork is filed nowhere until the user files it.
+            copied["group_id"] = None
             copied["messages"] = []
-            for message in source["messages"][: position + 1]:
+            forked_at = datetime.now(timezone.utc)
+            for index, message in enumerate(source["messages"][: position + 1]):
                 copied["messages"].append(
                     {
                         **deepcopy(message),
                         "id": self._new_message_id(),
-                        "timestamp": self._timestamp(),
+                        # Each message keeps the time it was really sent; the
+                        # offset only orders messages that never had one.
+                        "timestamp": message.get("timestamp")
+                        or (forked_at + timedelta(microseconds=index)).isoformat(),
                     }
                 )
+            copied["revision"] = len(copied["messages"])
             self._chats[new_thread_id] = copied
 
     def replace_message(
@@ -475,14 +518,17 @@ def replace_message(
                         raise ChatRepositoryError("Only assistant messages can be replaced.")
                     message.update(
                         content=content,
-                        sources=deepcopy(sources),
+                        sources=deepcopy(sources) or None,
                         thoughts=thoughts,
-                        stats=deepcopy(stats),
+                        stats=deepcopy(stats) or None,
                         original_content=original_content,
                         timestamp=self._timestamp(),
                     )
                     if attachments is not None:
                         message["attachments"] = deepcopy(attachments)
                     chat["timestamp"] = self._timestamp()
+                    # The count did not change, the reply did: the revision
+                    # moves so a second regeneration from the same state is stale.
+                    chat["revision"] += 1
                     return
             raise MessageNotFound("Message does not exist.")