fix(security): R-9 — a purge destroys only what was retired first

.claude/rules/file-map.md

@@ -274,6 +274,8 @@ Full module index for LocalChat. **Keep this current** — update in the same co
 | `tests/unit/test_jwt_library_is_pyjwt.py` | P2-6 — a `python-jose`-signed token still verifies after the swap to PyJWT (the fixture cannot be re-made once jose is uninstalled), and neither lock pins `python-jose` or `ecdsa` again |
 | `tests/unit/test_local_folder_least_privilege.py` | P0-3 — the `local_folder` connector reaches only inside `CONNECTOR_LOCAL_ROOTS`, and only a global admin may create or repoint one. Covers `/etc`, `..`, a symlink out of an allowed root, and a sibling sharing a path prefix (`commonpath`, not `startswith`) |
 | `tests/unit/test_object_authorization_matrix.py` | P0-1 — every route addressing an object by id is scoped to a workspace. Walks the AST of `src/` and fails on any call to a workspace-scoped database method that omits `scope=`, so a *new* route cannot repeat C1/C2 |
+| `tests/unit/test_purge_requires_retired.py` | R-9 (QP-9 review) — `require_retired` locks the row and refuses a live or missing one, `purge_refusals` makes that a 409 or a 404, and all five purge routes send it |
+| `tests/integration/test_purge_requires_retired.py` | R-9 against Postgres, for the document, conversation, connector, workspace and user purges: a live row is refused and survives, a missing one is not found, a retired one is destroyed |
 | `tests/unit/test_purge_preconditions.py` | The Clark-Wilson purge TPs — a cited conversation or a user with memberships is refused before any DELETE |
 | `tests/unit/test_token_accounting.py` | OBS-1a — a streamed chat against `fake_ollama` adds exactly the tokens its `done` object reported, the cloud path adds a canned OpenAI-compatible stream's tokens and its cost at the configured prices, a call without both prices still counts tokens, and no counter carries more than the model and path labels |
 | `tests/unit/test_degradation_paths.py` | P2-4b — the exception paths that change behaviour: the four narrowed handlers absorb a malformed timestamp and let anything else escape, and the settings page still renders when its stats do not (the bug P2-4b itself introduced) |

CHANGELOG.md

@@ -45,6 +45,12 @@ OpenAI, `https://api.openai.com/v1`) and drop any `provider/` prefix from `CLOUD
   `CLOUD_PRICE_OUTPUT_PER_MTOK`; without both, only tokens are counted. A provider that is
   not OpenAI-shaped (Anthropic's native API, Vertex) is no longer reachable, as ADR-4
   accepted.
+- **A purge destroys only what was retired first (R-9).** The single-object purges of a
+  document, conversation, workspace, connector and user checked references and nothing
+  else, so a live row could be hard-deleted in one call, skipping the reversible step
+  before it, and the only active admin could be purged. Each purge now locks the row and
+  refuses unless it is retired: a live row is a 409, a missing one a 404 (a missing
+  document used to report success, a missing connector a 409 about synced documents).
 - **Every installation-wide scope carries a reason (R-5b).** A unit test walks `src/`, `mcp_servers/` and `plugins/` and fails on a function that passes `ALL_WORKSPACES` without a recorded reason, on a reason whose function no longer passes it, and on an import that renames it. The 44 functions that pass it today were each checked against every caller that can reach them; none lets a workspace user reach another workspace's rows.
 - **GraphRAG's query expansion reads only the caller's workspace (R-12).** Entities are shared across workspaces, so with `GRAPH_RAG_ENABLED` on, the 1-hop lookup added names that co-occurred in any workspace's documents, retired ones included, to the caller's query. A co-occurrence now counts only through a live document of the scope. GraphRAG is off by default.
 - **A password change ends the account's other sessions (`SECURITY.md` item 13).** A

CLAUDE.md

@@ -128,7 +128,7 @@ Hard-deleting a CDI while other data still holds its ID breaks IVPs — citation
   `conversation_messages` has neither (a message goes with its conversation).
 - A "delete" TP sets `deleted_at`; it never issues `DELETE FROM` on a CDI.
 - All SELECT queries on CDIs filter `WHERE deleted_at IS NULL`.
-- A hard purge is a *separate, explicitly authorized TP* with a precondition: no active references exist. It is never the default operation.
+- A hard purge is a *separate, explicitly authorized TP* with a precondition: the row is retired, and no active references exist. It is never the default operation.
 
 **Audit trail** — state changes on CDIs must be traceable: who changed what, when. `deleted_at` + `deleted_by` is the minimum required.
 

SECURITY.md

@@ -334,7 +334,7 @@ fails on the code before it.
 | R-6 | Medium | `GET /settings` rendered installation-wide admin statistics for any caller | **Fixed** (#507) — `tests/unit/test_settings_page_stats_are_admin_only.py` |
 | R-7 | Low | A connector's sync history and manual sync took the id alone | **Fixed** (#510) — `TestAConnectorOutsideTheScopeIsNotFound` |
 | R-8 | Low | Feedback and annotations accept conversation and chunk ids from another workspace; a viewer can make any document un-purgeable | **Fixed** — every id a rating or annotation names is looked up in the caller's scope first, the global tier's chunks included except for a workspace API key; the message id too, which the review did not name. `tests/integration/test_object_authorization_over_the_wire.py`, `tests/integration/test_rating_ids_are_in_scope.py` |
-| R-9 | Low | Single-object purges do not require the row to be retired first | **Open** (admin-only). Includes a user: purge has no last-admin check, so the only active admin, with no memberships, can be hard-deleted (found by the review of #525) |
+| R-9 | Low | Single-object purges do not require the row to be retired first | **Fixed** — each of the five purges locks its row and refuses unless it is retired, inside its own transaction: a live row is a 409 and survives, a missing one a 404. A user purge thereby cannot reach an active admin. `tests/unit/test_purge_requires_retired.py`, `tests/integration/test_purge_requires_retired.py` |
 | R-10 | Low | Admin user routes take raw dicts; the last-admin count ignores `is_active` | **Fixed** — the three routes validate a Pydantic body, so a string `"false"` no longer passes the last-admin rule (Postgres stored it as false), and the rule counts active admins only. The env-var admin's bootstrap check still counts live admins. `tests/unit/test_admin_user_bodies_are_validated.py`, `tests/integration/test_admin_count_is_of_active_admins.py`. Known residual: the rule reads the count and writes on separate connections, so the last two active admins deactivating each other at the same moment both succeed; the repair is SQL by hand (older than R-10) |
 | R-11 | Low | The image's source is owned by the runtime user | **Open** |
 | R-12 | Low | GraphRAG's entity expansion is not scoped (off by default) | **Fixed** — a co-occurrence counts only through a live document of the caller's workspace; retired documents contributed too. `tests/integration/test_graph_expansion_scope.py` |

docs/ROADMAP.md

@@ -1925,7 +1925,7 @@ From `SECURITY.md`'s QP-9 register, in this order:
    migration 0024).
 3. ✅ R-10: admin user routes take raw dicts; the last-admin count ignores `is_active`.
 4. ✅ R-5b: a ratchet over the `ALL_WORKSPACES` call sites, each with a recorded reason.
-5. R-9, R-11, R-12 ✅: purge preconditions, the image's file ownership, GraphRAG's scope.
+5. R-9 ✅, R-11, R-12 ✅: purge preconditions, the image's file ownership, GraphRAG's scope.
 
 #### QP-11 — Replace `litellm` (ADR-4) ✅ (built 2026-11-07, taken ahead of the rest of QP-10)
 

src/db/connection.py

@@ -252,6 +252,28 @@ class DatabaseUnavailableError(Exception):
     pass
 
 
+class NotRetiredError(Exception):
+    """A purge named a row that is missing, or still live: Destroy follows Retire (R-9)."""
+
+    def __init__(self, table: str, *, missing: bool) -> None:
+        super().__init__(f"{table} row is {'missing' if missing else 'not retired'}")
+        self.missing = missing
+
+
+def require_retired(cur: Any, table: str, row_id: object) -> None:
+    """Lock the row and refuse unless it is retired, inside the purge's own transaction.
+
+    Locking it means a restore cannot slip in between this check and the DELETE.
+    """
+    cur.execute(
+        sql.SQL("SELECT deleted_at IS NOT NULL FROM {} WHERE id = %s FOR UPDATE").format(sql.Identifier(table)),
+        (row_id,),
+    )
+    row = cur.fetchone()
+    if row is None or not row[0]:
+        raise NotRetiredError(table, missing=row is None)
+
+
 # ============================================================================
 # PGVECTOR TYPE ADAPTERS FOR PSYCOPG3
 # ============================================================================

src/db/connectors.py

@@ -14,7 +14,7 @@
 
 from ..utils.logging_config import get_logger
 from ..utils.scope import ALL_WORKSPACES, Scope, scope_predicate
-from .connection import DatabaseConnection, DatabaseUnavailableError
+from .connection import DatabaseConnection, DatabaseUnavailableError, require_retired
 
 logger = get_logger(__name__)
 
@@ -164,6 +164,7 @@ def purge(self, connector_id: str) -> bool:
             raise DatabaseUnavailableError("Cannot purge connector: DB not connected")
         with self._db.get_connection(scope=ALL_WORKSPACES) as conn:
             with conn.cursor() as cur:
+                require_retired(cur, "connectors", connector_id)
                 cur.execute(
                     "SELECT 1 FROM documents WHERE source_id = %s AND deleted_at IS NULL LIMIT 1",
                     (connector_id,),

src/db/conversations.py

@@ -7,7 +7,7 @@
 from ..utils.encryption import encrypt as _encrypt
 from ..utils.logging_config import get_logger, sanitize_log_value
 from ..utils.scope import ALL_WORKSPACES, Scope, scope_predicate
-from .connection import DatabaseConnection, DatabaseUnavailableError
+from .connection import DatabaseConnection, DatabaseUnavailableError, require_retired
 
 logger = get_logger(__name__)
 
@@ -364,6 +364,7 @@ def purge(self, conversation_id: str) -> bool:
 
         with self._db.get_connection(scope=ALL_WORKSPACES) as conn:
             with conn.cursor() as cursor:
+                require_retired(cursor, "conversations", conversation_id)
                 cursor.execute(
                     "SELECT 1 FROM memories WHERE source_conv = %s LIMIT 1",
                     (conversation_id,),

src/db/documents.py

@@ -9,7 +9,7 @@
 from ..utils.sanitization import escape_sql_like
 from ..utils.scope import ALL_WORKSPACES, Scope, scope_predicate
 from ..utils.text import content_terms
-from .connection import DatabaseConnection, DatabaseUnavailableError
+from .connection import DatabaseConnection, DatabaseUnavailableError, require_retired
 
 if TYPE_CHECKING:
     import psycopg
@@ -232,7 +232,7 @@ def retire(
 
     def purge(self, doc_id: int) -> bool:
         """
-        Hard-delete only when no chunk_stats row exists for any of this document's chunks.
+        Hard-delete a retired document only when no chunk_stats row exists for any of its chunks.
         Returns False when citations block the purge; True on success.
         """
         if not self._db.is_connected:
@@ -241,6 +241,7 @@ def purge(self, doc_id: int) -> bool:
         logger.debug("Purge requested for document ID: %s", doc_id)
         with self._db.get_connection(scope=ALL_WORKSPACES) as conn:
             with conn.cursor() as cursor:
+                require_retired(cursor, "documents", doc_id)
                 # Citation precondition: any chunk that was retrieved blocks the purge.
                 cursor.execute(
                     """

src/db/users.py

@@ -10,7 +10,7 @@
 
 from ..utils.logging_config import get_logger, sanitize_log_value
 from ..utils.scope import ALL_WORKSPACES
-from .connection import DatabaseConnection, DatabaseUnavailableError
+from .connection import DatabaseConnection, DatabaseUnavailableError, require_retired
 
 logger = get_logger(__name__)
 
@@ -228,11 +228,12 @@ def retire(self, user_id: str, deleted_by: str | None = None) -> bool:
                 return bool(cur.rowcount > 0)
 
     def purge(self, user_id: str) -> bool:
-        """Hard-delete a user if no workspace memberships exist. Returns False when blocked."""
+        """Hard-delete a retired user if no workspace memberships exist. Returns False when blocked."""
         if not self._db.is_connected:
             raise DatabaseUnavailableError("Cannot purge user: DB not connected")
         with self._db.get_connection(scope=ALL_WORKSPACES) as conn:
             with conn.cursor() as cur:
+                require_retired(cur, "users", user_id)
                 cur.execute(
                     "SELECT 1 FROM workspace_members WHERE user_id = %s LIMIT 1",
                     (user_id,),

src/db/workspaces.py

@@ -10,7 +10,7 @@
 
 from ..utils.logging_config import get_logger
 from ..utils.scope import ALL_WORKSPACES
-from .connection import DatabaseConnection, DatabaseUnavailableError
+from .connection import DatabaseConnection, DatabaseUnavailableError, require_retired
 
 logger = get_logger(__name__)
 
@@ -189,6 +189,7 @@ def purge(self, workspace_id: str) -> bool:
             raise DatabaseUnavailableError("Cannot purge workspace: DB not connected")
         with self._db.get_connection(scope=ALL_WORKSPACES) as conn:
             with conn.cursor() as cur:
+                require_retired(cur, "workspaces", workspace_id)
                 cur.execute(
                     "SELECT 1 FROM documents WHERE workspace_id = %s AND deleted_at IS NULL LIMIT 1",
                     (workspace_id,),

src/routes_fastapi/auth_routes.py

@@ -308,6 +308,8 @@ def purge_user(user_id: str, users: Users, admin_user_id: Annotated[str, Depends
                 status_code=409,
             )
         return {"success": True}
+    except ServiceError as exc:
+        return _refused(exc)
     except Exception:
         logger.exception("[Users] purge error")
         return JSONResponse({"success": False, "message": _ERR_INTERNAL}, status_code=500)

src/routes_fastapi/connector_routes.py

@@ -171,6 +171,8 @@ def purge_connector(
                 status_code=409,
             )
         return {"success": True}
+    except ServiceError as exc:
+        return JSONResponse({"success": False, "message": exc.message}, status_code=exc.status_code)
     except Exception:
         logger.exception("[Connectors] purge error")
         return JSONResponse({"success": False, "message": _ERR_INTERNAL}, status_code=500)

src/routes_fastapi/memory_routes.py

@@ -10,6 +10,7 @@
 
 from ..security_fastapi import get_current_user_id, require_admin_dep
 from ..services.conversations import ConversationsService
+from ..services.errors import ServiceError
 from ..utils.logging_config import get_logger
 from ..utils.workspace import get_scope, get_workspace_id
 from ._authz import deny as _deny
@@ -195,7 +196,10 @@ def purge_conversation(
     conversations: Conversations,
     _admin: Annotated[str, Depends(require_admin_dep)],
 ) -> Any:
-    purged = conversations.purge(conversation_id)
+    try:
+        purged = conversations.purge(conversation_id)
+    except ServiceError as exc:
+        return JSONResponse({"success": False, "message": exc.message}, status_code=exc.status_code)
     if not purged:
         return JSONResponse(
             {

src/routes_fastapi/workspace_routes.py

@@ -181,6 +181,8 @@ def purge_workspace(
                 status_code=409,
             )
         return {"success": True}
+    except ServiceError as exc:
+        return JSONResponse({"success": False, "message": exc.message}, status_code=exc.status_code)
     except Exception:
         logger.exception("[Workspaces] purge error")
         return JSONResponse({"success": False, "message": _ERR_INTERNAL}, status_code=500)

src/services/connectors.py

@@ -22,7 +22,7 @@
 
 from ..utils.logging_config import get_logger
 from ..utils.scope import Scope
-from .errors import ServiceError
+from .errors import ServiceError, purge_refusals
 
 if TYPE_CHECKING:
     from ..connectors.base import BaseConnector
@@ -167,7 +167,8 @@ def retire(self, connector_id: str, actor: str | None, scope: Scope) -> bool:
         return retired
 
     def purge(self, connector_id: str) -> bool:
-        return self._db.connectors.purge(connector_id)
+        with purge_refusals("Connector"):
+            return self._db.connectors.purge(connector_id)
 
     def sync_history(self, connector_id: str, limit: int, scope: Scope) -> list[dict[str, Any]] | None:
         """The connector's sync log, or None when *scope* has no such connector.

src/services/conversations.py

@@ -11,6 +11,7 @@
 from typing import TYPE_CHECKING, Any
 
 from ..utils.scope import Scope
+from .errors import purge_refusals
 
 if TYPE_CHECKING:
     from ..db import Database
@@ -57,7 +58,8 @@ def retire_all(self, actor: str | None, scope: Scope) -> int:
         return self._db.conversations.retire_all(deleted_by=_actor(actor), scope=scope)
 
     def purge(self, conversation_id: str) -> bool:
-        return self._db.conversations.purge(conversation_id)
+        with purge_refusals("Conversation"):
+            return self._db.conversations.purge(conversation_id)
 
 
 def _actor(user_id: str | None) -> str | None:

src/services/documents.py

@@ -15,7 +15,7 @@
 from ..db.connection import DatabaseUnavailableError
 from ..utils.logging_config import get_logger
 from ..utils.scope import Scope
-from .errors import ServiceError
+from .errors import ServiceError, purge_refusals
 
 if TYPE_CHECKING:
     from ..db import Database
@@ -80,7 +80,7 @@ def retire_all(self, actor: str | None, scope: Scope) -> int:
             return self._db.documents.retire_all(scope=scope, deleted_by=_actor(actor))
 
     def purge(self, doc_id: int) -> bool:
-        with _database_required():
+        with _database_required(), purge_refusals("Document"):
             return self._db.documents.purge(doc_id)
 
     def purge_all(self) -> int:

src/services/errors.py

@@ -6,6 +6,11 @@
 
 from __future__ import annotations
 
+from collections.abc import Iterator
+from contextlib import contextmanager
+
+from ..db.connection import NotRetiredError
+
 
 class ServiceError(Exception):
     """A refusal the caller should see. ``message`` is safe to send as-is."""
@@ -14,3 +19,14 @@ def __init__(self, status_code: int, message: str) -> None:
         super().__init__(message)
         self.status_code = status_code
         self.message = message
+
+
+@contextmanager
+def purge_refusals(what: str) -> Iterator[None]:
+    """A purge of a missing row is a 404; of a live one, a 409 (R-9)."""
+    try:
+        yield
+    except NotRetiredError as exc:
+        if exc.missing:
+            raise ServiceError(404, f"{what} not found") from exc
+        raise ServiceError(409, f"{what} must be retired before it is purged") from exc

src/services/users.py

@@ -15,7 +15,7 @@
 
 from ..security_fastapi import verify_credentials_db
 from ..utils.logging_config import get_logger
-from .errors import ServiceError
+from .errors import ServiceError, purge_refusals
 from .passwords import hash_user_password, verify_user_password
 from .workspaces import WorkspacesService
 
@@ -158,7 +158,8 @@ def retire_user(self, user_id: str, actor: str) -> bool:
         return self._db.users.retire(user_id, deleted_by=actor)
 
     def purge_user(self, user_id: str) -> bool:
-        return self._db.users.purge(user_id)
+        with purge_refusals("User"):
+            return self._db.users.purge(user_id)
 
     def user_workspaces(self, user_id: str) -> list[dict[str, Any]]:
         workspaces: list[dict[str, Any]] = self._db.members.for_user(user_id)

src/services/workspaces.py

@@ -14,7 +14,7 @@
 from ..db.workspaces import LastOwnerError
 from ..utils.logging_config import get_logger
 from ..utils.scope import Scope
-from .errors import ServiceError
+from .errors import ServiceError, purge_refusals
 
 if TYPE_CHECKING:
     from ..db import Database
@@ -86,7 +86,8 @@ def default_id(self) -> str | None:
         return self._db.workspaces.default_id()
 
     def purge(self, workspace_id: str) -> bool:
-        return self._db.workspaces.purge(workspace_id)
+        with purge_refusals("Workspace"):
+            return self._db.workspaces.purge(workspace_id)
 
     def ontology(self, top_n: int, scope: Scope) -> dict[str, Any]:
         return self._db.entities.workspace_ontology(top_n=top_n, scope=scope)