PR6: Blob-storage reclaim GC

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

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: