diff --git a/README.md b/README.md index 35b3da64..47d23d5d 100644 --- a/README.md +++ b/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). diff --git a/context_intelligence_server/blob_store.py b/context_intelligence_server/blob_store.py index 94511781..e2ffbdfb 100644 --- a/context_intelligence_server/blob_store.py +++ b/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 ``//``. 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*. diff --git a/context_intelligence_server/config.py b/context_intelligence_server/config.py index 4e36ba09..afeb07ae 100644 --- a/context_intelligence_server/config.py +++ b/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 # ------------------------------------------------------------------------- diff --git a/context_intelligence_server/deletion.py b/context_intelligence_server/deletion.py new file mode 100644 index 00000000..14a3fd4f --- /dev/null +++ b/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 diff --git a/context_intelligence_server/graph_store.py b/context_intelligence_server/graph_store.py index aa491398..b3309032 100644 --- a/context_intelligence_server/graph_store.py +++ b/context_intelligence_server/graph_store.py @@ -43,9 +43,85 @@ from __future__ import annotations +from dataclasses import dataclass +from datetime import datetime from typing import Any, Protocol, runtime_checkable +@dataclass(frozen=True) +class SessionGraph: + """Whole-graph resolution result backing the session-summary facts. + + A session graph spans many session nodes (root + subsessions + forks), + never just the one passed in (see B1 in + docs/context-intelligence-delete-session-data.md). ``session_ids`` is + the authoritative set: a later delete operation reuses this exact + resolution so its dry-run preview and its apply step can never disagree. + + This carries the session ids and the graph facts (node/edge counts, + who started it, when it last changed) -- it does not carry the graph's + blobs. Blobs belong to the blob store, which is already keyed by + session id and can list every blob for a session directly + (``BlobStore.list``). The caller that needs blob counts or sizes (the + deletion service) asks the blob store for each session id in + ``session_ids`` and adds the results up, instead of this store trying to + find blob markers hidden inside node data. + + ``node_count``/``edge_count`` cover the graph's own subgraph only -- + traversal stops at (but includes) any ``:SST_CONCEPT`` node, since + concept nodes (Agent/Orchestrator/Recipe) are shared across sessions and + are never owned by one graph. + """ + + root_id: str + session_ids: frozenset[str] + node_count: int + edge_count: int + created_by: str | None + started_at: datetime | None + last_change: datetime | None + subsession_count: int + workspace: str + working_dir: str | None + + +class AmbiguousSessionError(Exception): + """Raised when a session id is found in more than one workspace. + + ``resolve_session_graph`` and ``delete_session_graph`` no longer take a + workspace as input -- they look up which workspace a session id belongs + to and use that. Session ids are supposed to be unique, so this should + not normally happen, but if it ever does, refusing loudly here is much + safer than picking one workspace and deleting the wrong graph. + """ + + def __init__(self, session_id: str, workspaces: list[str]) -> None: + self.session_id = session_id + self.workspaces = workspaces + super().__init__( + f"session id {session_id!r} exists in more than one workspace: " + f"{workspaces!r}" + ) + + +@dataclass(frozen=True) +class GraphDeleteResult: + """Result of a whole-graph ``delete_session_graph`` call. + + ``nodes_deleted``/``relationships_deleted`` count only the OWNED graph + subgraph that was actually removed -- i.e. the same node set + ``resolve_session_graph`` reports EXCLUDING any shared ``:SST_CONCEPT`` + node (Agent/Orchestrator/Recipe). Concept nodes are never deleted, only + detached: ``relationships_deleted`` includes any edge from a deleted + owned node into a surviving concept node, since ``DETACH DELETE`` removes + that edge as a side effect of removing the owned endpoint. + """ + + root_id: str + nodes_deleted: int + relationships_deleted: int + + @runtime_checkable class GraphStore(Protocol): """Protocol for a workspace-scoped, buffered graph store. @@ -111,6 +187,62 @@ async def find_delegation_by_sub_session( """ ... + async def resolve_session_graph(self, session_id: str) -> SessionGraph | None: + """Resolve the whole session graph (root + all descendants) for *session_id*. + + *session_id* is the only input needed -- there is no separate + workspace argument. The workspace a session lives in is looked up + from *session_id* itself (session ids are unique), and that + discovered workspace is what the rest of the resolution is scoped + to, so a session graph is still resolved within one workspace. + + If *session_id* is not itself a root, first expands UP to the root by + walking ``HAS_SUBSESSION``/``FORKED`` edges, then returns the root plus + every descendant session -- never just the passed session alone (B1). + A sub-session id and its root id MUST resolve to the identical graph. + + Returns ``None`` if *session_id* does not resolve to any known + ``:Session`` node. + + Raises: + AmbiguousSessionError: If *session_id* is found in more than one + workspace. Session ids are supposed to be unique, so this + should not happen in practice, but this method refuses to + guess which workspace was meant rather than silently picking + one. + """ + ... + + async def delete_session_graph(self, session_id: str) -> GraphDeleteResult | None: + """Permanently delete the whole OWNED session graph for *session_id*. + + *session_id* is the only input needed -- there is no separate + workspace argument. Like ``resolve_session_graph``, the workspace is + looked up from *session_id* itself, not supplied by the caller. + + Resolves the graph via the identical traversal ``resolve_session_graph`` + uses (root + all descendants, bounded at but not past any + ``:SST_CONCEPT`` node), then removes every node it owns -- the graph + nodes EXCLUDING shared ``:SST_CONCEPT`` nodes. Shared concept nodes + (Agent/Orchestrator/Recipe) are NEVER deleted, only detached: only + their edges into the deleted graph are removed, as a side effect of + removing the owned endpoint of each such edge. + + A sub-session id and its root id MUST resolve to, and delete, the + identical graph -- this reuses ``resolve_session_graph``'s exact + resolution so a delete can never diverge from what a prior preview + summary promised. + + Returns ``None`` if *session_id* does not resolve to any known + ``:Session`` node -- no writes occur in that case. + + Raises: + AmbiguousSessionError: If *session_id* is found in more than one + workspace (see ``resolve_session_graph``). Nothing is deleted + in that case. + """ + ... + async def flush(self) -> None: """Persist all buffered writes to the backing store. diff --git a/context_intelligence_server/main.py b/context_intelligence_server/main.py index 6ef89655..79dce3de 100644 --- a/context_intelligence_server/main.py +++ b/context_intelligence_server/main.py @@ -47,8 +47,10 @@ ) from context_intelligence_server.registry import SessionRegistry from context_intelligence_server.routers.admin import router as admin_router +from context_intelligence_server.routers.deletion import router as deletion_router from context_intelligence_server.routers.queues import router as queues_router from context_intelligence_server.routers.version import router as version_router +from context_intelligence_server.routers.whoami import router as whoami_router from context_intelligence_server.status import build_status_response _settings = get_settings() @@ -454,6 +456,8 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: app.include_router(admin_router) app.include_router(version_router) app.include_router(queues_router) +app.include_router(deletion_router) +app.include_router(whoami_router) _start_time = time.time() registry = SessionRegistry() # Expose the registry singleton on app.state so routers can read it via diff --git a/context_intelligence_server/neo4j_store.py b/context_intelligence_server/neo4j_store.py index fbfd1599..a1f0668c 100644 --- a/context_intelligence_server/neo4j_store.py +++ b/context_intelligence_server/neo4j_store.py @@ -23,6 +23,11 @@ from neo4j.exceptions import DriverError, Neo4jError from context_intelligence_server.config import Neo4jClientConfig +from context_intelligence_server.graph_store import ( + AmbiguousSessionError, + GraphDeleteResult, + SessionGraph, +) _LOG = logging.getLogger(__name__) @@ -228,6 +233,141 @@ def _edge_merge_cypher(edge_type: str) -> str: # size, so this constant is interpolated (never user-supplied). _NODE_BACKFILL_BATCH = 10_000 +# --------------------------------------------------------------------------- +# Entry-session workspace discovery (resolve_session_graph/delete_session_graph) +# --------------------------------------------------------------------------- +# Neither method takes a workspace as input any more. Instead, this query +# finds every ":Session" node with the given node_id -- no workspace filter +# at all -- and returns the distinct set of workspace values found on those +# nodes. Session ids are supposed to be unique, so in the normal case this +# returns exactly one row (one workspace). Zero rows means the session id is +# unknown anywhere. More than one distinct workspace means the same session +# id exists in more than one workspace -- the caller must refuse and raise +# rather than silently pick one and risk resolving (or deleting) the wrong +# graph. +_ENTRY_SESSION_WORKSPACE_CYPHER = ( + "MATCH (start:Session {node_id: $session_id}) " + "RETURN DISTINCT start.workspace AS workspace" +) + +# --------------------------------------------------------------------------- +# Session-graph resolution (resolve_session_graph) +# --------------------------------------------------------------------------- +# Graph membership is defined purely via HAS_SUBSESSION/FORKED edges, which +# _handle_start/_handle_fork (session.py) create exactly once per session as +# it is classified -- each session has at most one such incoming edge, so the +# graph is a tree (no cycles, unique parent). +# +# $workspace here is the workspace ALREADY DISCOVERED by +# _ENTRY_SESSION_WORKSPACE_CYPHER above, not something the caller supplied -- +# by the time this query runs, the caller has already confirmed $session_id +# lives in exactly this one workspace, so scoping every step of the walk by +# it keeps the whole resolution inside that one workspace. +# +# Step 1: expand UP from $session_id via incoming HAS_SUBSESSION/FORKED edges. +# OPTIONAL MATCH returns one row per ancestor found (or one null row if +# $session_id is already the root); the ancestor reached by the LONGEST +# upward path is the root, by construction of the tree (no bound needed -- +# the walk always terminates because the graph is acyclic). +# Step 2: expand DOWN from that root via the same edge types, `*0..` so the +# root itself is included, to enumerate every graph member. +_GRAPH_RESOLVE_CYPHER = ( + "MATCH (start:Session {node_id: $session_id, workspace: $workspace}) " + "OPTIONAL MATCH p = (start)<-[:HAS_SUBSESSION|FORKED*]-(ancestor:Session {workspace: $workspace}) " + "WITH start, ancestor, p " + "ORDER BY length(p) DESC " + "LIMIT 1 " + "WITH coalesce(ancestor, start) AS root " + "MATCH (root)-[:HAS_SUBSESSION|FORKED*0..]->(member:Session {workspace: $workspace}) " + "RETURN DISTINCT member.node_id AS session_id, root.node_id AS root_id, " + "properties(member) AS props" +) + +# Graph subgraph: from every graph session node, expand outward along ANY +# relationship type, `*0..` so the session nodes themselves are included. +# `ALL(x IN nodes(path)[0..-1] WHERE NOT x:SST_CONCEPT)` stops expansion AT +# (but includes) a :SST_CONCEPT node (Agent/Orchestrator/Recipe) -- those are +# shared across sessions and are never owned by one graph, so a path that +# would continue past one is excluded entirely (any such path also exists at +# the shorter length ending exactly at the concept node, which IS kept). +_GRAPH_SUBGRAPH_CYPHER = ( + "UNWIND $session_ids AS sid " + "MATCH (s:Session {node_id: sid, workspace: $workspace}) " + "WITH collect(s) AS seeds " + "CALL { " + "WITH seeds " + "UNWIND seeds AS s " + "MATCH path = (s)-[*0..]->(n {workspace: $workspace}) " + "WHERE ALL(x IN nodes(path)[0..-1] WHERE NOT x:SST_CONCEPT) " + "RETURN collect(DISTINCT n) AS graph_nodes " + "} " + "WITH graph_nodes " + "UNWIND graph_nodes AS a " + "OPTIONAL MATCH (a)-[r]->(b) " + "WHERE b IN graph_nodes " + "WITH graph_nodes, collect(DISTINCT r) AS rels " + "RETURN size(graph_nodes) AS node_count, size(rels) AS edge_count" +) + +# --------------------------------------------------------------------------- +# Whole-graph delete (delete_session_graph) +# --------------------------------------------------------------------------- +# Same seed set (graph session_ids) and boundary rule (stop AT but not past +# a :SST_CONCEPT node) as _GRAPH_SUBGRAPH_CYPHER above, so delete can never +# diverge from what resolve_session_graph already resolved. Unlike the +# subgraph query, this one partitions the traversal result into OWNED nodes +# (deleted) vs boundary :SST_CONCEPT nodes (kept, verified to survive) and +# returns elementId -- the identity DETACH DELETE by elementId needs. +_GRAPH_NODE_PARTITION_CYPHER = ( + "UNWIND $session_ids AS sid " + "MATCH (s:Session {node_id: sid, workspace: $workspace}) " + "WITH collect(s) AS seeds " + "CALL { " + "WITH seeds " + "UNWIND seeds AS s " + "MATCH path = (s)-[*0..]->(n {workspace: $workspace}) " + "WHERE ALL(x IN nodes(path)[0..-1] WHERE NOT x:SST_CONCEPT) " + "RETURN collect(DISTINCT n) AS graph_nodes " + "} " + "UNWIND graph_nodes AS n " + "RETURN elementId(n) AS eid, n:SST_CONCEPT AS is_concept" +) + +# Distinct count of relationships incident (either direction) to any of the +# owned nodes named in $element_ids -- exactly the set DETACH DELETE removes. +# Must run BEFORE the delete batches below (the relationships are gone after). +_GRAPH_REL_COUNT_CYPHER = ( + "UNWIND $element_ids AS eid " + "MATCH (n) WHERE elementId(n) = eid " + "OPTIONAL MATCH (n)-[r]-() " + "RETURN count(DISTINCT r) AS rel_count" +) + +# 2-phase delete primitive: enumerate elementIds (above), then DETACH DELETE +# by elementId in batches -- the proven pattern from ci_session_purge, which +# deleted a 2601-node graph this way. DETACH DELETE on an owned node removes +# every relationship touching it, including any edge into a surviving +# :SST_CONCEPT node -- that edge's removal IS the "detach" the design calls for. +_GRAPH_DELETE_BATCH_CYPHER = ( + "UNWIND $element_ids AS eid " + "MATCH (n) WHERE elementId(n) = eid " + "DETACH DELETE n" +) + +# elementId-list existence count -- shared by both post-delete gate checks +# (owned nodes must be gone, concept nodes must survive). +_COUNT_NODES_BY_ELEMENT_ID_CYPHER = ( + "UNWIND $element_ids AS eid " + "MATCH (n) WHERE elementId(n) = eid " + "RETURN count(n) AS c" +) + +# Row cap per DETACH DELETE batch. elementId strings are tiny, so only the row +# bound matters in practice; the byte bound is kept generous as a defensive +# ceiling, mirroring the dual-bound _chunk_list contract used by the flush path. +_DELETE_BATCH_ROWS = 500 +_DELETE_BATCH_BYTES = 10_000_000 + def _validate_identifier(name: str, kind: str) -> None: """Raise ``ValueError`` if *name* is not a safe Neo4j label / relationship-type identifier. @@ -1480,6 +1620,239 @@ async def find_delegation_by_sub_session( return None + async def _discover_session_workspace(self, session_id: str) -> str | None: + """Find the one workspace *session_id* lives in, with no workspace given. + + Looks up every ``:Session`` node with this node_id -- across every + workspace, not just this store's own -- via + ``_ENTRY_SESSION_WORKSPACE_CYPHER``, and returns the single workspace + value found on those nodes. + + Returns ``None`` if *session_id* does not match any known ``:Session`` + node anywhere. + + Raises: + AmbiguousSessionError: If *session_id* matches ``:Session`` nodes + in more than one distinct workspace. This should not happen + in practice (session ids are unique), but if it ever does, + this refuses to guess rather than silently picking one + workspace and resolving (or deleting) the wrong graph. + """ + try: + lookup_result = await self._driver.execute_query( + _ENTRY_SESSION_WORKSPACE_CYPHER, + {"session_id": session_id}, + database_=self._database, + ) + except Neo4jError: + return None + + workspaces = {row["workspace"] for row in lookup_result.records} + if not workspaces: + return None + if len(workspaces) > 1: + raise AmbiguousSessionError(session_id, sorted(workspaces)) + return next(iter(workspaces)) + + async def resolve_session_graph(self, session_id: str) -> SessionGraph | None: + """Resolve the whole session graph for *session_id* against Neo4j. + + *session_id* is the only input -- there is no workspace argument. + The workspace is looked up from *session_id* itself (see + ``_discover_session_workspace``), then that discovered workspace + scopes the rest of the resolution, so the resolved graph still comes + from exactly one workspace. + + See ``_GRAPH_RESOLVE_CYPHER``/``_GRAPH_SUBGRAPH_CYPHER`` above for + the exact traversal. Reads only the flushed/persisted graph (no + buffer-first fallback) -- summary/delete operate on already-ingested + sessions. + + Returns ``None`` if *session_id* does not resolve to any known + ``:Session`` node in any workspace. + + Raises: + AmbiguousSessionError: If *session_id* is found in more than one + workspace. See ``_discover_session_workspace``. + """ + workspace = await self._discover_session_workspace(session_id) + if workspace is None: + return None + try: + resolve_result = await self._driver.execute_query( + _GRAPH_RESOLVE_CYPHER, + {"session_id": session_id, "workspace": workspace}, + database_=self._database, + ) + except Neo4jError: + return None + + rows = resolve_result.records + if not rows: + return None + + root_id: str = rows[0]["root_id"] + session_ids: set[str] = set() + member_props: dict[str, dict[str, Any]] = {} + for row in rows: + sid: str = row["session_id"] + session_ids.add(sid) + member_props[sid] = { + k: _normalize_temporal(v) for k, v in dict(row["props"]).items() + } + + node_count = 0 + edge_count = 0 + try: + subgraph_result = await self._driver.execute_query( + _GRAPH_SUBGRAPH_CYPHER, + {"session_ids": sorted(session_ids), "workspace": workspace}, + database_=self._database, + ) + except Neo4jError: + subgraph_result = None + if subgraph_result is not None and subgraph_result.records: + subgraph_row = subgraph_result.records[0] + node_count = subgraph_row["node_count"] + edge_count = subgraph_row["edge_count"] + + root_props = member_props.get(root_id, {}) + created_by = root_props.get("created_by") + started_at = root_props.get("started_at") + working_dir = root_props.get("working_dir") + + last_change: datetime | None = None + for props in member_props.values(): + candidate = ( + props.get("last_updated") + or props.get("ended_at") + or props.get("started_at") + ) + if isinstance(candidate, datetime) and ( + last_change is None or candidate > last_change + ): + last_change = candidate + + return SessionGraph( + root_id=root_id, + session_ids=frozenset(session_ids), + node_count=node_count, + edge_count=edge_count, + created_by=created_by if isinstance(created_by, str) else None, + started_at=started_at if isinstance(started_at, datetime) else None, + last_change=last_change, + subsession_count=len(session_ids) - 1, + workspace=workspace, + working_dir=working_dir if isinstance(working_dir, str) else None, + ) + + async def delete_session_graph(self, session_id: str) -> GraphDeleteResult | None: + """DETACH DELETE the whole OWNED session graph for *session_id* in Neo4j. + + *session_id* is the only input -- there is no workspace argument. + Reuses ``resolve_session_graph`` to find the graph (same seeds, same + boundary rule, same discovered-workspace lookup -- see that method's + docstring), then: + + 1. Partitions the graph's reachable nodes into OWNED (deleted) vs + boundary ``:SST_CONCEPT`` (kept) via ``_GRAPH_NODE_PARTITION_CYPHER``, + scoped by the workspace ``resolve_session_graph`` already + discovered (``graph.workspace``) -- never this store's own bound + workspace, which the delete no longer depends on. + 2. Counts the relationships the delete will remove (incident to any + owned node, either direction) BEFORE deleting -- they no longer + exist to count afterwards. + 3. ``DETACH DELETE``s the owned nodes by elementId, in batches + (``_DELETE_BATCH_ROWS`` per query) -- the proven 2-phase pattern + from ``ci_session_purge``, which scales to large families. + 4. Verifies the gate this feature promises: every owned node is + actually gone, and every boundary concept node actually survives. + A violation raises ``RuntimeError`` rather than returning a result + that misrepresents what happened. + + Returns ``None`` if *session_id* does not resolve to any known + ``:Session`` node -- no writes occur in that case. Once a graph is + resolved, failures during the delete/verify steps are NOT swallowed + (unlike the lenient read-path queries above): a destructive operation + that cannot prove its own result must fail loud, not report zero. + + Raises: + AmbiguousSessionError: If *session_id* is found in more than one + workspace (raised by ``resolve_session_graph`` before any + write is attempted -- nothing is deleted in that case). + """ + graph = await self.resolve_session_graph(session_id) + if graph is None: + return None + + workspace = graph.workspace + partition_result = await self._driver.execute_query( + _GRAPH_NODE_PARTITION_CYPHER, + {"session_ids": sorted(graph.session_ids), "workspace": workspace}, + database_=self._database, + ) + owned_ids: list[str] = [] + concept_ids: list[str] = [] + for row in partition_result.records: + (concept_ids if row["is_concept"] else owned_ids).append(row["eid"]) + + if not owned_ids: + # Every graph session node is itself owned, so this should be + # unreachable in practice -- defensive, not a real code path. + return GraphDeleteResult( + root_id=graph.root_id, nodes_deleted=0, relationships_deleted=0 + ) + + rel_result = await self._driver.execute_query( + _GRAPH_REL_COUNT_CYPHER, + {"element_ids": owned_ids}, + database_=self._database, + ) + relationships_deleted = ( + rel_result.records[0]["rel_count"] if rel_result.records else 0 + ) + + for batch in _chunk_list(owned_ids, _DELETE_BATCH_ROWS, _DELETE_BATCH_BYTES): + await self._driver.execute_query( + _GRAPH_DELETE_BATCH_CYPHER, + {"element_ids": batch}, + database_=self._database, + ) + + # Gate: every owned node must actually be gone. + survivor_result = await self._driver.execute_query( + _COUNT_NODES_BY_ELEMENT_ID_CYPHER, + {"element_ids": owned_ids}, + database_=self._database, + ) + survivor_count = survivor_result.records[0]["c"] if survivor_result.records else 0 + if survivor_count: + raise RuntimeError( + f"delete_session_graph: {survivor_count} owned node(s) survived " + f"DETACH DELETE for graph root {graph.root_id!r}" + ) + + # Gate: every boundary concept node must still exist. + if concept_ids: + concept_result = await self._driver.execute_query( + _COUNT_NODES_BY_ELEMENT_ID_CYPHER, + {"element_ids": concept_ids}, + database_=self._database, + ) + concept_count = concept_result.records[0]["c"] if concept_result.records else 0 + if concept_count != len(concept_ids): + raise RuntimeError( + f"delete_session_graph: expected {len(concept_ids)} shared " + f"concept node(s) to survive for graph root " + f"{graph.root_id!r}, found {concept_count}" + ) + + return GraphDeleteResult( + root_id=graph.root_id, + nodes_deleted=len(owned_ids), + relationships_deleted=relationships_deleted, + ) + async def get_edge(self, src_id: str, dst_id: str) -> dict[str, Any] | None: """Return edge data, checking the in-memory buffer first. diff --git a/context_intelligence_server/queue_manager.py b/context_intelligence_server/queue_manager.py index a52eca18..298051b4 100644 --- a/context_intelligence_server/queue_manager.py +++ b/context_intelligence_server/queue_manager.py @@ -500,6 +500,57 @@ def _delete(guard: _KeyGuard) -> bool: del self._guards[session_id] return ok + async def delete_session(self, session_id: str) -> bool: + """Permanently remove ALL queue artifacts for ``session_id``. + + Unlike ``delete_drained`` (which deliberately keeps ``.dead.jsonl``), + this removes ``.log``, ``.offset``, AND ``.dead.jsonl`` -- a + session-data delete must not leave dead letters behind. Returns True + if anything was removed; idempotent (a session with no files at all + returns False). + + Refuses (raises ``RuntimeError``, deletes nothing) if the session + still has pending (uncommitted) records. The pending check is + computed under ``guard.file_lock`` -- atomically with the delete + itself, using the same ``_read_committed_offset`` / + ``_complete_data_end`` / ``_count_newlines`` helpers ``recover()`` + and ``pending_count`` use -- so it can never race a concurrent + append and can never disagree with the drainer's own definition of + "pending". + """ + self._validate_session_id(session_id) + log = self._log_path(session_id) + offset = self._offset_path(session_id) + dead = self._dead_path(session_id) + + def _delete(guard: _KeyGuard) -> bool: + with guard.file_lock: + committed = self._read_committed_offset(session_id) + complete_end = self._complete_data_end(session_id) + pending = self._count_newlines(session_id, committed, complete_end) + if pending: + raise RuntimeError( + f"delete_session refused: session={session_id!r} has " + f"{pending} pending (uncommitted) record(s); drain " + "before deleting" + ) + removed = False + for path in (log, offset, dead): + try: + path.unlink() + removed = True + except FileNotFoundError: + pass + return removed + + with self._guard(session_id) as guard: + async with guard.admission: + removed = await _await_uninterrupted(asyncio.to_thread(_delete, guard)) + # Same three-part removal condition as delete_drained. + if guard.waiters == 1 and self._guards.get(session_id) is guard: + del self._guards[session_id] + return removed + async def read_dead_letters(self, session_id: str) -> list[dict]: """Return all dead-letter records for ``session_id`` in append order. @@ -600,6 +651,28 @@ def _scan() -> list[str]: return await asyncio.to_thread(_scan) + async def pending_count(self, session_id: str) -> int: + """Count of complete, uncommitted lines pending in a session's ``.log``. + + Authoritative "how much is left to drain" query -- reuses the exact + ``_read_committed_offset`` / ``_complete_data_end`` / ``_count_newlines`` + helpers ``recover()`` and ``recovery_seed_counts`` already use to + decide the same thing, so it can never disagree with the drainer's + own definition of "pending". 0 means drained: nothing left to + commit. A torn trailing fragment (bytes after the final newline) + never counts as pending, matching ``recover()``. This is the query + the delete-session precondition ("drained, nothing pending") must + call before deleting. + """ + self._validate_session_id(session_id) + + def _count() -> int: + committed = self._read_committed_offset(session_id) + complete_end = self._complete_data_end(session_id) + return self._count_newlines(session_id, committed, complete_end) + + return await asyncio.to_thread(_count) + def _count_dead(self, worker_key: str) -> int: """Count complete (newline-terminated) dead-letter lines for a key. diff --git a/context_intelligence_server/routers/deletion.py b/context_intelligence_server/routers/deletion.py new file mode 100644 index 00000000..76cf4def --- /dev/null +++ b/context_intelligence_server/routers/deletion.py @@ -0,0 +1,207 @@ +"""Routes for deleting a session's stored data. + +These routes are a thin layer over ``DeletionService``. Each route reads the +request, builds the graph store, blob store, and queue manager, creates a +``DeletionService`` with them, calls it, and turns the answer into JSON. The +routes do not talk to Neo4j or the file system on their own -- +``DeletionService`` already knows how to do that. + +There are two routes, split by HTTP method instead of a query flag: + +- ``GET /sessions/{session_id}/summary`` -- read-only. Reports what deleting + this session's data would do, without deleting anything. This is the + preview step. +- ``DELETE /sessions/{session_id}`` -- always performs the delete. There is + no dry-run flag here any more: the GET route above is the preview, and + this route is the one that actually removes the data. + +Neither route takes a ``workspace`` query parameter. A session id is the +unique identifier the caller passes; the server looks up which workspace +that session lives in on its own (see ``Neo4jGraphStore.resolve_session_graph``). +""" + +from __future__ import annotations + +import logging +from datetime import datetime +from typing import Any + +from fastapi import APIRouter, Depends, HTTPException, Request + +from context_intelligence_server.authz import require_read, require_write +from context_intelligence_server.blob_store import AsyncDiskBlobStore +from context_intelligence_server.config import get_settings +from context_intelligence_server.deletion import ( + DeletionPreview, + DeletionResult, + DeletionService, + SessionsPendingError, +) +from context_intelligence_server.graph_store import AmbiguousSessionError +from context_intelligence_server.neo4j_store import Neo4jGraphStore + +logger = logging.getLogger(__name__) + +router = APIRouter() + +# Plain-language detail shown to the caller when a session id is found in +# more than one workspace. This should not happen in practice (session ids +# are unique) -- see AmbiguousSessionError for why the server refuses to +# guess rather than picking one workspace. +_AMBIGUOUS_SESSION_DETAIL = ( + "the session id exists in more than one workspace; this should not " + "happen and needs to be looked into before it can be resolved" +) + + +def _iso(value: datetime | None) -> str | None: + """Turn a datetime into a plain ISO-8601 string, or leave ``None`` as ``None``.""" + return value.isoformat() if value is not None else None + + +def _preview_to_dict(preview: DeletionPreview) -> dict[str, Any]: + """Turn a DeletionPreview into a plain dict, ready to send as JSON.""" + return { + "root_id": preview.root_id, + "session_ids": sorted(preview.session_ids), + "node_count": preview.node_count, + "edge_count": preview.edge_count, + "blob_count": preview.blob_count, + "created_by": preview.created_by, + "started_at": _iso(preview.started_at), + "last_change": _iso(preview.last_change), + "subsession_count": preview.subsession_count, + "workspace": preview.workspace, + "working_dir": preview.working_dir, + "deletable": preview.deletable, + "pending_sessions": preview.pending_sessions, + } + + +def _result_to_dict(result: DeletionResult) -> dict[str, Any]: + """Turn a DeletionResult into a plain dict, ready to send as JSON.""" + return { + "root_id": result.root_id, + "session_count": result.session_count, + "nodes_deleted": result.nodes_deleted, + "relationships_deleted": result.relationships_deleted, + "blobs_deleted": result.blobs_deleted, + "queue_sessions_cleaned": result.queue_sessions_cleaned, + } + + +def _build_service(graph_store: Neo4jGraphStore, request: Request) -> DeletionService: + """Assemble a DeletionService from one graph store plus the blob store + and queue manager every route uses the same way.""" + settings = get_settings() + blob_store = AsyncDiskBlobStore(root=settings.blob_path) + queue_manager = request.app.state.registry.queue_manager + return DeletionService(graph_store, blob_store, queue_manager) + + +async def read_deletion_service(request: Request) -> DeletionService: + """Build a DeletionService that only reads, through the read-only Neo4j + connection (``app.state.neo4j_query_driver``). + + Used by the summary route. No workspace is supplied here -- the graph + store looks up which workspace a session id belongs to on its own. + """ + graph_store = Neo4jGraphStore( + uri="", + driver=request.app.state.neo4j_query_driver, + ) + return _build_service(graph_store, request) + + +async def delete_route_service(request: Request) -> DeletionService: + """Build the DeletionService the delete route uses. + + The delete route always changes stored data, so it always uses the + admin Neo4j connection (``app.state.neo4j_driver``) -- unlike the + summary route, there is no read-only path here any more, because there + is no more dry run on this route. No workspace is supplied -- the graph + store looks up which workspace a session id belongs to on its own. + """ + graph_store = Neo4jGraphStore(uri="", driver=request.app.state.neo4j_driver) + return _build_service(graph_store, request) + + +def _caller_id(request: Request) -> str | None: + """Return the authenticated caller's id, or None when auth is off. + + The auth middleware stores this under ``contributor_id`` in the request's + scope state (see ``authz.py`` and ``routers/admin.py`` for the same + read). + """ + state: dict = request.scope.get("state", {}) + return state.get("contributor_id") + + +@router.get( + "/sessions/{session_id}/summary", + dependencies=[Depends(require_read)], +) +async def get_session_summary( + session_id: str, + service: DeletionService = Depends(read_deletion_service), +) -> dict[str, Any]: + """Report what deleting this session's data would do. Deletes nothing. + + Returns 404 when ``session_id`` does not match any known session, and + 409 when ``session_id`` is somehow found in more than one workspace + (see ``AmbiguousSessionError`` -- this should not happen in practice). + """ + try: + preview = await service.preview(session_id) + except AmbiguousSessionError as exc: + raise HTTPException(status_code=409, detail=_AMBIGUOUS_SESSION_DETAIL) from exc + if preview is None: + raise HTTPException(status_code=404, detail=f"session {session_id!r} not found") + return _preview_to_dict(preview) + + +@router.delete( + "/sessions/{session_id}", + dependencies=[Depends(require_write)], +) +async def delete_session( + session_id: str, + request: Request, + service: DeletionService = Depends(delete_route_service), +) -> dict[str, Any]: + """Delete a session's data. This always deletes -- to preview what would + be removed first, without deleting anything, call + ``GET /sessions/{session_id}/summary``. + + Returns 404 when ``session_id`` does not match any known session, and 409 + in two distinct, machine-distinguishable cases: + + - The session (or a related session in the same graph) still has undrained + queue records -- ``reason: "sessions_pending"``. This is transient and + **retryable**: the response carries a ``Retry-After`` header and a body + ``retry_after_seconds`` + ``pending_sessions`` so the caller can back off + and retry once the drain finishes. + - The session id is somehow found in more than one workspace + (``AmbiguousSessionError``) -- not retryable; no ``Retry-After``. + """ + try: + result = await service.apply(session_id, requested_by=_caller_id(request)) + except SessionsPendingError as exc: + retry_after = get_settings().delete_retry_after_seconds + raise HTTPException( + status_code=409, + detail={ + "reason": "sessions_pending", + "message": str(exc), + "pending_sessions": exc.pending_sessions, + "retry_after_seconds": retry_after, + }, + headers={"Retry-After": str(retry_after)}, + ) from exc + except AmbiguousSessionError as exc: + raise HTTPException(status_code=409, detail=_AMBIGUOUS_SESSION_DETAIL) from exc + except RuntimeError as exc: + raise HTTPException(status_code=409, detail=str(exc)) from exc + if result is None: + raise HTTPException(status_code=404, detail=f"session {session_id!r} not found") + return _result_to_dict(result) diff --git a/context_intelligence_server/routers/whoami.py b/context_intelligence_server/routers/whoami.py new file mode 100644 index 00000000..9dc3aa15 --- /dev/null +++ b/context_intelligence_server/routers/whoami.py @@ -0,0 +1,58 @@ +"""Route for the authenticated caller's own identity. + +``GET /whoami`` answers "who am I" for the calling credential. It exists so a +client -- in particular the deletion bundle's agent -- can compare its own +identity against a session's ``created_by`` before acting on it (for example, +warning before deleting a session someone else created), without +re-implementing the server's auth extraction itself. + +This route does not add a new identity source. It reads the exact same +``contributor_id`` the auth middleware (``auth.py``) already writes to +``request.scope["state"]`` for every authenticated request -- the same value +``routers/deletion.py``'s ``_caller_id`` reads to stamp a delete's +``requested_by``, and ``routers/admin.py``'s ``_admin_who`` reads for audit +logging. +""" + +from __future__ import annotations + +from typing import Any + +from fastapi import APIRouter, Depends, Request + +from context_intelligence_server.authz import require_read + +router = APIRouter() + + +def _caller_id(request: Request) -> str | None: + """Return the authenticated caller's id, or None when auth is off. + + Mirrors ``routers/deletion.py``'s ``_caller_id`` and ``routers/admin.py``'s + ``_admin_who`` -- both read the identical ``contributor_id`` key the auth + middleware (``auth.py``) stores in the request's scope state. Kept as a + small local copy rather than a cross-router import, matching how those two + routers each already keep their own copy of this one-line read. + """ + state: dict = request.scope.get("state", {}) + return state.get("contributor_id") + + +@router.get( + "/whoami", + dependencies=[Depends(require_read)], +) +async def get_whoami(request: Request) -> dict[str, Any]: + """Report the authenticated caller's identity. + + Returns the same ``contributor_id`` value the server stamps onto + ``created_by`` (on session data) and ``requested_by`` (on a delete) -- + this is what lets a client check "am I the one who created this session" + before acting on it. + + When auth is disabled (``allow_unauthenticated=True``, no credential + required) there is no caller identity to report, so ``contributor_id`` is + ``null`` rather than a 500 -- the shape of the response never changes, + only the value. + """ + return {"contributor_id": _caller_id(request)} diff --git a/context_intelligence_server/services.py b/context_intelligence_server/services.py index 6c344b9a..3617abfa 100644 --- a/context_intelligence_server/services.py +++ b/context_intelligence_server/services.py @@ -9,12 +9,35 @@ import fnmatch import logging +from collections.abc import Iterable from datetime import datetime from typing import Any +from context_intelligence_server.blob_store import BlobStore +from context_intelligence_server.graph_store import GraphDeleteResult, SessionGraph from context_intelligence_server.handlers.data_layer_2.state import DataLayer2State from context_intelligence_server.handlers.data_layer_3.state import DataLayer3State +_GRAPH_EDGE_TYPES = frozenset({"HAS_SUBSESSION", "FORKED"}) + + +def _parse_timestamp(value: Any) -> datetime | None: + """Parse an ISO-8601 timestamp string to ``datetime``; passes datetimes through. + + GraphState keeps whatever was written (usually a str); unlike + Neo4jGraphStore there is no driver-side temporal normalisation, so this is + the in-memory equivalent of that read-path conversion. + """ + if isinstance(value, datetime): + return value + if isinstance(value, str) and value: + try: + return datetime.fromisoformat(value) + except ValueError: + return None + return None + + logger = logging.getLogger(__name__) @@ -172,6 +195,186 @@ async def find_delegation_by_sub_session( return dict(data) return None + async def resolve_session_graph(self, session_id: str) -> SessionGraph | None: + """In-memory equivalent of ``Neo4jGraphStore.resolve_session_graph``. + + *session_id* is the only input -- there is no workspace argument. + The workspace reported on the returned ``SessionGraph`` is read off + the session node itself (``node["workspace"]``), the same "discover + it, do not require it" rule ``Neo4jGraphStore`` follows, falling + back to this store's own bound workspace when a node was written + without an explicit ``workspace`` property. + + ``GraphState`` holds the nodes for exactly one workspace per + instance (``self._nodes`` is one flat dict, keyed by node_id alone), + so a node_id can never be found under two different workspaces + here -- the ambiguity ``Neo4jGraphStore`` must guard against + (``AmbiguousSessionError``) cannot structurally occur in this + in-memory store. + + Walks ``HAS_SUBSESSION``/``FORKED`` edges up to the root, then back + down to every descendant. See that method's docstring for the + graph-subgraph (node/edge) traversal rule. + """ + start = self._nodes.get(session_id) + if start is None or "Session" not in start.get("labels", []): + return None + workspace = start.get("workspace") or self._workspace + + parent_of: dict[str, str] = {} + children: dict[str, list[str]] = {} + outgoing: dict[str, list[str]] = {} + for (src, dst), edata in self._edges.items(): + outgoing.setdefault(src, []).append(dst) + if edata.get("type") in _GRAPH_EDGE_TYPES: + parent_of[dst] = src + children.setdefault(src, []).append(dst) + + # Walk up to the root (tree structure: at most one parent per node). + root_id = session_id + visited_up = {root_id} + while root_id in parent_of: + root_id = parent_of[root_id] + if root_id in visited_up: + break # defensive cycle guard; graph is acyclic by construction + visited_up.add(root_id) + + # Walk down from the root to every descendant. + session_ids: set[str] = set() + stack = [root_id] + while stack: + nid = stack.pop() + if nid in session_ids: + continue + session_ids.add(nid) + stack.extend(children.get(nid, [])) + + # Graph subgraph: expand outward from every graph session, stopping + # at (but including) any :SST_CONCEPT node. + graph_nodes: set[str] = set() + stack = list(session_ids) + while stack: + nid = stack.pop() + if nid in graph_nodes: + continue + graph_nodes.add(nid) + node_data = self._nodes.get(nid) or {} + if "SST_CONCEPT" in node_data.get("labels", []): + continue + stack.extend(outgoing.get(nid, [])) + + edge_count = sum( + 1 + for (src, dst) in self._edges + if src in graph_nodes and dst in graph_nodes + ) + + root_props = self._nodes.get(root_id) or {} + # GraphState has no per-node created_by stamp (unlike Neo4jGraphStore's + # `ON CREATE SET n.created_by`) -- fall back to the store-level value. + created_by = root_props.get("created_by") or self._created_by + started_at = _parse_timestamp(root_props.get("started_at")) + working_dir = root_props.get("working_dir") + + last_change: datetime | None = None + for sid in session_ids: + props = self._nodes.get(sid) or {} + candidate = _parse_timestamp( + props.get("last_updated") + or props.get("ended_at") + or props.get("started_at") + ) + if candidate is not None and ( + last_change is None or candidate > last_change + ): + last_change = candidate + + return SessionGraph( + root_id=root_id, + session_ids=frozenset(session_ids), + node_count=len(graph_nodes), + edge_count=edge_count, + created_by=created_by, + started_at=started_at, + last_change=last_change, + subsession_count=len(session_ids) - 1, + workspace=workspace, + working_dir=working_dir if isinstance(working_dir, str) else None, + ) + + async def delete_session_graph(self, session_id: str) -> GraphDeleteResult | None: + """In-memory equivalent of ``Neo4jGraphStore.delete_session_graph``. + + Reuses ``resolve_session_graph`` to find the graph, then repeats its + exact traversal (up to the root, back down, then outward stopping at + but not past ``:SST_CONCEPT``) to partition the reachable nodes into + OWNED (removed) vs boundary concept (kept) -- see that method's + docstring for the traversal rule this must never diverge from. + + Raises ``RuntimeError`` if, after removal, any owned node still exists + or any boundary concept node was wrongly removed -- the same gate + ``Neo4jGraphStore.delete_session_graph`` enforces. + """ + graph = await self.resolve_session_graph(session_id) + if graph is None: + return None + + parent_of: dict[str, str] = {} + children: dict[str, list[str]] = {} + outgoing: dict[str, list[str]] = {} + for (src, dst), edata in self._edges.items(): + outgoing.setdefault(src, []).append(dst) + if edata.get("type") in _GRAPH_EDGE_TYPES: + parent_of[dst] = src + children.setdefault(src, []).append(dst) + + owned: set[str] = set() + concept: set[str] = set() + stack = list(graph.session_ids) + visited: set[str] = set() + while stack: + nid = stack.pop() + if nid in visited: + continue + visited.add(nid) + node_data = self._nodes.get(nid) or {} + if "SST_CONCEPT" in node_data.get("labels", []): + concept.add(nid) + continue + owned.add(nid) + stack.extend(outgoing.get(nid, [])) + + relationships_deleted = sum( + 1 for (src, dst) in self._edges if src in owned or dst in owned + ) + + for nid in owned: + self._nodes.pop(nid, None) + self._edges = { + key: data + for key, data in self._edges.items() + if key[0] not in owned and key[1] not in owned + } + + survivors = [nid for nid in owned if nid in self._nodes] + if survivors: + raise RuntimeError( + f"delete_session_graph: owned node(s) survived deletion for " + f"graph root {graph.root_id!r}: {survivors!r}" + ) + missing_concepts = [nid for nid in concept if nid not in self._nodes] + if missing_concepts: + raise RuntimeError( + f"delete_session_graph: shared concept node(s) were wrongly " + f"deleted for graph root {graph.root_id!r}: {missing_concepts!r}" + ) + + return GraphDeleteResult( + root_id=graph.root_id, + nodes_deleted=len(owned), + relationships_deleted=relationships_deleted, + ) + def remove_edge(self, src_id: str, dst_id: str) -> None: """Remove an edge from the in-memory store. @@ -405,3 +608,24 @@ async def touch_session(self, session_id: str, timestamp: str) -> None: timestamp, exc_info=True, ) + + +# --------------------------------------------------------------------------- +# Blob-size composition +# --------------------------------------------------------------------------- + + +async def total_blob_size(blob_store: BlobStore, blob_refs: Iterable[str]) -> int: + """Sum the byte size of every ``ci-blob://`` URI in *blob_refs*. + + Composes ``BlobStore.size()`` over whatever URIs the caller passes in -- + typically every URI the blob store itself listed for the graph's + sessions (``BlobStore.list``) -- the size lookup goes through the + ``BlobStore`` Protocol, never a raw filesystem stat, per the abstraction + principle in docs/02-server-design.md. A missing blob contributes 0 (same + idempotent-on-missing contract as ``BlobStore.size``/``delete_session``). + """ + total = 0 + for uri in blob_refs: + total += await blob_store.size(uri) + return total diff --git a/pyproject.toml b/pyproject.toml index e813d978..aae93b50 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "context-intelligence-server" -version = "6.7.1" +version = "6.9.0" description = "Context Intelligence Server for Amplifier" requires-python = ">=3.11" dependencies = [ diff --git a/tests/integration/test_delete_session_endpoint.py b/tests/integration/test_delete_session_endpoint.py new file mode 100644 index 00000000..c19ee0bb --- /dev/null +++ b/tests/integration/test_delete_session_endpoint.py @@ -0,0 +1,407 @@ +"""End-to-end proof: the session-delete HTTP routes work against a real +running server and a real Neo4j. + +Every other test for this feature either fakes the service (tests/routers/ +test_deletion.py) or calls DeletionService directly, skipping HTTP +(tests/neo4j/test_deletion_service.py). This is the one test that goes all +the way through: real POST /events -> real drain workers -> real Neo4j and +real blob store -> real GET/DELETE routes -> real Neo4j and blob store +again, to check the data is actually gone. + +How the graph is built: + - The session tree itself (one root, two subsessions, one fork) is built + by POSTING REAL EVENTS to /events and waiting for the server's own + drain workers to write them to Neo4j. This part is fully practical + through the normal client path, so that is what is used. + - The blobs on several of those sessions are ALSO produced by posting + real events: an event carrying a field the server offloads to disk + (see blob_processor.BLOB_FIELDS, e.g. "result") makes the server's own + ingest pipeline write a real blob file, exactly as a real client would + trigger. There is no hand-attached node property standing in for a + blob here -- the summary/delete blob count now comes from asking the + blob store what it holds for each session in the graph (see + DeletionService), so a blob written this way is counted honestly. + - The shared "concept" node (the thing every session is allowed to point + at without owning it, e.g. an Agent), and the "still receiving data" + session used for the 409 check, are added by writing directly through + the same Neo4jGraphStore class the server itself uses, rather than + posting more events. Two separate, deliberate reasons: + * The shared concept node: building this edge shape through the real + event pipeline would require reproducing a much longer + agent-delegation event sequence that has nothing to do with + delete. + * The pending/409 session: reusing one of the sessions above would + race against that session's own live background drain worker (it + polls every 50ms and would likely drain an appended record before + the test could observe it as pending) -- a separate session with + no worker attached avoids that race entirely. + +The GET summary call (the preview) and the DELETE calls (which now always +delete -- there is no dry-run flag any more) always go through HTTP, against +the real running FastAPI app. Neither call takes a workspace query +parameter: the server looks up which workspace a session id belongs to on +its own. +""" + +from __future__ import annotations + +import asyncio +from collections.abc import AsyncGenerator +from pathlib import Path +from typing import Any + +import context_intelligence_server.main as main_module +import context_intelligence_server.registry as registry_module +import context_intelligence_server.routers.deletion as deletion_router_module +import httpx +import pytest +from context_intelligence_server.blob_store import AsyncDiskBlobStore +from context_intelligence_server.config import Neo4jClientConfig +from context_intelligence_server.neo4j_store import ( + Neo4jGraphStore, + build_bounded_neo4j_driver, +) +from neo4j import AsyncGraphDatabase + +# Reuse the real-Neo4j-container fixture from the neo4j test tier. Importing +# the fixture function directly (rather than duplicating it) is the normal +# pytest way to share a fixture defined in another folder's conftest.py. +from tests.neo4j.conftest import neo4j_container # noqa: F401 + +pytestmark = [pytest.mark.neo4j, pytest.mark.timeout(120)] + +WORKSPACE = "delete-e2e-ws" + +ROOT = "delete-e2e-root" +SUB1 = "delete-e2e-sub1" +SUB2 = "delete-e2e-sub2" +FORK1 = "delete-e2e-fork1" +OTHER_ROOT = "delete-e2e-other-root" +SHARED_AGENT = "delete-e2e-shared-agent" +PENDING_ROOT = "delete-e2e-pending-root" + +T0 = "2026-01-01T00:00:00+00:00" +T1 = "2026-01-01T00:01:00+00:00" +T2 = "2026-01-01T00:02:00+00:00" +T3 = "2026-01-01T00:03:00+00:00" +T4 = "2026-01-01T00:04:00+00:00" + + +class _E2ESettings: + """A settings-shaped object pointing every store at the test Neo4j + container and at this test's own tmp_path directories. + + Both the ingest path (context_intelligence_server.registry.get_settings) + and the deletion routes (context_intelligence_server.routers.deletion. + get_settings) are pointed at ONE instance of this class, so a blob + written while posting an event and a blob looked up while building the + delete route's blob store resolve to the exact same directory on disk. + """ + + def __init__( + self, container: dict[str, Any], blob_path: str, queues_path: str + ) -> None: + self.blob_path = blob_path + self.queues_path = queues_path + self.write_concurrency = 8 + self.max_delivery_attempts = 5 + self.neo4j_flush_chunk_rows = 100 + self.neo4j_flush_chunk_bytes = 4_194_304 + self.neo4j_lock_timeout = 30.0 + self.neo4j_max_connection_pool_size = 50 + self._container = container + + def resolve_neo4j_admin(self) -> Neo4jClientConfig: + return Neo4jClientConfig( + url=self._container["bolt_url"], + username=self._container["user"], + password=self._container["password"], + access_mode="WRITE", + ) + + def resolve_neo4j_query(self) -> Neo4jClientConfig: + return Neo4jClientConfig( + url=self._container["bolt_url"], + username=self._container["user"], + password=self._container["password"], + access_mode="READ", + ) + + +@pytest.fixture +async def delete_e2e_client( + neo4j_container: dict[str, Any], # noqa: F811 -- pytest fixture parameter + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> AsyncGenerator[httpx.AsyncClient, None]: + """The real FastAPI app, wired to the real test Neo4j container. + + Patches both places that build settings-derived stores for this feature: + - registry.get_settings -- used when POST /events spawns a drain worker + (its blob store, queue paths, and Neo4j admin driver). + - routers.deletion.get_settings -- used when the summary/delete routes + build their own blob store. + Then builds the two Neo4j drivers the deletion routes read directly off + app.state (neo4j_driver for writes, neo4j_query_driver for reads) and + points them at the same container. + """ + settings = _E2ESettings( + neo4j_container, + blob_path=str(tmp_path / "blobs"), + queues_path=str(tmp_path / "queues"), + ) + + monkeypatch.setattr(registry_module, "get_settings", lambda: settings) + monkeypatch.setattr(deletion_router_module, "get_settings", lambda: settings) + + # registry.neo4j_driver is a lazily-built module-level singleton; force a + # rebuild against the patched settings above instead of reusing whatever + # (if anything) a previous test built it as. + main_module.registry._neo4j_driver = None + + admin_driver = build_bounded_neo4j_driver( + settings.resolve_neo4j_admin(), max_connection_pool_size=50 + ) + query_driver = build_bounded_neo4j_driver( + settings.resolve_neo4j_query(), max_connection_pool_size=50 + ) + monkeypatch.setattr( + main_module.app.state, "neo4j_driver", admin_driver, raising=False + ) + monkeypatch.setattr( + main_module.app.state, "neo4j_query_driver", query_driver, raising=False + ) + + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=main_module.app), base_url="http://test" + ) as client: + yield client + + # Shut down drain workers BEFORE closing the shared driver they use -- + # same ordering the real lifespan() uses (registry.shutdown_workers() + # docstring explains why the order matters). + await main_module.registry.shutdown_workers() + await admin_driver.close() + await query_driver.close() + await main_module.registry.close_neo4j_driver() + + # Remove this test's own data so nothing leaks into another test that + # might reuse the same (session-scoped) Neo4j container. + cleanup_driver = AsyncGraphDatabase.driver( + neo4j_container["bolt_url"], + auth=(neo4j_container["user"], neo4j_container["password"]), + ) + await cleanup_driver.execute_query( + "MATCH (n {workspace: $workspace}) DETACH DELETE n", + {"workspace": WORKSPACE}, + ) + await cleanup_driver.close() + + +async def _post_event( + client: httpx.AsyncClient, + event: str, + session_id: str, + timestamp: str, + *, + parent_id: str | None = None, + extra: dict[str, Any] | None = None, +) -> None: + """POST one real event to /events, exactly like a real client would.""" + data: dict[str, Any] = {"session_id": session_id, "timestamp": timestamp} + if parent_id is not None: + data["parent_id"] = parent_id + if extra: + data.update(extra) + response = await client.post( + "/events", + json={"event": event, "workspace": WORKSPACE, "data": data}, + ) + assert response.status_code == 202, response.text + + +async def _wait_drained(session_id: str, timeout_s: float = 15.0) -> None: + """Wait until the durable queue for *session_id* has no pending lines. + + Polls the real queue manager the running server's drain workers use, + the same way tests/integration/test_blob_pipeline.py does. + """ + loop = asyncio.get_event_loop() + deadline = loop.time() + timeout_s + while loop.time() < deadline: + batch = await main_module.registry.queue_manager.read_batch(session_id, 10) + if batch.lines == []: + return + await asyncio.sleep(0.02) + raise AssertionError(f"session {session_id!r} did not drain within {timeout_s}s") + + +async def test_delete_session_endpoint_end_to_end( + delete_e2e_client: httpx.AsyncClient, + neo4j_container: dict[str, Any], # noqa: F811 -- pytest fixture parameter + tmp_path: Path, +) -> None: + client = delete_e2e_client + + # ------------------------------------------------------------------ + # Step 1 -- build the session tree by posting real events. Root, sub2, + # and fork1 each also get a second event carrying a "result" field -- + # a field the server offloads to disk (blob_processor.BLOB_FIELDS) -- + # so each of those three sessions ends up with one real blob file, + # written by the server's own ingest pipeline exactly as a real client + # would trigger it. No blob is attached by hand anywhere in this test. + # ------------------------------------------------------------------ + await _post_event(client, "session:start", ROOT, T0) + await _post_event(client, "session:start", SUB1, T1, parent_id=ROOT) + await _post_event(client, "session:start", SUB2, T2, parent_id=SUB1) + await _post_event(client, "session:fork", FORK1, T3, parent_id=ROOT) + await _post_event(client, "session:start", OTHER_ROOT, T4) + + large_result = {"output": "result payload " * 500} + await _post_event( + client, + "tool:post", + ROOT, + "2026-01-01T00:00:01+00:00", + extra={"result": large_result}, + ) + await _post_event( + client, + "tool:post", + SUB2, + "2026-01-01T00:02:01+00:00", + extra={"result": large_result}, + ) + await _post_event( + client, + "tool:post", + FORK1, + "2026-01-01T00:03:01+00:00", + extra={"result": large_result}, + ) + + for session_id in (ROOT, SUB1, SUB2, FORK1, OTHER_ROOT): + await _wait_drained(session_id) + + # ------------------------------------------------------------------ + # Step 2 -- seed the shared concept node and the pending-check session + # directly through the real Neo4jGraphStore class (see module docstring + # for why these two pieces are not built through posted events). + # ------------------------------------------------------------------ + helper_store = Neo4jGraphStore( + uri=neo4j_container["bolt_url"], + auth=(neo4j_container["user"], neo4j_container["password"]), + workspace=WORKSPACE, + ) + # Same directory the server's own blob store just wrote the real blobs + # to (see _E2ESettings.blob_path) -- used here only to VERIFY what the + # server produced, never to write a blob by hand. + blob_store = AsyncDiskBlobStore(root=tmp_path / "blobs") + try: + await helper_store.upsert_node( + SHARED_AGENT, {"labels": ["Agent", "SST_CONCEPT"], "agent": SHARED_AGENT} + ) + # fork1 and the unrelated other-root session both point at the same + # shared concept node -- neither owns it. + await helper_store.upsert_edge( + FORK1, SHARED_AGENT, {"type": "HAS_AGENT", "sst_semantic": "EXPRESSES"} + ) + await helper_store.upsert_edge( + OTHER_ROOT, SHARED_AGENT, {"type": "HAS_AGENT", "sst_semantic": "EXPRESSES"} + ) + + # A separate, tiny session with no drain worker attached, used only + # for the "still receiving data" (409) check below. + await helper_store.upsert_node( + PENDING_ROOT, + {"labels": ["Session", "RootSession"], "started_at": T0}, + ) + await helper_store.flush() + + # -------------------------------------------------------------- + # Step 3 -- GET summary through HTTP, from a SUBsession id, with NO + # workspace query param -- the server looks up which workspace the + # session belongs to on its own. Checks it resolves the whole + # graph, and that GET deletes nothing: blob_count must be greater + # than 0 and match the number of real blobs the blob store lists + # for the graph's sessions (the proof that the summary now sees + # real blobs without any hand-attached node property), and every + # node/blob is still there afterwards. + # -------------------------------------------------------------- + real_blob_uris: list[str] = [] + for session_id in (ROOT, SUB1, SUB2, FORK1): + real_blob_uris += await blob_store.list(session_id) + assert len(real_blob_uris) == 3, ( + f"expected one real blob each for root/sub2/fork1, found {real_blob_uris!r}" + ) + + summary_resp = await client.get(f"/sessions/{SUB2}/summary") + assert summary_resp.status_code == 200, summary_resp.text + summary = summary_resp.json() + assert summary["root_id"] == ROOT + assert sorted(summary["session_ids"]) == sorted([ROOT, SUB1, SUB2, FORK1]) + assert summary["subsession_count"] == 3 + assert summary["blob_count"] > 0 + assert summary["blob_count"] == len(real_blob_uris) + assert summary["deletable"] is True + assert summary["pending_sessions"] == [] + + for session_id in (ROOT, SUB1, SUB2, FORK1): + assert await helper_store.get_node(session_id) is not None, ( + f"{session_id} should still exist after a GET summary" + ) + assert await blob_store.list(ROOT) != [] + assert await blob_store.list(SUB2) != [] + assert await blob_store.list(FORK1) != [] + + # -------------------------------------------------------------- + # Step 4 -- deleting a session that is still receiving data + # returns 409 and deletes nothing. Uses the separate pending + # session seeded above, with an uncommitted queue record and no + # live drain worker to race against. No workspace query param here + # either. + # -------------------------------------------------------------- + await main_module.registry.queue_manager.append( + PENDING_ROOT, b'{"event": "session:start", "data": {}}' + ) + pending_delete_resp = await client.delete(f"/sessions/{PENDING_ROOT}") + assert pending_delete_resp.status_code == 409, pending_delete_resp.text + assert await helper_store.get_node(PENDING_ROOT) is not None, ( + "a 409 refusal must not delete anything" + ) + + # -------------------------------------------------------------- + # Step 5 -- the real delete, through HTTP. DELETE always deletes + # now -- there is no apply flag, and no workspace query param. + # -------------------------------------------------------------- + delete_resp = await client.delete(f"/sessions/{SUB2}") + assert delete_resp.status_code == 200, delete_resp.text + result = delete_resp.json() + assert result["root_id"] == ROOT + assert result["session_count"] == 4 + assert result["blobs_deleted"] == 3 + assert result["queue_sessions_cleaned"] == 4 + + # -------------------------------------------------------------- + # Step 6 -- verify directly against the real stores. + # -------------------------------------------------------------- + for session_id in (ROOT, SUB1, SUB2, FORK1): + assert await helper_store.get_node(session_id) is None, ( + f"{session_id} should be gone" + ) + assert await blob_store.list(session_id) == [], ( + f"{session_id}'s blobs should be gone" + ) + qm = main_module.registry.queue_manager + assert not qm._log_path(session_id).exists() + assert not qm._offset_path(session_id).exists() + assert not qm._dead_path(session_id).exists() + + agent_node = await helper_store.get_node(SHARED_AGENT) + assert agent_node is not None + assert "SST_CONCEPT" in agent_node.get("labels", []) + + assert await helper_store.get_node(OTHER_ROOT) is not None + assert await helper_store.get_edge(OTHER_ROOT, SHARED_AGENT) is not None + finally: + await helper_store.close() diff --git a/tests/neo4j/test_delete_session_graph.py b/tests/neo4j/test_delete_session_graph.py new file mode 100644 index 00000000..602eceee --- /dev/null +++ b/tests/neo4j/test_delete_session_graph.py @@ -0,0 +1,231 @@ +"""Tier 3 - Neo4j integration proof for Neo4jGraphStore.delete_session_graph. + +Ingests a multi-session graph (root + 2 subsessions + 1 fork) that shares a +:SST_CONCEPT node (Agent) with a SEPARATE, unrelated graph, then deletes via +a SUB-session id and proves: + + (a) every owned graph node is gone (root + all descendants + their + Events/ToolCalls/Delegation); + (b) the shared :SST_CONCEPT node STILL EXISTS; + (c) the unrelated graph reachable only through that shared concept node is + UNTOUCHED (its own nodes and its edge to the concept node both survive); + (d) the returned counts (nodes_deleted / relationships_deleted) match what + was actually removed; + (e) deleting an unknown session id returns None / no-op (nothing deleted). + +Run: uv run --group dev pytest tests/neo4j/test_delete_session_graph.py -v -m neo4j +""" + +from __future__ import annotations + +from typing import Any + +import pytest + +pytestmark = pytest.mark.neo4j + + +async def _build_graph(store: Any) -> None: + """Seed a root + 2 subsessions + 1 fork, with blobs on several sessions. + + Tree shape: root -[HAS_SUBSESSION]-> sub1 -[HAS_SUBSESSION]-> sub2 + root -[FORKED]-> fork1 + + A shared Agent (:SST_CONCEPT) is reached from fork1's Delegation. A + completely SEPARATE, unrelated session graph also reaches the SAME + shared Agent, proving the delete of THIS graph must not touch the + other one. + """ + store.created_by = "colombod" + await store.upsert_node( + "df-fam-root", + {"labels": ["Session", "RootSession"], "started_at": "2026-01-01T00:00:00Z"}, + ) + await store.upsert_node( + "df-fam-sub1", + { + "labels": ["Session", "SubSession"], + "parent_id": "df-fam-root", + "started_at": "2026-01-01T00:01:00Z", + }, + ) + await store.upsert_edge("df-fam-root", "df-fam-sub1", {"type": "HAS_SUBSESSION"}) + await store.upsert_node( + "df-fam-sub2", + { + "labels": ["Session", "SubSession"], + "parent_id": "df-fam-sub1", + "started_at": "2026-01-01T00:02:00Z", + }, + ) + await store.upsert_edge("df-fam-sub1", "df-fam-sub2", {"type": "HAS_SUBSESSION"}) + await store.upsert_node( + "df-fam-fork1", + { + "labels": ["Session", "ForkedSession"], + "parent_id": "df-fam-root", + "started_at": "2026-01-01T00:03:00Z", + }, + ) + await store.upsert_edge("df-fam-root", "df-fam-fork1", {"type": "FORKED"}) + + await store.upsert_node( + "df-fam-root::orch::1", + { + "labels": ["OrchestratorRun", "SST_EVENT"], + "raw": {"$blob_ref": "ci-blob://df-fam-root/orch1"}, + }, + ) + await store.upsert_edge( + "df-fam-root", "df-fam-root::orch::1", {"type": "HAS_EXECUTION"} + ) + + await store.upsert_node( + "df-fam-sub2::tool::1", + { + "labels": ["ToolCall", "SST_EVENT"], + "result": {"$blob_ref": "ci-blob://df-fam-sub2/tool1"}, + }, + ) + await store.upsert_edge( + "df-fam-sub2", "df-fam-sub2::tool::1", {"type": "HAS_TOOL_CALL"} + ) + + await store.upsert_node( + "df-fam-fork1::delegation::1", + { + "labels": ["Delegation", "SST_EVENT"], + "messages": {"$blob_ref": "ci-blob://df-fam-fork1/del1"}, + }, + ) + await store.upsert_edge( + "df-fam-fork1", "df-fam-fork1::delegation::1", {"type": "TRIGGERED"} + ) + + # Shared concept node -- must survive the delete of df-fam-* below. + await store.upsert_node("df-agent-shared", {"labels": ["Agent", "SST_CONCEPT"]}) + await store.upsert_edge( + "df-fam-fork1::delegation::1", "df-agent-shared", {"type": "HAS_AGENT"} + ) + + await store.flush() + + +async def _build_unrelated_graph_sharing_concept(store: Any) -> None: + """A SEPARATE, unrelated graph reaching the SAME shared Agent node. + + Reachable ONLY through the shared concept -- deleting df-fam-* must not + touch any of this. + """ + await store.upsert_node( + "other-fam-root", + {"labels": ["Session", "RootSession"], "started_at": "2026-02-01T00:00:00Z"}, + ) + await store.upsert_node( + "other-fam-root::delegation::1", + {"labels": ["Delegation", "SST_EVENT"]}, + ) + await store.upsert_edge( + "other-fam-root", "other-fam-root::delegation::1", {"type": "TRIGGERED"} + ) + await store.upsert_edge( + "other-fam-root::delegation::1", "df-agent-shared", {"type": "HAS_AGENT"} + ) + await store.flush() + + +class TestDeleteSessionGraphNeo4j: + """Neo4jGraphStore.delete_session_graph against a real Neo4j.""" + + async def test_returns_none_for_unknown_session(self, neo4j_services: Any) -> None: + store = neo4j_services.graph + assert await store.delete_session_graph("does-not-exist") is None + + async def test_unknown_session_deletes_nothing(self, neo4j_services: Any) -> None: + store = neo4j_services.graph + await _build_graph(store) + await _build_unrelated_graph_sharing_concept(store) + + assert await store.delete_session_graph("does-not-exist") is None + + # Nothing from either graph was touched. + assert await store.get_node("df-fam-root") is not None + assert await store.get_node("other-fam-root") is not None + + async def test_owned_graph_nodes_all_removed(self, neo4j_services: Any) -> None: + store = neo4j_services.graph + await _build_graph(store) + await _build_unrelated_graph_sharing_concept(store) + + result = await store.delete_session_graph("df-fam-sub2") + assert result is not None + assert result.root_id == "df-fam-root" + + for node_id in ( + "df-fam-root", + "df-fam-sub1", + "df-fam-sub2", + "df-fam-fork1", + "df-fam-root::orch::1", + "df-fam-sub2::tool::1", + "df-fam-fork1::delegation::1", + ): + assert await store.get_node(node_id) is None, ( + f"{node_id} should have been deleted" + ) + + async def test_shared_concept_node_survives(self, neo4j_services: Any) -> None: + store = neo4j_services.graph + await _build_graph(store) + await _build_unrelated_graph_sharing_concept(store) + + await store.delete_session_graph("df-fam-root") + + agent = await store.get_node("df-agent-shared") + assert agent is not None + assert "SST_CONCEPT" in agent.get("labels", []) + + async def test_unrelated_graph_reachable_only_via_shared_concept_untouched( + self, neo4j_services: Any + ) -> None: + store = neo4j_services.graph + await _build_graph(store) + await _build_unrelated_graph_sharing_concept(store) + + await store.delete_session_graph("df-fam-fork1") + + assert await store.get_node("other-fam-root") is not None + assert await store.get_node("other-fam-root::delegation::1") is not None + assert ( + await store.get_edge("other-fam-root::delegation::1", "df-agent-shared") + is not None + ) + + async def test_counts_match_deleted_nodes_and_relationships( + self, neo4j_services: Any + ) -> None: + store = neo4j_services.graph + await _build_graph(store) + await _build_unrelated_graph_sharing_concept(store) + + result = await store.delete_session_graph("df-fam-root") + assert result is not None + # Owned = 4 sessions + orch + tool + delegation = 7 (agent excluded). + assert result.nodes_deleted == 7 + # root->sub1, sub1->sub2, root->fork1, root->orch, sub2->tool, + # fork1->delegation, delegation->agent = 7 (agent->other-delegation + # edge belongs to the unrelated graph and is excluded). + assert result.relationships_deleted == 7 + + async def test_sub_session_and_root_delete_identical_graph( + self, neo4j_services: Any + ) -> None: + """A sub-session id and its root id must delete the identical graph.""" + store_a = neo4j_services.graph + await _build_graph(store_a) + await _build_unrelated_graph_sharing_concept(store_a) + result_a = await store_a.delete_session_graph("df-fam-sub1") + assert result_a is not None + assert result_a.root_id == "df-fam-root" + assert result_a.nodes_deleted == 7 + assert result_a.relationships_deleted == 7 diff --git a/tests/neo4j/test_deletion_service.py b/tests/neo4j/test_deletion_service.py new file mode 100644 index 00000000..497c4b02 --- /dev/null +++ b/tests/neo4j/test_deletion_service.py @@ -0,0 +1,262 @@ +"""Tier 3 -- Neo4j integration proof for DeletionService (A4). + +DeletionService composed with a REAL Neo4jGraphStore, a REAL +AsyncDiskBlobStore, and a REAL QueueManager. Ingests a multi-session graph +(root + 2 subsessions + 1 fork) with blobs on several of its sessions that +shares a :SST_CONCEPT node (Agent) with a SEPARATE, unrelated graph, drains +each session's queue, then applies deletion via a SUB-session id and proves: + + (a) every session node in the graph is gone (root + all descendants), + (b) every graph blob is purged (count reconciles to the whole-graph total), + (c) queue/dead-letter artifacts for the graph are gone, + (d) the shared concept node survives with only its edges-into-graph removed, + (e) the unrelated graph reachable only through the shared concept is untouched. + +Run: uv run --group dev pytest tests/neo4j/test_deletion_service.py -q -m neo4j +""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + +import pytest +from context_intelligence_server.blob_store import AsyncDiskBlobStore +from context_intelligence_server.deletion import DeletionService +from context_intelligence_server.queue_manager import QueueManager + +pytestmark = pytest.mark.neo4j + + +async def _drain(queue_manager: QueueManager, session_id: str, raw: bytes) -> None: + await queue_manager.append(session_id, raw) + batch = await queue_manager.read_batch(session_id, max_items=10) + await queue_manager.commit(session_id, batch.end_offset) + + +async def _build_graph( + store: Any, blob_store: AsyncDiskBlobStore, queue_manager: QueueManager +) -> frozenset[str]: + """Seed a root + 2 subsessions + 1 fork, with blobs on several sessions. + + Tree shape: root -[HAS_SUBSESSION]-> sub1 -[HAS_SUBSESSION]-> sub2 + root -[FORKED]-> fork1 + + A shared Agent (:SST_CONCEPT) is reached from fork1's Delegation. Every + session's queue is drained. Returns the whole-graph blob URI set. + """ + store.created_by = "colombod" + + blob_root = await blob_store.write("ds-fam-root", "orch1", {"v": "root"}) + blob_sub2 = await blob_store.write("ds-fam-sub2", "tool1", {"v": "sub2"}) + blob_fork1 = await blob_store.write("ds-fam-fork1", "del1", {"v": "fork1"}) + + await store.upsert_node( + "ds-fam-root", + {"labels": ["Session", "RootSession"], "started_at": "2026-01-01T00:00:00Z"}, + ) + await store.upsert_node( + "ds-fam-sub1", + { + "labels": ["Session", "SubSession"], + "parent_id": "ds-fam-root", + "started_at": "2026-01-01T00:01:00Z", + }, + ) + await store.upsert_edge("ds-fam-root", "ds-fam-sub1", {"type": "HAS_SUBSESSION"}) + await store.upsert_node( + "ds-fam-sub2", + { + "labels": ["Session", "SubSession"], + "parent_id": "ds-fam-sub1", + "started_at": "2026-01-01T00:02:00Z", + }, + ) + await store.upsert_edge("ds-fam-sub1", "ds-fam-sub2", {"type": "HAS_SUBSESSION"}) + await store.upsert_node( + "ds-fam-fork1", + { + "labels": ["Session", "ForkedSession"], + "parent_id": "ds-fam-root", + "started_at": "2026-01-01T00:03:00Z", + }, + ) + await store.upsert_edge("ds-fam-root", "ds-fam-fork1", {"type": "FORKED"}) + + await store.upsert_node( + "ds-fam-root::orch::1", + {"labels": ["OrchestratorRun", "SST_EVENT"]}, + ) + await store.upsert_edge( + "ds-fam-root", "ds-fam-root::orch::1", {"type": "HAS_EXECUTION"} + ) + + await store.upsert_node( + "ds-fam-sub2::tool::1", + {"labels": ["ToolCall", "SST_EVENT"]}, + ) + await store.upsert_edge( + "ds-fam-sub2", "ds-fam-sub2::tool::1", {"type": "HAS_TOOL_CALL"} + ) + + await store.upsert_node( + "ds-fam-fork1::delegation::1", + {"labels": ["Delegation", "SST_EVENT"]}, + ) + await store.upsert_edge( + "ds-fam-fork1", "ds-fam-fork1::delegation::1", {"type": "TRIGGERED"} + ) + + # Shared concept node -- must survive the delete of ds-fam-* below. + await store.upsert_node("ds-agent-shared", {"labels": ["Agent", "SST_CONCEPT"]}) + await store.upsert_edge( + "ds-fam-fork1::delegation::1", "ds-agent-shared", {"type": "HAS_AGENT"} + ) + + await store.flush() + + for sid in ("ds-fam-root", "ds-fam-sub1", "ds-fam-sub2", "ds-fam-fork1"): + await _drain( + queue_manager, sid, json.dumps({"event": "tool:pre"}).encode("utf-8") + ) + await queue_manager.dead_letter("ds-fam-sub2", b"bad-line", "boom") + + return frozenset({blob_root, blob_sub2, blob_fork1}) + + +async def _build_unrelated_graph_sharing_concept(store: Any) -> None: + """A SEPARATE, unrelated graph reaching the SAME shared Agent node. + + Reachable ONLY through the shared concept -- deleting ds-fam-* must not + touch any of this. + """ + await store.upsert_node( + "ds-other-fam-root", + {"labels": ["Session", "RootSession"], "started_at": "2026-02-01T00:00:00Z"}, + ) + await store.upsert_node( + "ds-other-fam-root::delegation::1", + {"labels": ["Delegation", "SST_EVENT"]}, + ) + await store.upsert_edge( + "ds-other-fam-root", + "ds-other-fam-root::delegation::1", + {"type": "TRIGGERED"}, + ) + await store.upsert_edge( + "ds-other-fam-root::delegation::1", "ds-agent-shared", {"type": "HAS_AGENT"} + ) + await store.flush() + + +class TestDeletionServiceNeo4j: + """DeletionService composed with a real Neo4jGraphStore + real blob/queue.""" + + async def test_preview_reports_whole_graph_facts( + self, neo4j_services: Any, tmp_path: Path + ) -> None: + store = neo4j_services.graph + blob_store = AsyncDiskBlobStore(tmp_path / "blobs") + queue_manager = QueueManager(tmp_path / "queues") + service = DeletionService(store, blob_store, queue_manager) + + blob_refs = await _build_graph(store, blob_store, queue_manager) + await _build_unrelated_graph_sharing_concept(store) + + preview = await service.preview("ds-fam-sub2") + assert preview is not None + assert preview.root_id == "ds-fam-root" + assert preview.session_ids == frozenset( + {"ds-fam-root", "ds-fam-sub1", "ds-fam-sub2", "ds-fam-fork1"} + ) + assert preview.blob_count == len(blob_refs) == 3 + assert preview.deletable is True + assert preview.pending_sessions == [] + assert preview.created_by == "colombod" + + async def test_apply_deletes_whole_graph_and_preserves_shared_concept( + self, neo4j_services: Any, tmp_path: Path + ) -> None: + store = neo4j_services.graph + blob_store = AsyncDiskBlobStore(tmp_path / "blobs") + queue_manager = QueueManager(tmp_path / "queues") + service = DeletionService(store, blob_store, queue_manager) + + blob_refs = await _build_graph(store, blob_store, queue_manager) + await _build_unrelated_graph_sharing_concept(store) + + result = await service.apply("ds-fam-sub1", requested_by="tester") + + assert result is not None + assert result.root_id == "ds-fam-root" + assert result.session_count == 4 + # Owned = 4 sessions + orch + tool + delegation = 7 (agent excluded). + assert result.nodes_deleted == 7 + assert result.relationships_deleted == 7 + assert result.blobs_deleted == len(blob_refs) == 3 + assert result.queue_sessions_cleaned == 4 + + # (a) every session node in the graph is gone. + for node_id in ( + "ds-fam-root", + "ds-fam-sub1", + "ds-fam-sub2", + "ds-fam-fork1", + "ds-fam-root::orch::1", + "ds-fam-sub2::tool::1", + "ds-fam-fork1::delegation::1", + ): + assert await store.get_node(node_id) is None, f"{node_id} should be gone" + + # (b) every graph blob is purged. + for sid in ("ds-fam-root", "ds-fam-sub1", "ds-fam-sub2", "ds-fam-fork1"): + assert await blob_store.list(sid) == [] + + # (c) queue/dead-letter artifacts for the graph are gone. + for sid in ("ds-fam-root", "ds-fam-sub1", "ds-fam-sub2", "ds-fam-fork1"): + assert not queue_manager._log_path(sid).exists() + assert not queue_manager._offset_path(sid).exists() + assert not queue_manager._dead_path(sid).exists() + + # (d) the shared concept node survives. + agent = await store.get_node("ds-agent-shared") + assert agent is not None + assert "SST_CONCEPT" in agent.get("labels", []) + + # (e) the unrelated graph, reachable only via the shared concept, is untouched. + assert await store.get_node("ds-other-fam-root") is not None + assert await store.get_node("ds-other-fam-root::delegation::1") is not None + assert ( + await store.get_edge("ds-other-fam-root::delegation::1", "ds-agent-shared") + is not None + ) + + async def test_apply_refuses_when_a_graph_session_has_pending_records( + self, neo4j_services: Any, tmp_path: Path + ) -> None: + store = neo4j_services.graph + blob_store = AsyncDiskBlobStore(tmp_path / "blobs") + queue_manager = QueueManager(tmp_path / "queues") + service = DeletionService(store, blob_store, queue_manager) + + await _build_graph(store, blob_store, queue_manager) + # Leave sub2 with an uncommitted append after the drain above. + await queue_manager.append("ds-fam-sub2", b'{"event": "late"}') + + with pytest.raises(RuntimeError, match="pending"): + await service.apply("ds-fam-root") + + # Nothing deleted. + assert await store.get_node("ds-fam-root") is not None + assert await store.get_node("ds-fam-sub2") is not None + + async def test_apply_unknown_session_returns_none( + self, neo4j_services: Any, tmp_path: Path + ) -> None: + store = neo4j_services.graph + blob_store = AsyncDiskBlobStore(tmp_path / "blobs") + queue_manager = QueueManager(tmp_path / "queues") + service = DeletionService(store, blob_store, queue_manager) + + assert await service.apply("does-not-exist") is None diff --git a/tests/neo4j/test_resolve_session_graph.py b/tests/neo4j/test_resolve_session_graph.py new file mode 100644 index 00000000..97d6cfb4 --- /dev/null +++ b/tests/neo4j/test_resolve_session_graph.py @@ -0,0 +1,274 @@ +"""Tier 3 - Neo4j integration proof for Neo4jGraphStore.resolve_session_graph. + +Ingests a multi-session graph (root + 2 subsessions + 1 fork) with a few +extra event nodes attached to nodes across several of those sessions (used +to build up node/edge counts -- blobs themselves are not this store's job, +see BlobStore.list instead), then proves: + + (a) resolving from a SUB-session id yields the SAME graph (same session-id + set) as resolving from the root id; + (b) started_at and last_change are present, and last_change reflects the + MAX across the graph (not just the root -- touch_session only ever + updates the direct node's last_updated, per services.py); + (c) node/edge counts match the constructed graph exactly. + +Run: uv run pytest tests/neo4j/test_resolve_session_graph.py -v -m neo4j +""" + +from __future__ import annotations + +from datetime import datetime, timezone +from typing import Any + +import pytest + +pytestmark = pytest.mark.neo4j + + +async def _build_graph(store: Any) -> None: + """Seed a root + 2 subsessions + 1 fork, with extra event nodes on several sessions. + + Tree shape: root -[HAS_SUBSESSION]-> sub1 -[HAS_SUBSESSION]-> sub2 + root -[FORKED]-> fork1 + + Event nodes attached at three different graph sessions (root, sub2, + fork1), plus a shared Agent (:SST_CONCEPT) reached from fork1's + Delegation, with an out-of-graph node hanging off the Agent to prove + traversal stops there. + """ + store.created_by = "colombod" + await store.upsert_node( + "nf-fam-root", + { + "labels": ["Session", "RootSession"], + "started_at": "2026-01-01T00:00:00Z", + "working_dir": "/mnt/workspaces/project-2501", + }, + ) + await store.upsert_node( + "nf-fam-sub1", + { + "labels": ["Session", "SubSession"], + "parent_id": "nf-fam-root", + "started_at": "2026-01-01T00:01:00Z", + }, + ) + await store.upsert_edge("nf-fam-root", "nf-fam-sub1", {"type": "HAS_SUBSESSION"}) + await store.upsert_node( + "nf-fam-sub2", + { + "labels": ["Session", "SubSession"], + "parent_id": "nf-fam-sub1", + "started_at": "2026-01-01T00:02:00Z", + "last_updated": "2026-01-01T00:05:00Z", + }, + ) + await store.upsert_edge("nf-fam-sub1", "nf-fam-sub2", {"type": "HAS_SUBSESSION"}) + await store.upsert_node( + "nf-fam-fork1", + { + "labels": ["Session", "ForkedSession"], + "parent_id": "nf-fam-root", + "started_at": "2026-01-01T00:03:00Z", + }, + ) + await store.upsert_edge("nf-fam-root", "nf-fam-fork1", {"type": "FORKED"}) + + # Extra event nodes across several graph sessions. + await store.upsert_node( + "nf-fam-root::orch::1", + {"labels": ["OrchestratorRun", "SST_EVENT"]}, + ) + await store.upsert_edge( + "nf-fam-root", "nf-fam-root::orch::1", {"type": "HAS_EXECUTION"} + ) + + await store.upsert_node( + "nf-fam-sub2::tool::1", + {"labels": ["ToolCall", "SST_EVENT"]}, + ) + await store.upsert_edge( + "nf-fam-sub2", "nf-fam-sub2::tool::1", {"type": "HAS_TOOL_CALL"} + ) + + await store.upsert_node( + "nf-fam-fork1::delegation::1", + {"labels": ["Delegation", "SST_EVENT"]}, + ) + await store.upsert_edge( + "nf-fam-fork1", "nf-fam-fork1::delegation::1", {"type": "TRIGGERED"} + ) + + # Shared concept node -- traversal must stop here, not continue past it. + await store.upsert_node("nf-agent-shared", {"labels": ["Agent", "SST_CONCEPT"]}) + await store.upsert_edge( + "nf-fam-fork1::delegation::1", "nf-agent-shared", {"type": "HAS_AGENT"} + ) + await store.upsert_node( + "nf-other-session-xyz::leak", + {"labels": ["ToolCall", "SST_EVENT"]}, + ) + await store.upsert_edge( + "nf-agent-shared", "nf-other-session-xyz::leak", {"type": "SOME_EDGE"} + ) + + await store.flush() + + +class TestResolveSessionGraphNeo4j: + """Neo4jGraphStore.resolve_session_graph against a real Neo4j.""" + + async def test_returns_none_for_unknown_session(self, neo4j_services: Any) -> None: + store = neo4j_services.graph + assert await store.resolve_session_graph("does-not-exist") is None + + async def test_sub_session_and_root_resolve_identical_graph( + self, neo4j_services: Any + ) -> None: + store = neo4j_services.graph + await _build_graph(store) + + from_root = await store.resolve_session_graph("nf-fam-root") + from_sub = await store.resolve_session_graph("nf-fam-sub2") + from_fork = await store.resolve_session_graph("nf-fam-fork1") + + assert from_root is not None + expected_ids = frozenset( + {"nf-fam-root", "nf-fam-sub1", "nf-fam-sub2", "nf-fam-fork1"} + ) + assert from_root.root_id == "nf-fam-root" + assert from_root.session_ids == expected_ids + + assert from_sub is not None + assert from_sub.root_id == "nf-fam-root" + assert from_sub.session_ids == expected_ids + + assert from_fork is not None + assert from_fork.root_id == "nf-fam-root" + assert from_fork.session_ids == expected_ids + + async def test_subsession_count_excludes_root(self, neo4j_services: Any) -> None: + store = neo4j_services.graph + await _build_graph(store) + graph = await store.resolve_session_graph("nf-fam-root") + assert graph is not None + assert graph.subsession_count == 3 + + async def test_started_at_and_last_change_present( + self, neo4j_services: Any + ) -> None: + """(b) started_at/last_change present; last_change is the graph MAX.""" + store = neo4j_services.graph + await _build_graph(store) + graph = await store.resolve_session_graph("nf-fam-root") + assert graph is not None + + assert graph.started_at is not None + assert graph.started_at == datetime(2026, 1, 1, 0, 0, 0, tzinfo=timezone.utc) + + # fam-sub2's last_updated (00:05) is later than root's own timestamps, + # proving last_change is computed across the graph, not just root. + assert graph.last_change is not None + assert graph.last_change == datetime(2026, 1, 1, 0, 5, 0, tzinfo=timezone.utc) + + async def test_node_and_edge_counts_match_constructed_graph( + self, neo4j_services: Any + ) -> None: + """(c) node/edge counts match the constructed graph exactly.""" + store = neo4j_services.graph + await _build_graph(store) + graph = await store.resolve_session_graph("nf-fam-root") + assert graph is not None + + # 4 sessions + orchestrator run + tool call + delegation + agent + # (boundary node, included but not expanded past) = 8. + assert graph.node_count == 8 + # root->sub1, sub1->sub2, root->fork1, root->orch, sub2->tool, + # fork1->delegation, delegation->agent = 7 (agent->leak excluded). + assert graph.edge_count == 7 + + async def test_created_by_from_root(self, neo4j_services: Any) -> None: + store = neo4j_services.graph + await _build_graph(store) + graph = await store.resolve_session_graph("nf-fam-fork1") + assert graph is not None + assert graph.created_by == "colombod" + + async def test_working_dir_from_root(self, neo4j_services: Any) -> None: + store = neo4j_services.graph + await _build_graph(store) + graph = await store.resolve_session_graph("nf-fam-fork1") + assert graph is not None + assert graph.working_dir == "/mnt/workspaces/project-2501" + + async def test_resolves_regardless_of_callers_bound_workspace( + self, neo4j_services: Any, neo4j_container: dict[str, Any] + ) -> None: + """resolve_session_graph takes no workspace argument -- it discovers + the workspace from the session id itself. A store instance bound to + a DIFFERENT workspace at construction must still resolve the graph, + and must report the workspace the session actually lives in (not + the store's own bound workspace).""" + from context_intelligence_server.neo4j_store import Neo4jGraphStore + + store = neo4j_services.graph + await _build_graph(store) + + other_store = Neo4jGraphStore( + uri=neo4j_container["bolt_url"], + auth=(neo4j_container["user"], neo4j_container["password"]), + workspace="other-workspace", + ) + try: + found = await other_store.resolve_session_graph("nf-fam-root") + assert found is not None, ( + "the workspace is discovered from the session id, not " + "supplied by the caller, so a differently-bound store must " + "still find it" + ) + assert found.workspace == "test", ( + "the reported workspace must be the one the session " + "actually lives in, not other_store's own bound workspace" + ) + finally: + await other_store.close() + + async def test_ambiguous_session_id_across_workspaces_raises( + self, neo4j_services: Any, neo4j_container: dict[str, Any] + ) -> None: + """The SAME session id used in TWO DIFFERENT workspaces is refused + loudly rather than silently resolving one of them. + + Session ids are supposed to be unique, so this should not happen in + practice -- but if it ever does, resolving (or deleting) by id alone + must raise instead of guessing which workspace was meant. + """ + from context_intelligence_server.graph_store import AmbiguousSessionError + from context_intelligence_server.neo4j_store import Neo4jGraphStore + + store = neo4j_services.graph + await _build_graph(store) + + other_store = Neo4jGraphStore( + uri=neo4j_container["bolt_url"], + auth=(neo4j_container["user"], neo4j_container["password"]), + workspace="other-workspace", + ) + try: + # Put a session with the SAME node_id "nf-fam-root" into a + # SECOND, different workspace. + await other_store.upsert_node( + "nf-fam-root", + { + "labels": ["Session", "RootSession"], + "started_at": "2026-05-01T00:00:00Z", + }, + ) + await other_store.flush() + + with pytest.raises(AmbiguousSessionError, match="more than one workspace"): + await store.resolve_session_graph("nf-fam-root") + with pytest.raises(AmbiguousSessionError, match="more than one workspace"): + await other_store.resolve_session_graph("nf-fam-root") + finally: + await other_store.close() diff --git a/tests/routers/test_deletion.py b/tests/routers/test_deletion.py new file mode 100644 index 00000000..b27d4f88 --- /dev/null +++ b/tests/routers/test_deletion.py @@ -0,0 +1,358 @@ +"""Tests for the session-deletion routes: GET .../summary and DELETE .../{session_id}. + +These are router tests, not full end-to-end tests: they replace the two +dependency functions that build a real DeletionService (``read_deletion_service`` +and ``delete_route_service``) with a fake service, using +``app.dependency_overrides``. This proves the route-to-service wiring (which +session id reaches the service, how a 404/409 is produced) and the read/write +auth gating, without needing a real Neo4j connection. A real-server-plus-Neo4j +end-to-end test is a separate, later item. +""" + +from __future__ import annotations + +from collections.abc import AsyncIterator, MutableMapping +from contextlib import asynccontextmanager +from datetime import datetime +from typing import Any + +import httpx +import pytest +from context_intelligence_server.authz import require_write +from context_intelligence_server.deletion import ( + DeletionPreview, + DeletionResult, + SessionsPendingError, +) +from context_intelligence_server.graph_store import AmbiguousSessionError +from context_intelligence_server.main import app +from context_intelligence_server.routers import deletion as deletion_router +from fastapi import HTTPException + + +def _sample_preview(**overrides: Any) -> DeletionPreview: + values: dict[str, Any] = { + "root_id": "root-1", + "session_ids": frozenset({"root-1", "sub-1"}), + "node_count": 10, + "edge_count": 5, + "blob_count": 2, + "created_by": "alice", + "started_at": datetime(2024, 1, 1, 12, 0, 0), + "last_change": datetime(2024, 1, 2, 8, 30, 0), + "subsession_count": 1, + "workspace": "ws1", + "working_dir": "/home/alice/project", + "deletable": True, + "pending_sessions": [], + } + values.update(overrides) + return DeletionPreview(**values) + + +def _sample_result(**overrides: Any) -> DeletionResult: + values: dict[str, Any] = { + "root_id": "root-1", + "session_count": 2, + "nodes_deleted": 10, + "relationships_deleted": 5, + "blobs_deleted": 2, + "queue_sessions_cleaned": 2, + } + values.update(overrides) + return DeletionResult(**values) + + +class _FakeDeletionService: + """Stands in for a real DeletionService. Records what it was called with.""" + + def __init__( + self, + preview: DeletionPreview | None = None, + result: DeletionResult | None = None, + apply_error: Exception | None = None, + preview_error: Exception | None = None, + ) -> None: + self._preview = preview + self._result = result + self._apply_error = apply_error + self._preview_error = preview_error + self.preview_calls: list[str] = [] + self.apply_calls: list[tuple[str, str | None]] = [] + + async def preview(self, session_id: str) -> DeletionPreview | None: + self.preview_calls.append(session_id) + if self._preview_error is not None: + raise self._preview_error + return self._preview + + async def apply( + self, session_id: str, *, requested_by: str | None = None + ) -> DeletionResult | None: + self.apply_calls.append((session_id, requested_by)) + if self._apply_error is not None: + raise self._apply_error + return self._result + + +def _reject_write() -> None: + """A require_write stand-in for a caller who is not write-capable.""" + raise HTTPException(status_code=403, detail="write access refused (test double)") + + +@pytest.fixture(autouse=True) +def _clear_overrides() -> Any: + """Every test starts and ends with a clean dependency_overrides map.""" + yield + app.dependency_overrides.pop(deletion_router.read_deletion_service, None) + app.dependency_overrides.pop(deletion_router.delete_route_service, None) + app.dependency_overrides.pop(require_write, None) + + +def _override_read_service(fake: _FakeDeletionService) -> None: + async def _fake() -> _FakeDeletionService: + return fake + + app.dependency_overrides[deletion_router.read_deletion_service] = _fake + + +def _override_delete_service(fake: _FakeDeletionService) -> None: + async def _fake() -> _FakeDeletionService: + return fake + + app.dependency_overrides[deletion_router.delete_route_service] = _fake + + +@asynccontextmanager +async def _client_with_scope_state( + state: dict[str, Any], +) -> AsyncIterator[httpx.AsyncClient]: + """A client whose requests carry the given scope state (e.g. contributor_id). + + ``app`` (used by the plain ``client`` fixture) has no auth middleware, so + ``request.scope["state"]`` is otherwise empty. This wraps ``app`` with a + minimal ASGI layer that injects the given state before the request + reaches any route -- enough to prove the router reads + ``contributor_id`` correctly, without building a real bearer token. + """ + + async def _wrapped( + scope: MutableMapping[str, Any], receive: Any, send: Any + ) -> None: + if scope["type"] == "http": + scope = {**scope, "state": dict(state)} + await app(scope, receive, send) + + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=_wrapped), base_url="http://test" + ) as c: + yield c + + +class TestGetSessionSummary: + @pytest.mark.anyio + async def test_known_session_returns_expected_fields( + self, client: httpx.AsyncClient + ) -> None: + preview = _sample_preview() + fake = _FakeDeletionService(preview=preview) + _override_read_service(fake) + + response = await client.get("/sessions/root-1/summary") + + assert response.status_code == 200 + body = response.json() + assert body == { + "root_id": "root-1", + "session_ids": ["root-1", "sub-1"], + "node_count": 10, + "edge_count": 5, + "blob_count": 2, + "created_by": "alice", + "started_at": "2024-01-01T12:00:00", + "last_change": "2024-01-02T08:30:00", + "subsession_count": 1, + "workspace": "ws1", + "working_dir": "/home/alice/project", + "deletable": True, + "pending_sessions": [], + } + assert fake.preview_calls == ["root-1"] + + @pytest.mark.anyio + async def test_no_workspace_query_param_needed( + self, client: httpx.AsyncClient + ) -> None: + """The summary route takes no workspace query param at all -- the + session id alone is enough to reach the service.""" + fake = _FakeDeletionService(preview=_sample_preview()) + _override_read_service(fake) + + response = await client.get("/sessions/root-1/summary") + + assert response.status_code == 200 + assert fake.preview_calls == ["root-1"] + + @pytest.mark.anyio + async def test_unknown_session_returns_404(self, client: httpx.AsyncClient) -> None: + fake = _FakeDeletionService(preview=None) + _override_read_service(fake) + + response = await client.get("/sessions/does-not-exist/summary") + + assert response.status_code == 404 + + @pytest.mark.anyio + async def test_ambiguous_session_id_returns_409( + self, client: httpx.AsyncClient + ) -> None: + """A session id found in more than one workspace is a 409, not a 500 + or a silent guess.""" + fake = _FakeDeletionService( + preview_error=AmbiguousSessionError("root-1", ["ws1", "ws2"]) + ) + _override_read_service(fake) + + response = await client.get("/sessions/root-1/summary") + + assert response.status_code == 409 + assert "more than one workspace" in response.json()["detail"] + + @pytest.mark.anyio + async def test_allows_a_caller_who_fails_require_write( + self, client: httpx.AsyncClient + ) -> None: + """The summary route is gated by require_read only. A caller who is + refused by require_write must still be able to read the summary.""" + fake = _FakeDeletionService(preview=_sample_preview()) + _override_read_service(fake) + app.dependency_overrides[require_write] = _reject_write + + response = await client.get("/sessions/root-1/summary") + + assert response.status_code == 200 + + +class TestDeleteSession: + @pytest.mark.anyio + async def test_delete_returns_result(self, client: httpx.AsyncClient) -> None: + result = _sample_result() + fake = _FakeDeletionService(result=result) + _override_delete_service(fake) + + response = await client.delete("/sessions/root-1") + + assert response.status_code == 200 + assert response.json() == { + "root_id": "root-1", + "session_count": 2, + "nodes_deleted": 10, + "relationships_deleted": 5, + "blobs_deleted": 2, + "queue_sessions_cleaned": 2, + } + assert fake.preview_calls == [] # DELETE never previews -- GET does + assert len(fake.apply_calls) == 1 + assert fake.apply_calls[0][0] == "root-1" + + @pytest.mark.anyio + async def test_no_workspace_query_param_needed( + self, client: httpx.AsyncClient + ) -> None: + """The delete route takes no workspace query param at all -- the + session id alone is enough to reach the service.""" + fake = _FakeDeletionService(result=_sample_result()) + _override_delete_service(fake) + + response = await client.delete("/sessions/root-1") + + assert response.status_code == 200 + assert fake.apply_calls == [("root-1", None)] + + @pytest.mark.anyio + async def test_passes_the_authenticated_caller_as_requested_by(self) -> None: + result = _sample_result() + fake = _FakeDeletionService(result=result) + _override_delete_service(fake) + + async with _client_with_scope_state({"contributor_id": "alice"}) as client: + response = await client.delete("/sessions/root-1") + + assert response.status_code == 200 + assert fake.apply_calls == [("root-1", "alice")] + + @pytest.mark.anyio + async def test_unknown_session_returns_404(self, client: httpx.AsyncClient) -> None: + fake = _FakeDeletionService(result=None) + _override_delete_service(fake) + + response = await client.delete("/sessions/does-not-exist") + + assert response.status_code == 404 + + @pytest.mark.anyio + async def test_conflict_when_sessions_still_receiving_data( + self, client: httpx.AsyncClient + ) -> None: + """An undrained graph is a retryable 409: it carries a Retry-After + header and a machine-readable body (reason + pending_sessions + + retry_after_seconds) so the caller can back off and retry.""" + fake = _FakeDeletionService( + apply_error=SessionsPendingError("root-1", ["sub-1"]) + ) + _override_delete_service(fake) + + response = await client.delete("/sessions/root-1") + + assert response.status_code == 409 + retry_after = response.headers.get("Retry-After") + assert retry_after is not None and int(retry_after) > 0 + detail = response.json()["detail"] + assert detail["reason"] == "sessions_pending" + assert detail["pending_sessions"] == ["sub-1"] + assert detail["retry_after_seconds"] == int(retry_after) + + @pytest.mark.anyio + async def test_graph_vanished_race_is_a_plain_409( + self, client: httpx.AsyncClient + ) -> None: + """A non-pending RuntimeError (graph vanished between resolve and + delete) is still a 409, but NOT retryable -- no Retry-After header.""" + fake = _FakeDeletionService( + apply_error=RuntimeError("graph vanished between resolve and delete") + ) + _override_delete_service(fake) + + response = await client.delete("/sessions/root-1") + + assert response.status_code == 409 + assert "Retry-After" not in response.headers + assert "vanished" in response.json()["detail"] + + @pytest.mark.anyio + async def test_ambiguous_session_id_returns_409( + self, client: httpx.AsyncClient + ) -> None: + """A session id found in more than one workspace is a 409, not a 500 + or a silent guess.""" + fake = _FakeDeletionService( + apply_error=AmbiguousSessionError("root-1", ["ws1", "ws2"]) + ) + _override_delete_service(fake) + + response = await client.delete("/sessions/root-1") + + assert response.status_code == 409 + assert "more than one workspace" in response.json()["detail"] + + @pytest.mark.anyio + async def test_refuses_a_caller_who_fails_require_write( + self, client: httpx.AsyncClient + ) -> None: + fake = _FakeDeletionService(result=_sample_result()) + _override_delete_service(fake) + app.dependency_overrides[require_write] = _reject_write + + response = await client.delete("/sessions/root-1") + + assert response.status_code == 403 diff --git a/tests/routers/test_whoami.py b/tests/routers/test_whoami.py new file mode 100644 index 00000000..0bc7c6d8 --- /dev/null +++ b/tests/routers/test_whoami.py @@ -0,0 +1,58 @@ +"""Tests for the GET /whoami endpoint. + +Mirrors the style of ``tests/routers/test_deletion.py``: the ``auth_client`` +fixture (auth middleware applied, a test static key mapped to contributor id +``"owner"``) proves the authenticated case, and the plain ``client`` fixture +(no auth middleware, ``allow_unauthenticated`` in effect for the test suite) +proves the anonymous/no-auth case. See ``tests/conftest.py`` for both +fixtures. +""" + +from __future__ import annotations + +import httpx +import pytest + + +class TestGetWhoamiAuthenticated: + """GET /whoami returns the caller's contributor id for a bearer-token request.""" + + @pytest.mark.anyio + async def test_returns_200(self, auth_client: httpx.AsyncClient) -> None: + response = await auth_client.get( + "/whoami", headers={"Authorization": "Bearer test-secret"} + ) + assert response.status_code == 200 + + @pytest.mark.anyio + async def test_returns_the_authenticated_contributor_id( + self, auth_client: httpx.AsyncClient + ) -> None: + response = await auth_client.get( + "/whoami", headers={"Authorization": "Bearer test-secret"} + ) + assert response.json() == {"contributor_id": "owner"} + + @pytest.mark.anyio + async def test_rejects_missing_bearer_token( + self, auth_client: httpx.AsyncClient + ) -> None: + """Auth is enabled on this client -- no Authorization header is a 401, + not an anonymous whoami.""" + response = await auth_client.get("/whoami") + assert response.status_code == 401 + + +class TestGetWhoamiNoAuth: + """GET /whoami when auth is disabled (allow_unauthenticated) reports a + null/anonymous identity rather than erroring.""" + + @pytest.mark.anyio + async def test_returns_200(self, client: httpx.AsyncClient) -> None: + response = await client.get("/whoami") + assert response.status_code == 200 + + @pytest.mark.anyio + async def test_returns_null_contributor_id(self, client: httpx.AsyncClient) -> None: + response = await client.get("/whoami") + assert response.json() == {"contributor_id": None} diff --git a/tests/test_blob_store.py b/tests/test_blob_store.py index 4a3a31ad..5c6d96a4 100644 --- a/tests/test_blob_store.py +++ b/tests/test_blob_store.py @@ -26,10 +26,8 @@ from unittest.mock import patch import pytest - from context_intelligence_server.blob_store import AsyncDiskBlobStore, BlobStore - # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @@ -321,9 +319,8 @@ async def test_write_is_atomic_no_torn_file_on_failure( with patch( "context_intelligence_server.blob_store.os.replace", side_effect=OSError("simulated replace failure"), - ): - with pytest.raises(OSError): - await store.write(session_id, key, {"v": 1}) + ), pytest.raises(OSError): + await store.write(session_id, key, {"v": 1}) final_path = store.blob_path(session_id, key) # No torn file observable at the final path. @@ -348,3 +345,57 @@ async def test_write_replaces_atomically_on_success( assert final_path.read_text(encoding="utf-8") == '{"v": 1}' # No leftover temp files. assert list(final_path.parent.glob("*.tmp")) == [] + + +# --------------------------------------------------------------------------- +# delete_session +# --------------------------------------------------------------------------- + + +async def test_delete_session_removes_all_blobs(store: AsyncDiskBlobStore) -> None: + """delete_session removes every blob and returns the count.""" + await store.write("sess-del", "k1", {"v": 1}) + await store.write("sess-del", "k2", {"v": 2}) + + removed = await store.delete_session("sess-del") + + assert removed == 2 + assert await store.list("sess-del") == [] + + +async def test_delete_session_missing_is_noop(store: AsyncDiskBlobStore) -> None: + """Deleting a session with no blobs returns 0 and does not raise.""" + assert await store.delete_session("never-existed") == 0 + + +async def test_delete_session_isolates_other_sessions( + store: AsyncDiskBlobStore, +) -> None: + """Deleting one session leaves other sessions' blobs intact.""" + await store.write("sess-a", "k1", {"v": 1}) + await store.write("sess-b", "k1", {"v": 1}) + + await store.delete_session("sess-a") + + assert await store.list("sess-a") == [] + assert await store.list("sess-b") == ["ci-blob://sess-b/k1"] + + +# --------------------------------------------------------------------------- +# size() +# --------------------------------------------------------------------------- + + +async def test_size_returns_byte_size_of_written_blob( + store: AsyncDiskBlobStore, tmp_path: Path +) -> None: + """size() returns the exact on-disk byte size of the JSON file.""" + uri = await store.write("sess-size", "k1", {"v": 1}) + expected = (tmp_path / "sess-size" / "blobs" / "k1.json").stat().st_size + assert await store.size(uri) == expected + assert expected > 0 + + +async def test_size_missing_blob_returns_zero(store: AsyncDiskBlobStore) -> None: + """size() is idempotent-on-missing: returns 0, never raises.""" + assert await store.size("ci-blob://never-existed/missing_key") == 0 diff --git a/tests/test_deletion.py b/tests/test_deletion.py new file mode 100644 index 00000000..841f9ef0 --- /dev/null +++ b/tests/test_deletion.py @@ -0,0 +1,339 @@ +"""Unit tests for DeletionService (A4). + +Uses the in-memory GraphState (services.py) alongside a REAL +AsyncDiskBlobStore and a REAL QueueManager rooted at pytest's tmp_path -- +only Neo4j itself is faked. See tests/neo4j/test_deletion_service.py for the +real-Neo4j proof. +""" + +from __future__ import annotations + +import json +import logging +from pathlib import Path + +import pytest +from context_intelligence_server.blob_store import AsyncDiskBlobStore +from context_intelligence_server.deletion import DeletionService, SessionsPendingError +from context_intelligence_server.queue_manager import QueueManager +from context_intelligence_server.services import GraphState + +pytestmark = pytest.mark.asyncio + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture +def graph() -> GraphState: + return GraphState(workspace="test") + + +@pytest.fixture +def blob_store(tmp_path: Path) -> AsyncDiskBlobStore: + return AsyncDiskBlobStore(tmp_path / "blobs") + + +@pytest.fixture +def queue_manager(tmp_path: Path) -> QueueManager: + return QueueManager(tmp_path / "queues") + + +@pytest.fixture +def service( + graph: GraphState, blob_store: AsyncDiskBlobStore, queue_manager: QueueManager +) -> DeletionService: + return DeletionService(graph, blob_store, queue_manager) + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +async def _drain(queue_manager: QueueManager, session_id: str, raw: bytes) -> None: + """Append one record and commit it -- session ends up with pending_count == 0.""" + await queue_manager.append(session_id, raw) + batch = await queue_manager.read_batch(session_id, max_items=10) + await queue_manager.commit(session_id, batch.end_offset) + + +async def _build_drained_multi_session_graph( + graph: GraphState, + blob_store: AsyncDiskBlobStore, + queue_manager: QueueManager, + *, + root: str, + sub1: str, + sub2: str, + concept: str, +) -> frozenset[str]: + """Seed root -> sub1 -> sub2, with 4 blobs total and a shared SST_CONCEPT node. + + Every session's queue is drained (committed). sub2 additionally carries a + dead-letter record, to prove ``.dead.jsonl`` is also removed by + ``QueueManager.delete_session`` (unlike ``delete_drained``). + + Returns the set of blob URIs written. + """ + blob_root = await blob_store.write(root, "k1", {"v": "root"}) + blob_sub1 = await blob_store.write(sub1, "k1", {"v": "sub1"}) + blob_sub2a = await blob_store.write(sub2, "k1", {"v": "sub2a"}) + blob_sub2b = await blob_store.write(sub2, "k2", {"v": "sub2b"}) + + await graph.upsert_node( + root, + { + "labels": ["Session", "RootSession"], + "started_at": "2026-01-01T00:00:00", + "created_by": "colombod", + }, + ) + await graph.upsert_node( + sub1, + { + "labels": ["Session", "SubSession"], + "started_at": "2026-01-01T00:01:00", + }, + ) + await graph.upsert_edge(root, sub1, {"type": "HAS_SUBSESSION"}) + await graph.upsert_node( + sub2, + { + "labels": ["Session", "SubSession"], + "started_at": "2026-01-01T00:02:00", + }, + ) + await graph.upsert_edge(sub1, sub2, {"type": "HAS_SUBSESSION"}) + await graph.upsert_node(concept, {"labels": ["Agent", "SST_CONCEPT"]}) + await graph.upsert_edge(sub2, concept, {"type": "HAS_AGENT"}) + + for sid in (root, sub1, sub2): + await _drain( + queue_manager, sid, json.dumps({"event": "tool:pre"}).encode("utf-8") + ) + await queue_manager.dead_letter(sub2, b"bad-line", "boom") + + return frozenset({blob_root, blob_sub1, blob_sub2a, blob_sub2b}) + + +# --------------------------------------------------------------------------- +# preview() +# --------------------------------------------------------------------------- + + +async def test_preview_unknown_session_returns_none(service: DeletionService) -> None: + assert await service.preview("does-not-exist") is None + + +async def test_preview_returns_facts_and_mutates_nothing( + service: DeletionService, + graph: GraphState, + blob_store: AsyncDiskBlobStore, + queue_manager: QueueManager, +) -> None: + root, sub1, sub2, concept = "p-root", "p-sub1", "p-sub2", "p-agent" + blob_refs = await _build_drained_multi_session_graph( + graph, + blob_store, + queue_manager, + root=root, + sub1=sub1, + sub2=sub2, + concept=concept, + ) + + preview = await service.preview(sub1) + assert preview is not None + assert preview.root_id == root + assert preview.session_ids == frozenset({root, sub1, sub2}) + assert preview.blob_count == len(blob_refs) == 4 + assert preview.node_count == 4 # root, sub1, sub2, concept (boundary, included) + assert preview.edge_count == 3 # root->sub1, sub1->sub2, sub2->concept + assert preview.subsession_count == 2 + assert preview.created_by == "colombod" + assert preview.deletable is True + assert preview.pending_sessions == [] + + # Nothing mutated: same facts on a second call, graph/blobs/queue untouched. + preview_again = await service.preview(root) + assert preview_again == preview + assert await graph.get_node(root) is not None + assert await graph.get_node(sub2) is not None + assert await blob_store.list(sub2) != [] + assert queue_manager._log_path(sub2).exists() + + +async def test_preview_deletable_false_when_session_has_pending( + service: DeletionService, graph: GraphState, queue_manager: QueueManager +) -> None: + root, sub1 = "pp-root", "pp-sub1" + await graph.upsert_node(root, {"labels": ["Session", "RootSession"]}) + await graph.upsert_node(sub1, {"labels": ["Session", "SubSession"]}) + await graph.upsert_edge(root, sub1, {"type": "HAS_SUBSESSION"}) + + # sub1 has an appended-but-uncommitted record -> pending_count > 0. + await queue_manager.append(sub1, b'{"event": "tool:pre"}') + + preview = await service.preview(root) + assert preview is not None + assert preview.deletable is False + assert preview.pending_sessions == [sub1] + + +# --------------------------------------------------------------------------- +# apply() -- refusal on pending +# --------------------------------------------------------------------------- + + +async def test_apply_refuses_and_deletes_nothing_when_pending( + service: DeletionService, + graph: GraphState, + blob_store: AsyncDiskBlobStore, + queue_manager: QueueManager, +) -> None: + root, sub1 = "ap-root", "ap-sub1" + blob_uri = await blob_store.write(root, "k1", {"v": 1}) + await graph.upsert_node(root, {"labels": ["Session", "RootSession"]}) + await graph.upsert_node(sub1, {"labels": ["Session", "SubSession"]}) + await graph.upsert_edge(root, sub1, {"type": "HAS_SUBSESSION"}) + await queue_manager.append(sub1, b'{"event": "tool:pre"}') # uncommitted + + with pytest.raises(SessionsPendingError, match="pending") as excinfo: + await service.apply(root) + # The retryable refusal names exactly which sessions are still draining. + assert excinfo.value.pending_sessions == [sub1] + assert excinfo.value.root_id == root + + # Nothing deleted: graph, blob, and queue artifacts all survive. + assert await graph.get_node(root) is not None + assert await graph.get_node(sub1) is not None + assert await blob_store.list(root) == [blob_uri] + assert queue_manager._log_path(sub1).exists() + + +# --------------------------------------------------------------------------- +# apply() -- full multi-session deletion +# --------------------------------------------------------------------------- + + +async def test_apply_deletes_graph_blobs_and_queue_for_every_session( + service: DeletionService, + graph: GraphState, + blob_store: AsyncDiskBlobStore, + queue_manager: QueueManager, +) -> None: + root, sub1, sub2, concept = "ad-root", "ad-sub1", "ad-sub2", "ad-agent" + blob_refs = await _build_drained_multi_session_graph( + graph, + blob_store, + queue_manager, + root=root, + sub1=sub1, + sub2=sub2, + concept=concept, + ) + + result = await service.apply(sub1, requested_by="tester") + + assert result is not None + assert result.root_id == root + assert result.session_count == 3 + assert result.nodes_deleted == 3 # root, sub1, sub2 -- concept excluded + assert result.relationships_deleted == 3 # root->sub1, sub1->sub2, sub2->concept + assert result.blobs_deleted == len(blob_refs) == 4 + assert result.queue_sessions_cleaned == 3 + + # Graph: every owned session node gone, shared concept survives. + for sid in (root, sub1, sub2): + assert await graph.get_node(sid) is None + concept_node = await graph.get_node(concept) + assert concept_node is not None + assert "SST_CONCEPT" in concept_node.get("labels", []) + + # Blobs: every graph session's blob dir gone. + for sid in (root, sub1, sub2): + assert await blob_store.list(sid) == [] + + # Queue: log/offset/dead-letter gone for every graph session. + for sid in (root, sub1, sub2): + assert not queue_manager._log_path(sid).exists() + assert not queue_manager._offset_path(sid).exists() + assert not queue_manager._dead_path(sid).exists() + + +async def test_apply_unknown_session_returns_none(service: DeletionService) -> None: + assert await service.apply("does-not-exist") is None + + +async def test_apply_logs_the_deletion( + service: DeletionService, + graph: GraphState, + blob_store: AsyncDiskBlobStore, + queue_manager: QueueManager, + caplog: pytest.LogCaptureFixture, +) -> None: + root, sub1, sub2, concept = "al-root", "al-sub1", "al-sub2", "al-agent" + await _build_drained_multi_session_graph( + graph, + blob_store, + queue_manager, + root=root, + sub1=sub1, + sub2=sub2, + concept=concept, + ) + + with caplog.at_level(logging.INFO, logger="context_intelligence_server.deletion"): + result = await service.apply(root, requested_by="colombod") + + assert result is not None + assert any( + r.levelno == logging.INFO + and "session_deletion_applied" in r.getMessage() + and f"root_id={root}" in r.getMessage() + and "requested_by=colombod" in r.getMessage() + and getattr(r, "session_id", None) == root + for r in caplog.records + ), "the applied deletion must be logged once at INFO with root_id/requested_by" + + +async def test_preview_and_apply_count_every_blob_the_store_holds( + service: DeletionService, + graph: GraphState, + blob_store: AsyncDiskBlobStore, + queue_manager: QueueManager, +) -> None: + """Blob counting goes through the blob store, not through node data. + + A blob written for a graph session is counted and deleted even though no + node property points at it -- the blob store already knows which blobs + belong to a session (``BlobStore.list``), so there is nothing left to + "reconcile" against a separate, node-derived count. + """ + root, sub1, sub2, concept = "rm-root", "rm-sub1", "rm-sub2", "rm-agent" + blob_refs = await _build_drained_multi_session_graph( + graph, + blob_store, + queue_manager, + root=root, + sub1=sub1, + sub2=sub2, + concept=concept, + ) + # A blob with no corresponding node property -- still a real file the + # blob store holds for this session. + await blob_store.write(sub2, "extra", {"v": "extra"}) + + preview = await service.preview(root) + assert preview is not None + assert preview.blob_count == len(blob_refs) + 1 + + result = await service.apply(root, requested_by="tester") + + assert result is not None + assert result.blobs_deleted == len(blob_refs) + 1 + for sid in (root, sub1, sub2): + assert await blob_store.list(sid) == [] diff --git a/tests/test_graph_store.py b/tests/test_graph_store.py index 0b088125..3925b4b9 100644 --- a/tests/test_graph_store.py +++ b/tests/test_graph_store.py @@ -10,7 +10,6 @@ from context_intelligence_server.graph_store import GraphStore, QueryableStore - # --------------------------------------------------------------------------- # Minimal conforming implementations for isinstance checks # --------------------------------------------------------------------------- @@ -58,6 +57,12 @@ async def find_delegation_by_sub_session( ) -> dict[str, Any] | None: return None + async def resolve_session_graph(self, session_id: str) -> Any: + return None + + async def delete_session_graph(self, session_id: str) -> Any: + return None + async def flush(self) -> None: pass @@ -136,6 +141,12 @@ async def find_delegation_by_sub_session( ) -> dict[str, Any] | None: return None + async def resolve_session_graph(self, session_id: str) -> Any: + return None + + async def delete_session_graph(self, session_id: str) -> Any: + return None + async def flush(self) -> None: pass @@ -243,6 +254,7 @@ def test_queryable_store_exported(): def test_no_graph_forest_name_references(): """Verify graph_forest_name does not appear anywhere in graph_store.py.""" import inspect + import context_intelligence_server.graph_store as m source = inspect.getsource(m) diff --git a/tests/test_queue_manager.py b/tests/test_queue_manager.py index e84c1646..2358739c 100644 --- a/tests/test_queue_manager.py +++ b/tests/test_queue_manager.py @@ -349,6 +349,88 @@ async def test_delete_drained_removes_log_and_offset_keeps_dead(tmp_path) -> Non assert len(await qm.read_dead_letters("s")) == 1 +# --------------------------------------------------------------------------- +# pending_count / delete_session (A3): drain precondition + full queue delete +# --------------------------------------------------------------------------- + + +async def test_pending_count_reflects_appended_uncommitted_records(qm): + await qm.append("s1", b"a") + await qm.append("s1", b"b") + assert await qm.pending_count("s1") == 2 + + +async def test_pending_count_zero_after_full_commit(qm): + await qm.append("s1", b"a") + await qm.append("s1", b"b") + batch = await qm.read_batch("s1", max_items=10) + await qm.commit("s1", batch.end_offset) + assert await qm.pending_count("s1") == 0 + + +async def test_pending_count_ignores_torn_trailing_line(qm, tmp_path): + log = tmp_path / "queues" / "s1.log" + log.parent.mkdir(parents=True, exist_ok=True) + log.write_bytes(b"complete\nTORN_PARTIAL") + assert await qm.pending_count("s1") == 1 # torn tail never counts + + +async def test_pending_count_zero_for_unknown_session(qm): + assert await qm.pending_count("never-written") == 0 + + +@pytest.mark.parametrize("bad_id", ["", "a/b", "a\\b", "a\x00b"]) +async def test_pending_count_rejects_unsafe_session_id(qm, bad_id): + with pytest.raises(ValueError): + await qm.pending_count(bad_id) + + +async def test_delete_session_removes_log_offset_and_dead_letters(qm, tmp_path): + await qm.append("s", b"line") + await qm.commit("s", 5) + await qm.dead_letter("s", b"bad\n", "boom") + + removed = await qm.delete_session("s") + + assert removed is True + queues_dir = tmp_path / "queues" + assert not (queues_dir / "s.log").exists() + assert not (queues_dir / "s.offset").exists() + assert not (queues_dir / "s.dead.jsonl").exists() # unlike delete_drained + + +async def test_delete_session_refuses_when_pending(qm, tmp_path): + await qm.append("s", b"line") # never committed -> pending + + with pytest.raises(RuntimeError): + await qm.delete_session("s") + + # Nothing was partially deleted. + queues_dir = tmp_path / "queues" + assert (queues_dir / "s.log").exists() + assert await qm.pending_count("s") == 1 + + +async def test_delete_session_idempotent_on_missing_session(qm): + assert await qm.delete_session("never-written") is False + + +async def test_delete_session_removes_dead_letter_only_session(qm, tmp_path): + """A session with no pending log (only dead letters) can be deleted.""" + await qm.dead_letter("s-dead", b"poison", error="boom") + + removed = await qm.delete_session("s-dead") + + assert removed is True + assert not (tmp_path / "queues" / "s-dead.dead.jsonl").exists() + + +@pytest.mark.parametrize("bad_id", ["", "a/b", "a\\b", "a\x00b"]) +async def test_delete_session_rejects_unsafe_session_id(qm, bad_id): + with pytest.raises(ValueError): + await qm.delete_session(bad_id) + + async def test_derive_all_stats_counts_pending_and_dead(qm): # s1: two complete pending (uncommitted) lines, no dead letters. await qm.append("s1", b"a") diff --git a/tests/test_services.py b/tests/test_services.py index 93c6d87c..35d256e9 100644 --- a/tests/test_services.py +++ b/tests/test_services.py @@ -2,16 +2,17 @@ from __future__ import annotations -import pytest from unittest.mock import AsyncMock, patch +import pytest +from context_intelligence_server.blob_store import AsyncDiskBlobStore from context_intelligence_server.services import ( GraphState, HookConfig, HookStateService, + total_blob_size, ) - # --------------------------------------------------------------------------- # HookConfig tests # --------------------------------------------------------------------------- @@ -777,3 +778,253 @@ def test_created_by_propagated_to_custom_graph_store(self) -> None: store = GraphState() svc = HookStateService(workspace="/ws", graph_store=store, created_by="carol") assert svc.graph.created_by == "carol" + + +# --------------------------------------------------------------------------- +# GraphState.resolve_session_graph tests +# --------------------------------------------------------------------------- + + +async def _build_graph(state: GraphState) -> None: + """Seed a root + 2 subsessions + 1 fork, with blobs on several sessions. + + Tree shape: root -[HAS_SUBSESSION]-> sub1 -[HAS_SUBSESSION]-> sub2 + root -[FORKED]-> fork1 + + Blob-bearing event nodes attached at three different graph sessions + (root, sub2, fork1), plus a shared Agent (:SST_CONCEPT) reached from + fork1's Delegation, with an out-of-graph node hanging off the Agent to + prove traversal stops there. + """ + state.created_by = "colombod" + await state.upsert_node( + "fam-root", + { + "labels": ["Session", "RootSession"], + "started_at": "2026-01-01T00:00:00+00:00", + }, + ) + await state.upsert_node( + "fam-sub1", + { + "labels": ["Session", "SubSession"], + "parent_id": "fam-root", + "started_at": "2026-01-01T00:01:00+00:00", + }, + ) + await state.upsert_edge("fam-root", "fam-sub1", {"type": "HAS_SUBSESSION"}) + await state.upsert_node( + "fam-sub2", + { + "labels": ["Session", "SubSession"], + "parent_id": "fam-sub1", + "started_at": "2026-01-01T00:02:00+00:00", + "last_updated": "2026-01-01T00:05:00+00:00", + }, + ) + await state.upsert_edge("fam-sub1", "fam-sub2", {"type": "HAS_SUBSESSION"}) + await state.upsert_node( + "fam-fork1", + { + "labels": ["Session", "ForkedSession"], + "parent_id": "fam-root", + "started_at": "2026-01-01T00:03:00+00:00", + }, + ) + await state.upsert_edge("fam-root", "fam-fork1", {"type": "FORKED"}) + + # Blob-bearing nodes across several graph sessions. + await state.upsert_node( + "fam-root::orch::1", + { + "labels": ["OrchestratorRun", "SST_EVENT"], + "raw": {"$blob_ref": "ci-blob://fam-root/orch1"}, + }, + ) + await state.upsert_edge("fam-root", "fam-root::orch::1", {"type": "HAS_EXECUTION"}) + + await state.upsert_node( + "fam-sub2::tool::1", + { + "labels": ["ToolCall", "SST_EVENT"], + "result": {"$blob_ref": "ci-blob://fam-sub2/tool1"}, + }, + ) + await state.upsert_edge("fam-sub2", "fam-sub2::tool::1", {"type": "HAS_TOOL_CALL"}) + + await state.upsert_node( + "fam-fork1::delegation::1", + { + "labels": ["Delegation", "SST_EVENT"], + "messages": {"$blob_ref": "ci-blob://fam-fork1/del1"}, + }, + ) + await state.upsert_edge( + "fam-fork1", "fam-fork1::delegation::1", {"type": "TRIGGERED"} + ) + + # Shared concept node -- traversal must stop here, not continue past it. + await state.upsert_node("agent-shared", {"labels": ["Agent", "SST_CONCEPT"]}) + await state.upsert_edge( + "fam-fork1::delegation::1", "agent-shared", {"type": "HAS_AGENT"} + ) + await state.upsert_node( + "other-session-xyz::leak", + { + "labels": ["ToolCall", "SST_EVENT"], + "raw": {"$blob_ref": "ci-blob://other-session-xyz/leak"}, + }, + ) + await state.upsert_edge( + "agent-shared", "other-session-xyz::leak", {"type": "SOME_EDGE"} + ) + + +class TestGraphStateResolveSessionGraph: + """GraphState.resolve_session_graph -- in-memory parity with Neo4jGraphStore.""" + + async def test_returns_none_for_unknown_session(self) -> None: + state = GraphState() + assert await state.resolve_session_graph("does-not-exist") is None + + async def test_sub_session_and_root_resolve_identical_graph(self) -> None: + state = GraphState() + await _build_graph(state) + + from_root = await state.resolve_session_graph("fam-root") + from_sub = await state.resolve_session_graph("fam-sub2") + from_fork = await state.resolve_session_graph("fam-fork1") + + assert from_root is not None + assert from_root.root_id == "fam-root" + assert from_root.session_ids == frozenset( + {"fam-root", "fam-sub1", "fam-sub2", "fam-fork1"} + ) + assert from_sub is not None + assert from_sub.session_ids == from_root.session_ids + assert from_sub.root_id == from_root.root_id + assert from_fork is not None + assert from_fork.session_ids == from_root.session_ids + + async def test_subsession_count_excludes_root(self) -> None: + state = GraphState() + await _build_graph(state) + graph = await state.resolve_session_graph("fam-root") + assert graph is not None + assert graph.subsession_count == 3 + + async def test_node_and_edge_counts_exclude_past_concept_boundary(self) -> None: + state = GraphState() + await _build_graph(state) + graph = await state.resolve_session_graph("fam-root") + assert graph is not None + # 4 sessions + orch + tool + delegation + agent (boundary, included) = 8 + assert graph.node_count == 8 + # root->sub1, sub1->sub2, root->fork1, root->orch, sub2->tool, + # fork1->delegation, delegation->agent = 7 (agent->leak excluded) + assert graph.edge_count == 7 + + async def test_created_by_and_started_at_from_root(self) -> None: + state = GraphState() + await _build_graph(state) + graph = await state.resolve_session_graph("fam-sub2") + assert graph is not None + assert graph.created_by == "colombod" + assert graph.started_at is not None + assert graph.started_at.isoformat() == "2026-01-01T00:00:00+00:00" + + async def test_last_change_is_max_across_graph_not_just_root(self) -> None: + state = GraphState() + await _build_graph(state) + graph = await state.resolve_session_graph("fam-root") + assert graph is not None + # fam-sub2's last_updated (00:05) is later than root's started_at (00:00) + assert graph.last_change is not None + assert graph.last_change.isoformat() == "2026-01-01T00:05:00+00:00" + + +# --------------------------------------------------------------------------- +# GraphState.delete_session_graph tests +# --------------------------------------------------------------------------- + + +class TestGraphStateDeleteSessionGraph: + """GraphState.delete_session_graph -- in-memory parity with Neo4jGraphStore.""" + + async def test_returns_none_for_unknown_session(self) -> None: + state = GraphState() + assert await state.delete_session_graph("does-not-exist") is None + + async def test_deletes_via_sub_session_id(self) -> None: + state = GraphState() + await _build_graph(state) + + result = await state.delete_session_graph("fam-sub2") + assert result is not None + assert result.root_id == "fam-root" + + for sid in ("fam-root", "fam-sub1", "fam-sub2", "fam-fork1"): + assert await state.get_node(sid) is None + assert await state.get_node("fam-root::orch::1") is None + assert await state.get_node("fam-sub2::tool::1") is None + assert await state.get_node("fam-fork1::delegation::1") is None + + async def test_shared_concept_node_survives(self) -> None: + state = GraphState() + await _build_graph(state) + await state.delete_session_graph("fam-root") + + assert await state.get_node("agent-shared") is not None + + async def test_unrelated_session_reachable_only_via_concept_is_untouched( + self, + ) -> None: + state = GraphState() + await _build_graph(state) + await state.delete_session_graph("fam-root") + + assert await state.get_node("other-session-xyz::leak") is not None + assert await state.get_edge("agent-shared", "other-session-xyz::leak") is not None + + async def test_counts_match_deleted_nodes_and_edges(self) -> None: + state = GraphState() + await _build_graph(state) + result = await state.delete_session_graph("fam-root") + assert result is not None + # Owned = graph node_count (8) minus the boundary concept node (1) = 7. + assert result.nodes_deleted == 7 + # Every edge except agent-shared -> leak (both endpoints not owned). + assert result.relationships_deleted == 7 + + +# --------------------------------------------------------------------------- +# total_blob_size() service helper +# --------------------------------------------------------------------------- + + +class TestTotalBlobSize: + """total_blob_size() composes BlobStore.size() over a list of blob URIs.""" + + async def test_sums_sizes_across_multiple_refs(self, tmp_path) -> None: + blob_store = AsyncDiskBlobStore(root=tmp_path) + uri_a = await blob_store.write("sess-a", "k1", {"v": 1}) + uri_b = await blob_store.write("sess-b", "k1", {"v": [1, 2, 3]}) + + expected = await blob_store.size(uri_a) + await blob_store.size(uri_b) + total = await total_blob_size(blob_store, [uri_a, uri_b]) + assert total == expected + assert total > 0 + + async def test_missing_refs_contribute_zero(self, tmp_path) -> None: + blob_store = AsyncDiskBlobStore(root=tmp_path) + uri = await blob_store.write("sess-a", "k1", {"v": 1}) + real_size = await blob_store.size(uri) + + total = await total_blob_size( + blob_store, [uri, "ci-blob://never-existed/missing"] + ) + assert total == real_size + + async def test_empty_refs_returns_zero(self, tmp_path) -> None: + blob_store = AsyncDiskBlobStore(root=tmp_path) + assert await total_blob_size(blob_store, []) == 0 diff --git a/uv.lock b/uv.lock index 68d8aef5..fd5b364c 100644 --- a/uv.lock +++ b/uv.lock @@ -233,7 +233,7 @@ wheels = [ [[package]] name = "context-intelligence-server" -version = "6.7.1" +version = "6.9.0" source = { editable = "." } dependencies = [ { name = "aiofiles" },