Blob storage API isolation (stacked on #70)

AGENTS.md

@@ -13,13 +13,37 @@ See [README.md](README.md) for full setup instructions.
 
 ---
 
-## Current Work: Documentation & Setup Cleanup
-
-We are cleaning up this repo's **documentation and setup instructions** so the
-server + Neo4j run locally **without Docker Compose**, with API keys primed for
-local runs. Scope is intentionally narrow — do this and nothing else.
-
-### KEEP (do NOT remove)
+## Current Work: Phase-2 review remediation (server data-quality + deploy safety)
+
+**Active engagement.** We are addressing the issues from the Phase-2 PR review:
+a blob-reclaim cap-inversion, duplicate `Iteration` nodes on the retry path, and —
+the big one — making the server **safe (not merely survivable) on un-migrated /
+degraded graph state**, via a first-class **maintenance mode** (gate ingest + query
+when the `:Node` uniqueness constraint is absent, structured `503` + `Retry-After`,
+`/status` advertising, live re-probe that self-clears without restart) plus an
+explicitly-triggered **`/admin/maintenance`** execution channel and an
+out-of-band upgrade path. Full plan + council verdicts live in the **workspace-root
+`docs/`** (one level up): `docs/plans/2026-08-13-review-remediation-plan.md`,
+`docs/plans/2026-08-13-ws3-implementation-spec.md`, `docs/council/2026-08-13-*`.
+
+**Engagement guardrails** (see the workspace-root `AGENTS.md` for the authoritative
+version): issue-driven only; **minimal, surgical, no-regression** diffs; every fix
+**evidence-backed** and **DTU-validated** (real Neo4j, real restart) before "done";
+migrations/rectification run **OUT-OF-BAND**, never in the server startup/critical
+path. "Out-of-band" now has **two sanctioned execution channels**, both explicitly
+triggered (never at startup) and sharing one mechanism (`neo4j_store.run_repair`):
+(1) the standalone `migrations/run.py --apply` (local/VM/direct-Neo4j), and (2) the
+network-reachable, admin-authenticated `POST /admin/maintenance` (the cloud channel).
+The server itself performs zero migration work automatically — it only assesses,
+advertises (`/status`, `/version`), and gates while degraded.
+
+> **Superseded:** the earlier "Documentation & Setup Cleanup" engagement (remove
+> Docker Compose / `start.sh`, add a local Neo4j script) is **complete/historical**.
+> Its `docs/`-boundary rule and the container base-image policy below are **standing
+> constraints** and still apply. The old "do not refactor server code" boundary does
+> **not** apply to this engagement — server code changes are the point of it.
+
+### Container base image policy (S360 / SCA) — STANDING CONSTRAINT (do NOT remove)
 
 - **`Dockerfile` (the server image).** This is the shipping-product container and
   it is **S360-compliant** via PR #50 (`payneio` — "adopt Azure Linux base +
@@ -141,8 +165,9 @@ docker run -d --name neo4j-ci \
 cp server-config.example.yaml server-config.yaml
 # Edit server-config.yaml with your Neo4j connection details
 
-# 3. Start
-uvicorn context_intelligence_server.main:app --reload
+# 3. Start  (use main:asgi_app — the middleware-wrapped app that enforces bearer auth;
+#            main:app is the bare app with NO auth middleware — dev/testing only)
+uvicorn context_intelligence_server.main:asgi_app --reload
 ```
 
 Or use Docker Compose to run everything together:

Dockerfile

@@ -23,6 +23,11 @@ COPY --from=ghcr.io/astral-sh/uv:latest /uv /uvx /bin/
 
 COPY pyproject.toml .
 COPY context_intelligence_server/ context_intelligence_server/
+# Ship the standalone, out-of-band maintenance/migration scripts in the image so
+# they can be run against a live deployment (e.g. `docker exec ... python
+# scripts/<name>.py`). These are data-rectification tools (never run at startup);
+# see CHANGELOG.md. Required for cloud VMs/ACI where there is no repo checkout.
+COPY scripts/ scripts/
 COPY docker-entrypoint.sh .
 RUN chmod +x docker-entrypoint.sh
 

README.md

@@ -68,6 +68,49 @@ for the full ingest/drain flow.
 
 ---
 
+## Upgrading
+
+**Healthy / already-migrated deployments upgrade to `6.8.0` with zero
+action and zero behavior change.** `SCHEMA_VERSION` is unchanged (`1`) --
+this release adds no stored node/edge shape, only server behavior. See
+[CHANGELOG.md](CHANGELOG.md) for the full `6.8.0` entry and
+[migrations/manifest.yaml](migrations/manifest.yaml) for the
+machine-readable upgrade-mechanism entry (`from -> to`, whether it's
+schema-affecting, which script to run, and how to verify).
+
+> ⚠️ **A deployment carrying pre-existing un-migrated / duplicate `:Node`
+> data now boots into MAINTENANCE MODE under `6.8.0`** -- `POST /events`
+> and the query/`cypher` surface return a structured `503` (`Retry-After` +
+> reason) instead of `6.7.1`'s "degraded but still writing" state, which
+> could silently manufacture new duplicates while un-migrated. `GET
+> /status` and `GET /version` stay up throughout and advertise the mode
+> (`healthy` / `maintenance` / `degraded` / `unknown`,
+> `maintenance_started_at`, `maintenance_elapsed_seconds`) -- see
+> [docs/maintenance-mode.md](docs/maintenance-mode.md) for the full
+> contract.
+>
+> **Rectify once, out-of-band -- pick whichever channel you can reach:**
+>
+> ```bash
+> # Local / VM / direct Neo4j access:
+> python migrations/run.py --status   # read-only report (safe, writes nothing)
+> python migrations/run.py --apply    # rectify: dedup + :Node backfill + constraint create
+> ```
+>
+> ```bash
+> # Cloud / ACA, where the private Neo4j is not directly reachable:
+> curl -X POST -H "Authorization: Bearer $ADMIN_KEY" https://<server>/admin/maintenance
+> curl -H "Authorization: Bearer $ADMIN_KEY" https://<server>/admin/maintenance   # poll to completion
+> ```
+>
+> The server **self-clears to healthy with no restart** once rectified (the
+> gate re-probes live; it never latches). Use
+> `context-intelligence-upload` to backfill any events that could not be
+> ingested during the maintenance window. A fresh/empty graph has nothing
+> to rectify and boots normally.
+
+---
+
 ## Neo4j Plugins (APOC + GDS)
 
 The server needs Neo4j 5.x reachable over Bolt with the **APOC** procedures
@@ -149,7 +192,7 @@ python scripts/prime-local-config.py --neo4j-password '<neo4j-password>'
 ```bash
 export AMPLIFIER_CONTEXT_INTELLIGENCE_SERVER_CONFIG_FILE="$(pwd)/server-config.yaml"
 uv sync
-uv run uvicorn context_intelligence_server.main:app --host 127.0.0.1 --port 8000
+uv run uvicorn context_intelligence_server.main:asgi_app --host 127.0.0.1 --port 8000
 ```
 
 Open [http://localhost:8000](http://localhost:8000) to confirm the server is

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,91 @@
 )
 
 
+# ---------------------------------------------------------------------------
+# Blob-ref carrier allowlist -- single source of truth (WS-5 runtime tripwire)
+# ---------------------------------------------------------------------------
+#
+# 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 is the WS-5 runtime tripwire: it 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 +174,16 @@ async def process_event_data(
     """
     _lift_raw_fields(data)
 
+    # WS-5 runtime tripwire: 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:
@@ -96,8 +192,8 @@ async def process_event_data(
 
         key = f"{node_id}__{field_name}"
         try:
-            uri = await blob_store.write(session_id, key, value)
-            data[field_name] = {"$blob_ref": uri}
+            ref = await blob_store.write(session_id, key, value)
+            data[field_name] = {"$blob_ref": ref.uri}
         except Exception as exc:  # noqa: BLE001
             logger.warning(
                 "blob_offload_failed session=%s field=%s node=%s: %s",