From caf89acbc5163c29cc48a4125195253504cc62e7 Mon Sep 17 00:00:00 2001 From: colombod Date: Wed, 26 Aug 2026 13:48:32 +0000 Subject: [PATCH] feat(admin): orphaned-blob reclaim GC + blob-carrier allowlist MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add POST /admin/blobs/reclaim, a protocol-only garbage collector for blob-store artifacts no longer referenced by the graph. Enumeration runs through BlobStore.scan() and deletion through the fenced BlobStore.delete(uri, if_unmodified=ref) compare-and-delete -- no filesystem path, glob, or os.unlink, and it never reaches the queue / identity / lease stores or graph data. Safety gates: a graph-wide reference scan over the blob-carrier allowlist, a not-live / durable is_fully_drained session gate, a hard min_age_minutes floor (>= 15), a destructive-apply single-flight (409 on overlap), a required max_delete blast-radius cap, and dry_run=true by default. One structured audit line per delete records only the ci-blob:// URI, never blob contents. Fold in the blob-carrier allowlist (BLOB_REF_CARRIER_PROPERTIES in blob_processor), reclaim's only consumer: the single source of truth for which graph properties may carry a ci-blob:// reference, validated at import and enforced at the mint site so an unregistered carrier fails loud instead of becoming a silent reclaim-GC hole. The reference-scan Cypher is generated from this tuple, so the two can never drift. Version 7.4.0. 🤖 Generated with [Amplifier](https://github.com/microsoft/amplifier) Co-Authored-By: Amplifier --- CHANGELOG.md | 23 + context_intelligence_server/blob_processor.py | 97 +++++ context_intelligence_server/routers/admin.py | 408 +++++++++++++++++- pyproject.toml | 2 +- tests/neo4j/test_blob_reclaim_e2e.py | 144 +++++++ tests/test_blob_carrier_allowlist.py | 160 +++++++ tests/test_blob_reclaim_endpoint.py | 187 ++++++++ uv.lock | 2 +- 8 files changed, 1019 insertions(+), 4 deletions(-) create mode 100644 tests/neo4j/test_blob_reclaim_e2e.py create mode 100644 tests/test_blob_carrier_allowlist.py create mode 100644 tests/test_blob_reclaim_endpoint.py diff --git a/CHANGELOG.md b/CHANGELOG.md index daf3532a..4f668165 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,29 @@ All notable changes to the Context Intelligence Server are recorded here. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [7.4.0] + +### Added + +- **Orphaned-blob garbage collection.** New `POST /admin/blobs/reclaim` + endpoint reclaims blob-store artifacts no longer referenced by the graph. It + is protocol-only — enumeration via `BlobStore.scan()`, deletion via the + fenced `BlobStore.delete(uri, if_unmodified=ref)` compare-and-delete — so it + never touches a filesystem path, glob, or `os.unlink`, and never reaches the + queue / identity / lease stores or graph data. Safety gates: a graph-wide + reference scan over the blob-carrier allowlist, a not-live / durable + `is_fully_drained` session gate, a hard `min_age_minutes` floor (>= 15), a + destructive-apply single-flight (409 on overlap), a required `max_delete` + blast-radius cap, and **`dry_run=true` by default** (a preview that deletes + nothing). One structured audit line per delete; blob contents are never + logged, only the `ci-blob://` URI. +- **Blob-carrier allowlist** (`BLOB_REF_CARRIER_PROPERTIES` in + `blob_processor`) — the single source of truth for which graph properties may + carry a `ci-blob://` reference, validated at import and enforced at the mint + site (`assert_carrier_registered`) so an unregistered carrier fails loud + rather than becoming a silent reclaim-GC hole. The reclaim reference-scan + Cypher is generated directly from this tuple, so the two can never drift. + ## [7.3.0] ### Added diff --git a/context_intelligence_server/blob_processor.py b/context_intelligence_server/blob_processor.py index 5be7e33c..9a39ae41 100644 --- a/context_intelligence_server/blob_processor.py +++ b/context_intelligence_server/blob_processor.py @@ -10,6 +10,7 @@ from __future__ import annotations import logging +import re from typing import TYPE_CHECKING, Any if TYPE_CHECKING: @@ -27,6 +28,92 @@ ) +# --------------------------------------------------------------------------- +# Blob-ref carrier allowlist -- single source of truth for which property +# names may carry a ci-blob:// URI. +# --------------------------------------------------------------------------- +# +# Every ``ci-blob://`` URI minted below (``process_event_data``) is written +# into ``data``, which ``DefaultHandler`` always persists wholesale as the +# JSON-serialized ``data`` property on the Event node +# (handlers/data_layer_1/default.py). The blob-reclaim reference scan +# (``routers.admin._scan_referenced_uris``) enumerates every ``ci-blob://`` +# reference anywhere in the graph by walking a FIXED allowlist of node +# properties -- never an all-property/all-node scan (this codebase has scar +# tissue from a 1.3M-node AllNodesScan stall). A blob whose reference lives +# on a node property the scan doesn't know about is invisible to it and can +# be deleted as a false orphan. +# +# BLOB_REF_CARRIER_PROPERTIES is THE single source of truth for that +# allowlist, imported by ``routers.admin`` to build the scan's Cypher +# directly from this tuple (so the query text can never drift from it) and +# checked here, at the mint site, via :func:`assert_carrier_registered`. +# +# Adding a new carrier (a future field-lifter/enricher that promotes a +# blob-ref-shaped value onto a new node property) means adding its name +# here. Forgetting to is now a fail-closed error, not a silent GC hole. +BLOB_REF_CARRIER_PROPERTIES: tuple[str, ...] = ( + "data", + "tool_input", + "prompt", + "response", +) + +# Defensive validation, run once at import time: every carrier name must be +# a legal Cypher property identifier, because routers.admin interpolates +# these names directly into a Cypher query string. Guards against a future +# careless addition (e.g. containing a space or backtick) turning into a +# broken or injectable query rather than a loud, immediate import error. +_VALID_CARRIER_NAME_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") + + +def _validate_carrier_names(names: tuple[str, ...]) -> None: + for name in names: + if not _VALID_CARRIER_NAME_RE.match(name): + raise ValueError( + f"BLOB_REF_CARRIER_PROPERTIES entry {name!r} is not a valid " + "Cypher property identifier -- refusing to load (this tuple " + "is interpolated directly into a Cypher query by " + "routers.admin._scan_referenced_uris)" + ) + + +_validate_carrier_names(BLOB_REF_CARRIER_PROPERTIES) + + +class UnregisteredBlobCarrierError(RuntimeError): + """A ``ci-blob://`` reference is destined for a node property that is + not in :data:`BLOB_REF_CARRIER_PROPERTIES`. + + This converts a silent reclaim-GC hole + (a live blob deleted as an orphan because its carrier property was never + added to the allowlist) into a loud, immediate failure at the point the + omission is introduced -- not after a live blob is gone. + """ + + +def assert_carrier_registered(property_name: str) -> None: + """Fail loud if *property_name* is not a registered blob-ref carrier. + + Cheap (single tuple-membership check) and safe to call on every + ``process_event_data`` invocation. Raises + :class:`UnregisteredBlobCarrierError` -- deliberately NOT caught by the + per-field ``except Exception`` below, so it propagates out of + ``process_event_data``, through ``pipeline.process_event``'s outer + handler (which logs and re-raises), and the event is dead-lettered + instead of silently minting an unprotected blob reference. + """ + if property_name not in BLOB_REF_CARRIER_PROPERTIES: + raise UnregisteredBlobCarrierError( + f"ci-blob:// reference destined for node property {property_name!r} " + f"is not in BLOB_REF_CARRIER_PROPERTIES {BLOB_REF_CARRIER_PROPERTIES!r} " + "-- the blob-reclaim scan (context_intelligence_server.routers.admin) " + "will not see refs stored there and could delete this blob as an " + f"orphan. Add {property_name!r} to BLOB_REF_CARRIER_PROPERTIES " + "(context_intelligence_server/blob_processor.py) before shipping." + ) + + # --------------------------------------------------------------------------- # Internal helpers # --------------------------------------------------------------------------- @@ -88,6 +175,16 @@ async def process_event_data( """ _lift_raw_fields(data) + # Every ci-blob:// URI minted below lands in + # `data`, which DefaultHandler always persists wholesale onto the Event + # node's "data" property. Fail loud, BEFORE any blob is written, if that + # destination is ever missing from the allowlist the reclaim scan reads + # (see BLOB_REF_CARRIER_PROPERTIES above). Deliberately outside the + # per-field try/except below so it is never downgraded to a swallowed + # $blob_error -- it propagates out of process_event_data and dead-letters + # the event instead of silently minting an unprotected blob reference. + assert_carrier_registered("data") + for field_name in BLOB_FIELDS: value = data.get(field_name) if value is None: diff --git a/context_intelligence_server/routers/admin.py b/context_intelligence_server/routers/admin.py index 4606a303..6b791c6b 100644 --- a/context_intelligence_server/routers/admin.py +++ b/context_intelligence_server/routers/admin.py @@ -33,13 +33,21 @@ from __future__ import annotations +import json import logging import re +import time +from typing import Any from fastapi import APIRouter, Depends, HTTPException, Request -from pydantic import BaseModel, field_validator +from neo4j import READ_ACCESS, WRITE_ACCESS +from pydantic import BaseModel, Field, field_validator -from context_intelligence_server.config import _ALL_ZEROS_GUID, _GUID_RE +from context_intelligence_server.blob_processor import ( + BLOB_REF_CARRIER_PROPERTIES as _BLOB_REF_CARRIER_PROPERTIES, +) +from context_intelligence_server.blob_store import BlobReference, create_blob_store +from context_intelligence_server.config import _ALL_ZEROS_GUID, _GUID_RE, get_settings from context_intelligence_server.identity_store import IdentityStore # --------------------------------------------------------------------------- @@ -53,6 +61,28 @@ # (non-empty, non-whitespace, sane upper bound for an identifier string). _MAX_CONTRIBUTOR_LEN = 256 +# --------------------------------------------------------------------------- +# Blob-reclaim constants. +# --------------------------------------------------------------------------- + +# Hard mtime-floor safety net, defense-in-depth behind +# the durable undrained-queue gate. min_age_minutes below this is rejected +# (422) rather than silently raised -- a caller passing 0 must not be able to +# disable the age gate entirely. +_MIN_AGE_FLOOR_MINUTES = 15 + +# "sample" is bounded to keep the response small; totals (orphans_found, +# reclaimable_bytes) remain authoritative even when the sample is truncated. +_MAX_SAMPLE = 50 + +# Single-flight guard for the DESTRUCTIVE reclaim apply. Two concurrent applies +# would each honour max_delete independently, so together they could delete +# twice the operator's intended blast radius; a second overlapping apply is +# refused (409) rather than admitted. Per-process only. Flipped False->True with +# no await between the check and the set, so the check-and-set is atomic under +# asyncio. Dry-run never takes it -- only the delete phase is serialized. +_reclaim_apply_inflight = False + # --------------------------------------------------------------------------- # Module-level audit logger # --------------------------------------------------------------------------- @@ -309,6 +339,275 @@ def _audit_delete(request: Request, *, target: str) -> None: ) +def _audit_blob_reclaim_delete(request: Request, *, uri: str) -> None: + """Emit one structured audit log line per successfully-deleted blob. + + NEVER logs blob contents -- only the ``ci-blob://`` URI is recorded. + """ + logger.info( + "admin.audit action=blob_reclaim target=%s who=%s", + uri, + _admin_who(request), + ) + + +# --------------------------------------------------------------------------- +# Blob reclaim -- orphaned-blob GC. +# --------------------------------------------------------------------------- + + +def _access_mode_const(mode: str) -> str: + """Map the configured query access-mode string ("READ"/"WRITE") to the + driver's access-mode constant. + + Deliberately duplicated (not imported) from ``main._neo4j_access_const``: + importing from ``main`` here would create a circular import (``main`` + already imports ``routers.admin`` at module load time). Two lines of + duplication is cheaper than that coupling. + """ + return READ_ACCESS if mode == "READ" else WRITE_ACCESS + + +def _collect_blob_refs(obj: Any, out: set[str]) -> None: + """Recursively walk a decoded JSON value collecting ``$blob_ref`` URIs. + + This is STRUCTURAL extraction + over the parsed object, never a regex over the serialized string. A + regex anchored on ``ci-blob://`` truncates at the first unescaped + special character (e.g. a literal ``"`` in a session_id, which + ``queue_manager._validate_session_id`` explicitly permits -- it only + rejects ``/ \\ \\0``), silently misclassifying a genuinely-referenced + blob as orphan. ``json.loads`` has already resolved all escaping by the + time this function runs, so any character in a URI (quotes, non-ASCII) + is handled correctly -- there is no APOC path and no regex path. + """ + if isinstance(obj, dict): + ref = obj.get("$blob_ref") + if isinstance(ref, str): + out.add(ref) + for v in obj.values(): + _collect_blob_refs(v, out) + elif isinstance(obj, list): + for item in obj: + _collect_blob_refs(item, out) + + +# Fallback extraction for carrier values that are NOT valid JSON (a bare +# string property, e.g. a plain-string tool_input/prompt that itself +# contains a ci-blob:// URI rather than the {"$blob_ref": "..."} wrapper). +# Safe here -- unlike a regex over a JSON-*serialized* string (see the +# docstring above) -- because these values are already fully-decoded Neo4j +# property strings with no JSON escaping left to trip over. +_BARE_BLOB_URI_RE = re.compile(r'ci-blob://[^"\s]+') + + +def _extract_blob_refs_from_value(val: str, out: set[str]) -> None: + """Extract every ``ci-blob://`` URI referenced by one carrier property value. + + Two extraction paths, unioned into *out*: + + 1. **Structural (preferred):** ``json.loads(val)`` then recurse with + :func:`_collect_blob_refs`. Handles the ``{"$blob_ref": "..."}`` + JSON-string carriers -- ``neo4j_store._sanitize_properties`` + JSON-serializes any dict/list property value on write, so + ``Event.data``, ``ToolCall.tool_input``, ``Prompt.prompt``, and + ``OrchestratorRun.response`` all round-trip through this path when + they hold a dict/list. + 2. **Regex fallback (only on JSON-parse failure):** a plain-string + carrier is written through VERBATIM (``_sanitize_properties`` only + JSON-serializes dict/list values -- a bare string is stored as-is), + so it is never valid JSON and always lands here. Extracts every bare + ``ci-blob://[^"\\s]+`` token directly from the decoded string -- + covers a lifted ``*.tool_input``/``*.prompt`` property that is a + plain string mentioning a blob URI. + + A value that is neither valid JSON nor contains a bare token contributes + nothing -- this can only ever fail to positively assert a reference, + never falsely assert one, matching the conservative-skip contract of + :func:`_scan_referenced_uris`. + """ + try: + obj = json.loads(val) + except (TypeError, ValueError): + out.update(_BARE_BLOB_URI_RE.findall(val)) + return + _collect_blob_refs(obj, out) + + +def _carrier_scan_clause(prop: str) -> str: + """Build one ``UNION ALL`` branch of the reclaim reference-scan query + for carrier property *prop*. + + ``data`` is special-cased: ``Event.data`` is always a JSON string + (``DefaultHandler`` writes ``json.dumps(data)`` -- see + ``handlers/data_layer_1/default.py``) and the scan for it is scoped to + ``:Event``, matching this carrier's scope in the pre-hardening scan. Every + other registered carrier is an unrestricted-label match with + ``toString()``, since the property may be lifted onto any node type as a + dict, list, or bare string. + """ + if prop == "data": + return "MATCH (n:Event) WHERE n.data CONTAINS 'ci-blob://' RETURN n.data AS val" + return ( + f"MATCH (n) WHERE n.{prop} IS NOT NULL " + f"AND toString(n.{prop}) CONTAINS 'ci-blob://' " + f"RETURN toString(n.{prop}) AS val" + ) + + +# Generated FROM _BLOB_REF_CARRIER_PROPERTIES (imported from +# blob_processor.BLOB_REF_CARRIER_PROPERTIES) -- not hand-duplicated -- so the +# query text can never drift from the allowlist. +_BLOB_REF_SCAN_QUERY = " UNION ALL ".join( + _carrier_scan_clause(prop) for prop in _BLOB_REF_CARRIER_PROPERTIES +) + + +async def _scan_referenced_uris(request: Request) -> set[str]: + """Enumerate every ``ci-blob://`` URI referenced anywhere in the graph. + + Deliberately GLOBAL, never workspace-filtered: blobs are session_id-scoped + while nodes are (node_id, workspace)-scoped, so a per-workspace scan could + delete another workspace's live data. + + The referenced set is computed GRAPH-WIDE over the known ``ci-blob://`` + carrier properties (:data:`_BLOB_REF_CARRIER_PROPERTIES` -- ``data``, + ``tool_input``, ``prompt``, ``response``). This makes the scan correct BY + CONSTRUCTION -- a strict superset of an ``Event.data``-only scan -- rather + than resting on the (empirically true today, but unenforced) + pipeline-ordering invariant that ``DefaultHandler`` always persists every + ref onto ``Event.data`` before any field-lifter/enricher can strip or + promote it elsewhere. Widening the scan can only ever *protect* more blobs, + never delete more. + + Query shape (performance-critical): a ``UNION ALL`` of single-property + predicates, each touching exactly ONE property per row (``data`` restricted + to ``:Event``; the others unrestricted across labels since + ToolCall/Prompt/OrchestratorRun are ordinary nodes). This avoids a + pathological ``MATCH (n) ... [k IN keys(n) WHERE toString(n[k]) ...]`` + all-property-all-node walk, which would toString() every key of every node + in the graph. ``UNION ALL`` (not plain ``UNION``) is deliberate: plain + ``UNION``'s implicit DISTINCT would force Neo4j to materialize and dedupe + every row before returning the first, defeating streaming; the Python + ``set`` below already dedupes, so ``UNION ALL`` costs nothing and preserves + the stream. + + No APOC. Each returned value is extracted via + :func:`_extract_blob_refs_from_value`. A malformed/unparseable value is + skipped conservatively -- it can never positively assert an orphan, only + fail to positively assert a reference. + """ + driver = request.app.state.neo4j_query_driver + access_mode = _access_mode_const(request.app.state.neo4j_query_access_mode) + referenced: set[str] = set() + async with driver.session(default_access_mode=access_mode) as session: + # Graph-wide, per-property UNION ALL -- see the docstring above for + # why this shape (not an all-property walk, not plain UNION). The + # query text is generated from _BLOB_REF_CARRIER_PROPERTIES + # (_BLOB_REF_SCAN_QUERY, module level) -- not hand-duplicated here -- + # so it can never drift from the allowlist. + result = await session.run(_BLOB_REF_SCAN_QUERY) + async for record in result: + val = record["val"] + if not isinstance(val, str): + # toString() on a non-null value is always a str; this guards + # conservatively against an unexpected driver type mapping. + continue + _extract_blob_refs_from_value(val, referenced) + return referenced + + +async def _select_orphans(request: Request, *, min_age_minutes: int) -> dict[str, Any]: + """The ONE selection path shared by dry-run and apply. + + Returns a dict with every response field EXCEPT ``dry_run``/``sample``/ + ``rescanned``/``deleted``/``deleted_bytes`` (the caller fills those in), + plus a ``candidates`` key (list[BlobReference], sorted by uri for + deterministic sampling/capping) that the caller pops before returning the + response and uses to actually delete in apply mode. + + Every blob is reached only through the BlobStore protocol -- ``scan()`` + streams a ``BlobReference`` (uri + size + last_modified) per blob; no + filesystem path or on-disk layout crosses into this router. + + Safety gates applied to every blob not in the referenced set: + 1. Undrained-queue gate (primary, durable): skipped when the + session has a live worker (``registry.active_sessions()``) OR its + queue is not fully drained (``QueueManager.is_fully_drained``, + durable across restarts). Counted as ``skipped_pending_session``. + 2. Age floor (defense-in-depth): skipped when younger than + ``min_age_minutes`` (already clamped >= ``_MIN_AGE_FLOOR_MINUTES`` by + the request body validator), measured by ``BlobReference.last_modified``. + Counted as ``skipped_recent``. + """ + settings = get_settings() + blob_store = create_blob_store(settings) + + referenced = await _scan_referenced_uris(request) + + registry = request.app.state.registry + queue_manager = registry.queue_manager + live_workers = set(registry.active_sessions()) + + now = time.time() + age_cutoff_seconds = min_age_minutes * 60 + + scanned = 0 + candidates: list[BlobReference] = [] + skipped_recent = 0 + skipped_pending_session = 0 + reclaimable_bytes = 0 + + async for ref in blob_store.scan(): + scanned += 1 + if ref.uri in referenced: + continue + if ref.session_id in live_workers or not await queue_manager.is_fully_drained( + ref.session_id + ): + skipped_pending_session += 1 + continue + if now - ref.last_modified < age_cutoff_seconds: + skipped_recent += 1 + continue + candidates.append(ref) + reclaimable_bytes += ref.size + + candidates.sort(key=lambda b: b.uri) + + return { + "scanned_disk_blobs": scanned, + "referenced_uris": len(referenced), + "orphans_found": len(candidates), + "reclaimable_bytes": reclaimable_bytes, + "skipped_recent": skipped_recent, + "skipped_pending_session": skipped_pending_session, + "candidates": candidates, + } + + +class BlobReclaimBody(BaseModel): + """Body for POST /admin/blobs/reclaim.""" + + dry_run: bool = True + min_age_minutes: int = 60 + max_delete: int | None = Field(default=None, ge=1) + + @field_validator("min_age_minutes") + @classmethod + def _min_age_at_least_floor(cls, v: int) -> int: + """Reject (422) below the hard safety floor + rather than silently raising -- a caller passing 0 must not be able + to disable the age gate. + """ + if v < _MIN_AGE_FLOOR_MINUTES: + raise ValueError( + f"min_age_minutes must be >= {_MIN_AGE_FLOOR_MINUTES} " + f"(hard safety floor); got {v}" + ) + return v + + # --------------------------------------------------------------------------- # Per-request store dependencies (via app.state — no circular import) # --------------------------------------------------------------------------- @@ -544,3 +843,108 @@ def list_keys( return { "keys": [{"hash": h, "id": record.get("id", "")} for h, record in store.items()] } + + +# --- Blob reclaim (orphaned-blob GC) ---------------------------------------- + + +@router.post("/blobs/reclaim", status_code=200) +async def reclaim_blobs(body: BlobReclaimBody, request: Request) -> dict[str, Any]: + """Preview (dry-run) or apply reclamation of orphaned blob artifacts. + + An orphan is a blob whose ``ci-blob://`` URI is referenced by NO registered + carrier property anywhere in the graph (scanned globally, across ALL + workspaces), and whose session is both fully drained (durable + ``QueueManager`` state, not in-memory worker liveness) and older than + ``min_age_minutes``. + + ``dry_run=true`` (default) computes and reports the candidate set without + deleting anything. ``dry_run=false`` requires ``max_delete`` (422 + otherwise -- a conscious blast-radius opt-in for an irreversible, + cross-workspace delete) and performs its OWN fresh, authoritative + ``_select_orphans`` scan at delete time (``rescanned: true`` in the + response) -- it never deletes a URI that is referenced or pending at the + moment of deletion, independent of any earlier dry-run preview. + + Deletion goes through ``BlobStore.delete(uri, if_unmodified=ref)`` -- a + fenced compare-and-delete that refuses (leaves the blob untouched) if the + blob was rewritten since ``scan()`` observed it, and is idempotent (a blob + already gone counts as not-deleted, never an error). Capped at + ``max_delete``; ``orphans_found`` and ``reclaimable_bytes`` always reflect + the FULL candidate set even when ``max_delete`` caps how many are actually + removed. One structured audit log line is emitted per successful delete; + blob CONTENTS are never logged, only the ``ci-blob://`` URI. + """ + if not body.dry_run and body.max_delete is None: + raise HTTPException( + status_code=422, + detail=( + "max_delete is required when dry_run=false -- a conscious " + "blast-radius opt-in for an irreversible, cross-workspace " + "delete. Omit dry_run (or set it true) to preview first." + ), + ) + + if body.dry_run: + # Preview only: compute the candidate set and report it, delete nothing. + selection = await _select_orphans(request, min_age_minutes=body.min_age_minutes) + candidates: list[BlobReference] = selection.pop("candidates") + return { + "dry_run": True, + **selection, + "sample": [b.uri for b in candidates[:_MAX_SAMPLE]], + "rescanned": False, + "deleted": 0, + "deleted_bytes": 0, + } + + assert body.max_delete is not None # guaranteed by the 422 guard above + + # Acquire the destructive-apply single-flight BEFORE the authoritative scan, + # so a rejected concurrent apply fails fast and never even scans. + global _reclaim_apply_inflight + if _reclaim_apply_inflight: + raise HTTPException( + status_code=409, + detail=( + "a blob-reclaim apply is already in progress -- concurrent " + "applies would each honour max_delete independently and " + "together exceed the intended blast radius. Retry once it " + "completes." + ), + ) + _reclaim_apply_inflight = True + try: + # The apply's OWN fresh, authoritative scan at delete time -- it never + # deletes a URI that is referenced or pending at the moment of deletion, + # independent of any earlier dry-run preview. + selection = await _select_orphans(request, min_age_minutes=body.min_age_minutes) + candidates = selection.pop("candidates") + response: dict[str, Any] = { + "dry_run": False, + **selection, + "sample": [b.uri for b in candidates[:_MAX_SAMPLE]], + "rescanned": True, + "deleted": 0, + "deleted_bytes": 0, + } + + blob_store = create_blob_store(get_settings()) + deleted = 0 + deleted_bytes = 0 + # Each delete is fenced against the reference just observed by the scan + # above (delete refuses if the blob was rewritten since), so a blob that + # was concurrently re-minted or modified is left intact and counted as + # not-deleted rather than destroyed. + for ref in candidates[: body.max_delete]: + if not await blob_store.delete(ref.uri, if_unmodified=ref): + continue # absent or changed since scan -- left untouched + deleted += 1 + deleted_bytes += ref.size + _audit_blob_reclaim_delete(request, uri=ref.uri) + finally: + _reclaim_apply_inflight = False + + response["deleted"] = deleted + response["deleted_bytes"] = deleted_bytes + return response diff --git a/pyproject.toml b/pyproject.toml index 35f981f0..aeb3cb7e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "context-intelligence-server" -version = "7.3.0" +version = "7.4.0" description = "Context Intelligence Server for Amplifier" requires-python = ">=3.11" dependencies = [ diff --git a/tests/neo4j/test_blob_reclaim_e2e.py b/tests/neo4j/test_blob_reclaim_e2e.py new file mode 100644 index 00000000..a27a4d81 --- /dev/null +++ b/tests/neo4j/test_blob_reclaim_e2e.py @@ -0,0 +1,144 @@ +"""End-to-end blob-reclaim against a REAL Neo4j graph and a REAL filesystem +blob store -- no mocks, no stubs. + +Proves the reshaped reclaim GC for real: the reference scan runs as Cypher +against a live graph, orphan selection uses the real QueueManager drain state, +and deletion goes through the real fenced BlobStore.delete -- the orphan file +actually disappears from disk while the referenced blob survives. + + uv run pytest tests/neo4j/test_blob_reclaim_e2e.py -q -m neo4j +""" + +from __future__ import annotations + +import asyncio +import json +import os +import time +from pathlib import Path +from types import SimpleNamespace +from typing import Any + +import pytest +from neo4j import READ_ACCESS, AsyncGraphDatabase + +from context_intelligence_server.blob_store import create_blob_store +from context_intelligence_server.config import Settings +from context_intelligence_server.neo4j_store import ensure_neo4j_schema +from context_intelligence_server.queue_manager import FileSystemQueueManager +from context_intelligence_server.registry import SessionRegistry +from context_intelligence_server.routers import admin +from context_intelligence_server.routers.admin import BlobReclaimBody, reclaim_blobs + +pytestmark = pytest.mark.neo4j + +_WS = "reclaim_e2e" + + +def _settings(tmp_path: Path) -> Settings: + # The reclaim path reaches Neo4j via app.state.neo4j_query_driver, so only + # the filesystem roots matter here. + s = Settings() + s.blob_path = str(tmp_path / "blobs") + s.queues_path = str(tmp_path / "queues") + return s + + +def _build_registry(queues_dir: Path) -> SessionRegistry: + reg = SessionRegistry() + reg._queue_manager = FileSystemQueueManager(queues_dir=queues_dir) + reg._write_semaphore = asyncio.Semaphore(8) + reg._max_delivery_attempts = 3 + return reg + + +def _backdate(path: Path, minutes: int) -> None: + """Age a blob file past the reclaim min-age floor (>= 15 min).""" + old = time.time() - minutes * 60 + os.utime(path, (old, old)) + + +def _blob_file(blob_root: Path, session_id: str, key: str) -> Path: + return blob_root / session_id / "blobs" / f"{key}.json" + + +@pytest.fixture +async def _driver(neo4j_container: dict[str, Any]): + driver = AsyncGraphDatabase.driver( + neo4j_container["bolt_url"], + auth=(neo4j_container["user"], neo4j_container["password"]), + ) + await ensure_neo4j_schema(driver) + async with driver.session() as session: + await session.run("MATCH (n) DETACH DELETE n") + yield driver + async with driver.session() as session: + await session.run("MATCH (n) DETACH DELETE n") + await driver.close() + + +async def test_reclaim_deletes_orphan_keeps_referenced_e2e( + tmp_path: Path, neo4j_container: dict[str, Any], _driver, monkeypatch +) -> None: + settings = _settings(tmp_path) + monkeypatch.setattr(admin, "get_settings", lambda: settings) + + # Real filesystem blob store: write two real blobs. + store = create_blob_store(settings) + ref_kept = await store.write("sess_ref", "kept", {"payload": "referenced"}) + ref_orphan = await store.write("sess_orphan", "orphan", {"payload": "dangling"}) + + blob_root = Path(settings.blob_path) + kept_file = _blob_file(blob_root, "sess_ref", "kept") + orphan_file = _blob_file(blob_root, "sess_orphan", "orphan") + assert kept_file.exists() and orphan_file.exists() + + # Age both past the 15-minute floor so neither is skipped_recent. + _backdate(kept_file, 60) + _backdate(orphan_file, 60) + + # Reference ONLY the kept blob in the real graph, via a carrier property. + async with _driver.session() as session: + await session.run( + "CREATE (:Event {node_id: 'e1', workspace: $ws, session_id: 'sess_ref', " + "data: $data})", + ws=_WS, + data=json.dumps({"$blob_ref": ref_kept.uri}), + ) + + # Real registry + real queue manager; both sessions fully drained (no logs). + registry = _build_registry(tmp_path / "queues") + + request = SimpleNamespace( + # request.scope["state"]["contributor_id"] is read by the audit logger. + scope={"state": {"contributor_id": "admin"}}, + app=SimpleNamespace( + state=SimpleNamespace( + registry=registry, + neo4j_query_driver=_driver, + neo4j_query_access_mode="READ", + ) + ), + ) + # Sanity: the exact access-mode constant the endpoint will use. + assert admin._access_mode_const("READ") is READ_ACCESS + + # 1) DRY-RUN: finds exactly the orphan, deletes nothing. + admin._reclaim_apply_inflight = False + dry = await reclaim_blobs( + BlobReclaimBody(dry_run=True, min_age_minutes=15), request + ) + assert dry["dry_run"] is True + assert dry["orphans_found"] == 1, dry + assert ref_orphan.uri in dry["sample"] + assert ref_kept.uri not in dry["sample"] + assert orphan_file.exists() # nothing deleted in dry-run + + # 2) APPLY: the orphan file actually disappears; the referenced one survives. + applied = await reclaim_blobs( + BlobReclaimBody(dry_run=False, min_age_minutes=15, max_delete=10), request + ) + assert applied["deleted"] == 1, applied + assert not orphan_file.exists(), "orphan blob must be gone from disk" + assert kept_file.exists(), "referenced blob must survive" + assert admin._reclaim_apply_inflight is False # single-flight released diff --git a/tests/test_blob_carrier_allowlist.py b/tests/test_blob_carrier_allowlist.py new file mode 100644 index 00000000..168445ce --- /dev/null +++ b/tests/test_blob_carrier_allowlist.py @@ -0,0 +1,160 @@ +"""Tests for the blob-carrier allowlist runtime tripwire. + +Covers the single source of truth shared by the mint path +(``blob_processor.BLOB_REF_CARRIER_PROPERTIES`` / +``blob_processor.assert_carrier_registered``) and the reclaim-scan path +(``routers.admin._BLOB_REF_CARRIER_PROPERTIES`` / ``_BLOB_REF_SCAN_QUERY``): + +1. The allowlist is exactly the current 4-item tuple (regression lock). +2. ``routers.admin`` imports the SAME object -- no local re-declaration to + drift out of sync. +3. The generated Cypher scan query references exactly the allowlist's + properties -- no more, no less (mint/scan agreement, structurally). +4. ``assert_carrier_registered`` is non-vacuous: it raises for an + unregistered property and is a no-op for a registered one. +5. ``process_event_data`` (the real mint call path) propagates the + tripwire's exception -- fail-closed, BEFORE any blob is written -- when + its destination carrier ("data") is not registered, and completes + normally when it is. +""" + +from __future__ import annotations + +import re +from typing import Any +from unittest.mock import AsyncMock + +import pytest +from context_intelligence_server.blob_processor import ( + BLOB_REF_CARRIER_PROPERTIES, + UnregisteredBlobCarrierError, + assert_carrier_registered, + process_event_data, +) +from context_intelligence_server.blob_store import BlobReference + + +def _ref(uri: str) -> BlobReference: + """A BlobReference for a mocked write() return (only .uri is read here).""" + session_id, _, key = uri.removeprefix("ci-blob://").partition("/") + return BlobReference( + uri=uri, session_id=session_id, key=key, size=0, last_modified=0.0 + ) +from context_intelligence_server.routers.admin import ( + _BLOB_REF_CARRIER_PROPERTIES as admin_carrier_properties, +) +from context_intelligence_server.routers.admin import ( + _BLOB_REF_SCAN_QUERY, +) + +# --------------------------------------------------------------------------- +# 1. Regression lock -- current 4-item allowlist +# --------------------------------------------------------------------------- + + +def test_carrier_properties_locked() -> None: + """BLOB_REF_CARRIER_PROPERTIES is exactly the specified 4-item tuple.""" + assert BLOB_REF_CARRIER_PROPERTIES == ("data", "tool_input", "prompt", "response") + + +# --------------------------------------------------------------------------- +# 2. admin.py imports the SAME allowlist -- no local re-declaration +# --------------------------------------------------------------------------- + + +def test_admin_imports_same_allowlist_object() -> None: + """routers.admin re-exports blob_processor's tuple by identity, not a + hand-copied duplicate -- proves there is exactly ONE allowlist object.""" + assert admin_carrier_properties is BLOB_REF_CARRIER_PROPERTIES + + +# --------------------------------------------------------------------------- +# 3. Generated scan query references exactly the allowlist's properties +# --------------------------------------------------------------------------- + + +def test_scan_query_matches_allowlist_exactly() -> None: + """The Cypher query built for the reclaim scan mentions exactly the + properties in BLOB_REF_CARRIER_PROPERTIES -- no more, no less. + + This is the structural lock that makes mint/scan drift impossible: if a + property is ever added to (or removed from) the allowlist without the + query being regenerated from it, this test fails. + """ + referenced_props = set(re.findall(r"n\.(\w+)", _BLOB_REF_SCAN_QUERY)) + assert referenced_props == set(BLOB_REF_CARRIER_PROPERTIES) + + +# --------------------------------------------------------------------------- +# 4. assert_carrier_registered -- non-vacuous tripwire +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize("carrier", list(BLOB_REF_CARRIER_PROPERTIES)) +def test_assert_carrier_registered_passes_for_registered_carrier( + carrier: str, +) -> None: + """A registered carrier (each of the current 4) does not trip the guard.""" + assert_carrier_registered(carrier) # must not raise + + +def test_assert_carrier_registered_raises_for_unregistered_carrier() -> None: + """An unregistered carrier property trips the guard immediately. + + Proves the tripwire is non-vacuous: it actually fires. Simulates a + plausible future scenario -- a new field-lifter/enricher promoting a + value onto a brand-new node property ("artifact_content") + that nobody added to BLOB_REF_CARRIER_PROPERTIES. + """ + with pytest.raises(UnregisteredBlobCarrierError, match="artifact_content"): + assert_carrier_registered("artifact_content") + + +# --------------------------------------------------------------------------- +# 5. process_event_data -- the real mint call path +# --------------------------------------------------------------------------- + + +async def test_process_event_data_fails_closed_when_data_carrier_unregistered( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """If "data" is (hypothetically) removed from the allowlist, the mint + call fails loud BEFORE writing any blob -- not after. + + Regression target: this is the exact failure mode the allowlist tripwire + exists to catch -- a carrier property silently dropping out while the + mint path keeps writing to it. blob_store.write must never be called: + the guard fires before any blob is persisted, matching "fail loud at the + source, not after a live blob is deleted." + """ + import context_intelligence_server.blob_processor as blob_processor_module + + monkeypatch.setattr( + blob_processor_module, + "BLOB_REF_CARRIER_PROPERTIES", + ("tool_input", "prompt", "response"), # "data" removed + ) + + data: dict[str, Any] = {"result": {"answer": 42}} + blob_store = AsyncMock() + blob_store.write = AsyncMock(return_value=_ref("ci-blob://sess/node__result")) + + with pytest.raises(UnregisteredBlobCarrierError, match="'data'"): + await process_event_data(data, blob_store, "sess", "node") + + blob_store.write.assert_not_called() + # data must be untouched -- the guard fired before any mutation/write + assert data == {"result": {"answer": 42}} + + +async def test_process_event_data_succeeds_when_data_carrier_registered() -> None: + """Sanity/non-regression: with the real (unmodified) allowlist, the mint + path completes normally -- the guard does not false-trip on the + ordinary, correctly-registered path.""" + data: dict[str, Any] = {"result": {"answer": 42}} + blob_store = AsyncMock() + blob_store.write = AsyncMock(return_value=_ref("ci-blob://sess/node__result")) + + await process_event_data(data, blob_store, "sess", "node") + + assert data["result"] == {"$blob_ref": "ci-blob://sess/node__result"} diff --git a/tests/test_blob_reclaim_endpoint.py b/tests/test_blob_reclaim_endpoint.py new file mode 100644 index 00000000..73363369 --- /dev/null +++ b/tests/test_blob_reclaim_endpoint.py @@ -0,0 +1,187 @@ +"""The POST /admin/blobs/reclaim orchestration: dry-run vs apply, the +blast-radius cap, the destructive-apply single-flight, and fenced deletion. + +The *selection* logic (which blobs are orphans) is covered against a real +graph elsewhere; these tests pin the endpoint's own contract by stubbing the +one selection call, so they run without neo4j. +""" + +from __future__ import annotations + +from types import SimpleNamespace +from typing import Any + +import pytest +from context_intelligence_server.blob_store import BlobReference +from context_intelligence_server.routers import admin +from context_intelligence_server.routers.admin import BlobReclaimBody, reclaim_blobs +from fastapi import HTTPException + +pytestmark = pytest.mark.integration + + +def _ref(uri: str, size: int = 10) -> BlobReference: + session_id, _, key = uri.removeprefix("ci-blob://").partition("/") + return BlobReference( + uri=uri, session_id=session_id, key=key, size=size, last_modified=1.0 + ) + + +def _request() -> Any: + return SimpleNamespace(app=SimpleNamespace(state=SimpleNamespace())) + + +def _stub_selection(monkeypatch, candidates: list[BlobReference]) -> None: + async def _fake_select(_request: Any, *, min_age_minutes: int) -> dict[str, Any]: + return { + "scanned_disk_blobs": len(candidates), + "referenced_uris": 0, + "orphans_found": len(candidates), + "reclaimable_bytes": sum(c.size for c in candidates), + "skipped_recent": 0, + "skipped_pending_session": 0, + "candidates": list(candidates), + } + + monkeypatch.setattr(admin, "_select_orphans", _fake_select) + + +class _FakeStore: + """A blob store whose delete() honours a per-uri fence verdict.""" + + def __init__(self, deletable: set[str]) -> None: + self._deletable = deletable + self.deleted: list[str] = [] + + async def delete(self, uri: str, if_unmodified: BlobReference | None = None) -> bool: + if uri in self._deletable: + self.deleted.append(uri) + return True + return False # absent or changed since scan -- fenced out + + +@pytest.fixture(autouse=True) +def _reset_single_flight(): + admin._reclaim_apply_inflight = False + yield + admin._reclaim_apply_inflight = False + + +async def test_apply_without_max_delete_is_422(monkeypatch) -> None: + """F2: a destructive apply must name its blast radius.""" + with pytest.raises(HTTPException) as exc: + await reclaim_blobs( + BlobReclaimBody(dry_run=False, max_delete=None), _request() + ) + assert exc.value.status_code == 422 + + +async def test_dry_run_reports_without_deleting(monkeypatch) -> None: + """F1: dry-run (the default) previews and deletes nothing.""" + cands = [_ref("ci-blob://s1/a"), _ref("ci-blob://s1/b")] + _stub_selection(monkeypatch, cands) + store = _FakeStore({"ci-blob://s1/a", "ci-blob://s1/b"}) + monkeypatch.setattr(admin, "create_blob_store", lambda _s: store) + + resp = await reclaim_blobs(BlobReclaimBody(dry_run=True), _request()) + + assert resp["dry_run"] is True + assert resp["rescanned"] is False + assert resp["orphans_found"] == 2 + assert resp["deleted"] == 0 + assert store.deleted == [] # nothing touched + + +async def test_apply_deletes_through_fenced_protocol(monkeypatch) -> None: + """Happy-path apply: fresh scan (rescanned), fenced delete, audit per delete.""" + cands = [_ref("ci-blob://s1/a"), _ref("ci-blob://s1/b")] + _stub_selection(monkeypatch, cands) + store = _FakeStore({"ci-blob://s1/a", "ci-blob://s1/b"}) + monkeypatch.setattr(admin, "create_blob_store", lambda _s: store) + audited: list[str] = [] + monkeypatch.setattr( + admin, "_audit_blob_reclaim_delete", lambda _r, *, uri: audited.append(uri) + ) + + resp = await reclaim_blobs( + BlobReclaimBody(dry_run=False, max_delete=10), _request() + ) + + assert resp["rescanned"] is True + assert resp["deleted"] == 2 + assert sorted(store.deleted) == ["ci-blob://s1/a", "ci-blob://s1/b"] + assert sorted(audited) == ["ci-blob://s1/a", "ci-blob://s1/b"] + assert admin._reclaim_apply_inflight is False # released + + +async def test_fenced_delete_refusal_is_not_counted(monkeypatch) -> None: + """R3/R4: a blob changed/re-referenced since the scan is fenced out -- + delete() returns False, it stays on disk and is NOT counted as deleted.""" + cands = [_ref("ci-blob://s1/a"), _ref("ci-blob://s1/b")] + _stub_selection(monkeypatch, cands) + # Only 'a' is still deletable; 'b' was re-minted since the scan. + store = _FakeStore({"ci-blob://s1/a"}) + monkeypatch.setattr(admin, "create_blob_store", lambda _s: store) + monkeypatch.setattr( + admin, "_audit_blob_reclaim_delete", lambda _r, *, uri: None + ) + + resp = await reclaim_blobs( + BlobReclaimBody(dry_run=False, max_delete=10), _request() + ) + + assert resp["deleted"] == 1 + assert store.deleted == ["ci-blob://s1/a"] # 'b' left intact + + +async def test_max_delete_caps_blast_radius(monkeypatch) -> None: + """F4: orphans_found reflects the FULL set; only max_delete are removed.""" + cands = [_ref(f"ci-blob://s1/{k}") for k in "abcde"] + _stub_selection(monkeypatch, cands) + store = _FakeStore({c.uri for c in cands}) + monkeypatch.setattr(admin, "create_blob_store", lambda _s: store) + monkeypatch.setattr( + admin, "_audit_blob_reclaim_delete", lambda _r, *, uri: None + ) + + resp = await reclaim_blobs( + BlobReclaimBody(dry_run=False, max_delete=2), _request() + ) + + assert resp["orphans_found"] == 5 # full candidate set + assert resp["deleted"] == 2 # capped + assert len(store.deleted) == 2 + + +async def test_concurrent_apply_is_single_flighted(monkeypatch) -> None: + """E1: a second apply while one is in flight is refused (409) before it + even scans -- two applies would each honour max_delete and jointly exceed + the operator's intended blast radius.""" + admin._reclaim_apply_inflight = True # simulate an apply already running + scanned = False + + async def _should_not_run(_request: Any, *, min_age_minutes: int) -> dict[str, Any]: + nonlocal scanned + scanned = True + return {"candidates": []} + + monkeypatch.setattr(admin, "_select_orphans", _should_not_run) + + with pytest.raises(HTTPException) as exc: + await reclaim_blobs( + BlobReclaimBody(dry_run=False, max_delete=1), _request() + ) + + assert exc.value.status_code == 409 + assert scanned is False # fail-fast: rejected before the authoritative scan + + +async def test_dry_run_is_never_single_flighted(monkeypatch) -> None: + """A preview must never be blocked by an in-flight apply.""" + admin._reclaim_apply_inflight = True + _stub_selection(monkeypatch, [_ref("ci-blob://s1/a")]) + + resp = await reclaim_blobs(BlobReclaimBody(dry_run=True), _request()) + + assert resp["dry_run"] is True + assert resp["orphans_found"] == 1 diff --git a/uv.lock b/uv.lock index 9cd64dbd..1d81407d 100644 --- a/uv.lock +++ b/uv.lock @@ -233,7 +233,7 @@ wheels = [ [[package]] name = "context-intelligence-server" -version = "7.3.0" +version = "7.4.0" source = { editable = "." } dependencies = [ { name = "aiofiles" },