Delete a session's data from Context Intelligence (server side)

README.md

@@ -320,6 +320,9 @@ Full runtime onboarding/offboarding runbook and the `/admin/*` API:
 | `GET` | `/blobs/{session_id}` | List all blob URIs for a session |
 | `GET` | `/blobs/{session_id}/{key}` | Retrieve a stored blob |
 | `POST` | `/cypher` | Proxy a Cypher query to Neo4j |
+| `GET` | `/sessions/{session_id}/summary` | Report what deleting this session's data would remove, without deleting anything (needs read access) |
+| `DELETE` | `/sessions/{session_id}` | Delete a whole session graph and its stored data (needs write access) |
+| `GET` | `/whoami` | Report the caller's own identity (`{"contributor_id": "..."}`, or `null` when auth is off) -- needs read access |
 | `GET` | `/queues/dead-letter` | List dead-letter queues — `worker_key`, `item_count`, `last_error`, `last_ts` (requires `Authorization: Bearer`) |
 | `POST` | `/queues/dead-letter/{worker_key}/replay` | Re-enqueue a worker's dead-letter records then purge; returns count re-enqueued (requires `Authorization: Bearer`) |
 | `POST` | `/queues/dead-letter/{worker_key}/purge` | Permanently delete a worker's dead-letter records; returns count purged (requires `Authorization: Bearer`) |
@@ -353,6 +356,47 @@ Full runtime onboarding/offboarding runbook and the `/admin/*` API:
 
 Use `"workspace": "*"` to query across all workspaces.
 
+### Deleting a session's data
+
+A user can remove data they contributed. Deleting a session removes the **whole session graph**
+— the root session and every session below it (its subsessions and forks) — together with the
+data those sessions point to: their blobs, their queue files, and their dead-letter records.
+
+You only ever give the server a **session id**. There is no workspace to pass — the server looks
+up which workspace that session id belongs to on its own.
+
+- Passing **any** session id deletes the whole graph it belongs to. If you pass a subsession id,
+  the server first finds that graph's root and then removes the whole graph. There is no way to
+  delete a single subsession on its own.
+- Nodes that are **shared** with other sessions (for example an agent that several sessions used)
+  are kept. Only the links from those shared nodes into the deleted graph are removed.
+- Deleting is **permanent**. There is no restore.
+
+Deleting is a two-step flow so nothing is removed by surprise:
+
+1. **Preview first.** `GET /sessions/{session_id}/summary` returns what would be removed: who
+   created it, how many sessions, nodes, edges, and blobs are in the graph, when it started and
+   last changed, and whether it is ready to delete. A session that changed less than a minute ago
+   may still be receiving data. Nothing is deleted by this call.
+2. **Then delete.** `DELETE /sessions/{session_id}` performs the delete and reports the counts of
+   what was removed. There is no preview flag on this call — it always deletes. Every applied
+   delete is written to the server log.
+
+A session can be deleted once it is **no longer receiving data** — that is, everything it sent
+has finished being written and nothing is still queued for it. This is the normal state for any
+session that has finished running, so in practice a session you want to remove can be deleted.
+
+The delete is refused with `409` only while a session in the graph is **still receiving data**
+(its queue has records that have not finished being written yet). This protects a session that is
+still live or still catching up: wait until it has finished, then delete. Because this is a
+temporary state, that `409` carries a `Retry-After` header and a body with `retry_after_seconds`
+and the `pending_sessions` still draining, so a caller can back off and retry without guessing.
+A `409` is also returned in the (should-not-happen) case where the session id is found in
+more than one workspace — the server refuses to guess which workspace was meant. That one has no
+`Retry-After`: it is not retryable.
+
+The summary uses the read-only Neo4j connection; the actual delete uses the admin connection.
+
 ---
 
 ## Configuration
@@ -435,6 +479,10 @@ recovered rather than lost across a restart. The durable per-session logs — no
 just Neo4j — are the record for events that have been accepted but not yet written
 to the graph.
 
+Deleting a session (see [Deleting a session's data](#deleting-a-sessions-data))
+removes its data from all three of these places at once: its nodes and edges in the
+Neo4j graph, its blob files, and its queue and dead-letter files.
+
 For persistence on Azure Container Apps (persistent storage + Neo4j on AuraDB),
 see [docs/azure-deployment.md](docs/azure-deployment.md).
 

context_intelligence_server/blob_store.py

@@ -51,6 +51,24 @@ async def list(self, session_id: str) -> list[str]:
         """Return all blob URIs for *session_id*, sorted lexicographically."""
         ...
 
+    async def size(self, uri: str) -> int:
+        """Return the byte size of the blob addressed by *uri*.
+
+        Idempotent: a missing blob returns 0, not an error (mirrors
+        ``delete_session``'s missing-is-zero contract).
+
+        Raises:
+            ValueError: If *uri* is not a valid ``ci-blob://`` URI.
+        """
+        ...
+
+    async def delete_session(self, session_id: str) -> int:
+        """Delete all blobs for *session_id* and return the number removed.
+
+        Idempotent: a session with no stored blobs returns 0, not an error.
+        """
+        ...
+
     async def dump(self, uri: str, dest_dir: Path | str | None = None) -> str:
         """Copy the blob file addressed by *uri* to *dest_dir*.
 
@@ -205,6 +223,37 @@ def _list() -> list[str]:
 
         return await asyncio.to_thread(_list)
 
+    async def size(self, uri: str) -> int:
+        """Return the byte size of the blob addressed by *uri*, or 0 if missing."""
+        session_id, key = self._parse_uri(uri)
+        path = self._blob_path(session_id, key)
+
+        def _size() -> int:
+            try:
+                return path.stat().st_size
+            except FileNotFoundError:
+                return 0
+
+        return await asyncio.to_thread(_size)
+
+    async def delete_session(self, session_id: str) -> int:
+        """Delete all blobs for *session_id* and return the number removed.
+
+        Removes ``<root>/<session_id>/``. Idempotent: a session with no
+        stored blobs returns 0 and is not an error.
+        """
+        session_dir = self._root / session_id
+        blobs_dir = session_dir / "blobs"
+
+        def _delete() -> int:
+            if not blobs_dir.exists():
+                return 0
+            count = sum(1 for _ in blobs_dir.glob("*.json"))
+            shutil.rmtree(session_dir)
+            return count
+
+        return await asyncio.to_thread(_delete)
+
     async def dump(self, uri: str, dest_dir: Path | str | None = None) -> str:
         """Copy the blob file addressed by *uri* to *dest_dir*.
 

context_intelligence_server/config.py

@@ -824,6 +824,12 @@ def _validate_neo4j_max_connection_pool_size(cls, v: int) -> int:
     blob_path: str = "/data/blobs"
     queues_path: str = "/data/queues"
 
+    # Seconds advertised in the Retry-After header (and body) of a 409 when a
+    # delete is refused because the graph still has undrained queue records.
+    # The condition is transient -- the drain clears it -- so the caller is told
+    # how long to wait before retrying. A small poll interval; configurable.
+    delete_retry_after_seconds: int = 2
+
     # -------------------------------------------------------------------------
     # Durable ingest queue
     # -------------------------------------------------------------------------

context_intelligence_server/deletion.py

@@ -0,0 +1,233 @@
+"""DeletionService -- composes the storage primitives to delete a session graph.
+
+See docs/02-server-design.md (DELETE section, "Abstraction principle",
+"Whole-graph, not one session") for the design this implements.
+
+This service composes ONLY the public Protocol-level APIs of ``GraphStore``,
+``BlobStore``, and ``QueueManager`` -- it never talks to Neo4j or the
+filesystem directly. Each backend's delete lives at its own storage
+abstraction; this module holds orchestration, precondition enforcement,
+ordering, and logging only.
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Protocol
+
+from context_intelligence_server.blob_store import BlobStore
+from context_intelligence_server.graph_store import GraphStore
+
+logger = logging.getLogger(__name__)
+
+
+class _QueueManagerLike(Protocol):
+    """The subset of ``QueueManager`` this service composes."""
+
+    async def pending_count(self, session_id: str) -> int: ...
+
+    async def delete_session(self, session_id: str) -> bool: ...
+
+
+class SessionsPendingError(Exception):
+    """Raised by ``apply`` when the graph still has undrained queue records.
+
+    The delete is refused (nothing is written) because one or more sessions in
+    the graph still have complete-but-uncommitted queue lines left to drain.
+    This is a *transient, retryable* condition: once the drain finishes,
+    ``pending_count`` returns to 0 and the same delete succeeds. The offending
+    session ids are carried on ``pending_sessions`` so the caller can report
+    exactly what is still in flight and can be told, machine-readably, to retry.
+    """
+
+    def __init__(self, root_id: str, pending_sessions: list[str]) -> None:
+        self.root_id = root_id
+        self.pending_sessions = pending_sessions
+        super().__init__(
+            f"apply refused: graph root={root_id!r} has pending "
+            f"(uncommitted) session(s) {pending_sessions!r}; drain before "
+            "deleting -- nothing was deleted"
+        )
+
+
+@dataclass(frozen=True)
+class DeletionPreview:
+    """Dry-run facts for a whole session graph -- mutates nothing.
+
+    Exactly what the router surfaces before ``apply``. ``blob_count`` and the
+    node/edge counts are the whole-graph totals (see ``SessionGraph``), not
+    the passed session's alone. ``deletable`` is False whenever any session in
+    the graph still has pending (uncommitted) queue records; ``pending_sessions``
+    names them.
+    """
+
+    root_id: str
+    session_ids: frozenset[str]
+    node_count: int
+    edge_count: int
+    blob_count: int
+    created_by: str | None
+    started_at: datetime | None
+    last_change: datetime | None
+    subsession_count: int
+    workspace: str
+    working_dir: str | None
+    deletable: bool
+    pending_sessions: list[str]
+
+
+@dataclass(frozen=True)
+class DeletionResult:
+    """Per-backend counts for an applied whole-graph deletion."""
+
+    root_id: str
+    session_count: int
+    nodes_deleted: int
+    relationships_deleted: int
+    blobs_deleted: int
+    queue_sessions_cleaned: int
+
+
+class DeletionService:
+    """Composes ``GraphStore``, ``BlobStore``, and ``QueueManager`` to delete
+    a whole session graph -- root + all descendants -- plus every blob and
+    queue artifact it references.
+
+    Constructed via dependency injection; never constructs or reaches into
+    Neo4j/the filesystem itself.
+    """
+
+    def __init__(
+        self,
+        graph_store: GraphStore,
+        blob_store: BlobStore,
+        queue_manager: _QueueManagerLike,
+    ) -> None:
+        self._graph = graph_store
+        self._blobs = blob_store
+        self._queue = queue_manager
+
+    async def _pending_sessions(self, session_ids: frozenset[str]) -> list[str]:
+        """Return the sorted subset of *session_ids* with pending queue records."""
+        pending: list[str] = []
+        for sid in sorted(session_ids):
+            if await self._queue.pending_count(sid) > 0:
+                pending.append(sid)
+        return pending
+
+    async def _blob_count(self, session_ids: frozenset[str]) -> int:
+        """Return the total number of blobs stored for every session in *session_ids*.
+
+        The blob store is keyed by session id and already knows every blob it
+        holds for a session (``BlobStore.list``), so this asks the blob store
+        directly instead of trying to find blob markers hidden inside graph
+        node data. This also means a blob is counted even if no node happens
+        to reference it -- the blob store is the one place that knows what it
+        is actually holding.
+        """
+        total = 0
+        for sid in session_ids:
+            total += len(await self._blobs.list(sid))
+        return total
+
+    async def preview(self, session_id: str) -> DeletionPreview | None:
+        """Resolve the whole session graph for *session_id* and report what
+        deleting it would do -- mutates nothing.
+
+        Returns ``None`` if *session_id* does not resolve to any known session.
+        """
+        graph = await self._graph.resolve_session_graph(session_id)
+        if graph is None:
+            return None
+
+        pending_sessions = await self._pending_sessions(graph.session_ids)
+        blob_count = await self._blob_count(graph.session_ids)
+        return DeletionPreview(
+            root_id=graph.root_id,
+            session_ids=graph.session_ids,
+            node_count=graph.node_count,
+            edge_count=graph.edge_count,
+            blob_count=blob_count,
+            created_by=graph.created_by,
+            started_at=graph.started_at,
+            last_change=graph.last_change,
+            subsession_count=graph.subsession_count,
+            workspace=graph.workspace,
+            working_dir=graph.working_dir,
+            deletable=not pending_sessions,
+            pending_sessions=pending_sessions,
+        )
+
+    async def apply(
+        self, session_id: str, *, requested_by: str | None = None
+    ) -> DeletionResult | None:
+        """Permanently delete the whole session graph for *session_id*.
+
+        Resolves the graph, enforces the drain precondition (every session in
+        the graph must have zero pending queue records) across the WHOLE
+        graph, then deletes in order: graph -> blobs (per session) -> queue
+        artifacts (per session). ``session_ids`` is captured from the
+        resolution BEFORE any delete, so losing the graph node set first does
+        not lose track of what else must be removed.
+
+        Returns ``None`` if *session_id* does not resolve to any known
+        session -- no writes occur in that case.
+
+        Raises:
+            SessionsPendingError: If any session in the graph still has pending
+                (uncommitted) queue records -- refuses, deletes nothing. This is
+                retryable: it carries ``pending_sessions`` and clears once drained.
+            RuntimeError: If the graph vanishes between resolve and delete.
+        """
+        graph = await self._graph.resolve_session_graph(session_id)
+        if graph is None:
+            return None
+
+        session_ids = graph.session_ids
+
+        pending_sessions = await self._pending_sessions(session_ids)
+        if pending_sessions:
+            raise SessionsPendingError(graph.root_id, pending_sessions)
+
+        graph_result = await self._graph.delete_session_graph(session_id)
+        if graph_result is None:
+            raise RuntimeError(
+                f"apply: graph for root={graph.root_id!r} vanished between "
+                "resolve and delete -- no writes were attempted"
+            )
+
+        blobs_deleted = 0
+        for sid in session_ids:
+            blobs_deleted += await self._blobs.delete_session(sid)
+
+        queue_sessions_cleaned = 0
+        for sid in session_ids:
+            if await self._queue.delete_session(sid):
+                queue_sessions_cleaned += 1
+
+        result = DeletionResult(
+            root_id=graph.root_id,
+            session_count=len(session_ids),
+            nodes_deleted=graph_result.nodes_deleted,
+            relationships_deleted=graph_result.relationships_deleted,
+            blobs_deleted=blobs_deleted,
+            queue_sessions_cleaned=queue_sessions_cleaned,
+        )
+
+        logger.info(
+            "session_deletion_applied root_id=%s created_by=%s requested_by=%s "
+            "session_count=%d nodes_deleted=%d relationships_deleted=%d "
+            "blobs_deleted=%d queue_sessions_cleaned=%d",
+            result.root_id,
+            graph.created_by,
+            requested_by,
+            result.session_count,
+            result.nodes_deleted,
+            result.relationships_deleted,
+            result.blobs_deleted,
+            result.queue_sessions_cleaned,
+            extra={"session_id": result.root_id},
+        )
+        return result