Context Intelligence data-quality & operability (Phase 2): trustworthy metrics, session attribution, durable fix, incomplete-label heal, blob reclaim

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:

context_intelligence_server/config.py

@@ -902,6 +902,17 @@ def _validate_crash_recovery_sweep_interval(cls, v: int) -> int:
     status_inactive_timeout: float = 1800.0  # 30 min  — /status visibility
     stale_session_timeout: float = 432000.0  # 5 days  — worker reap
 
+    # -------------------------------------------------------------------------
+    # Maintenance mode (WS-3a: live gate + /status; WS-3c wires
+    # POST/GET /admin/maintenance on top of this same coordinator/seam)
+    # -------------------------------------------------------------------------
+    maintenance_probe_ttl_seconds: float = 5.0  # :Node constraint probe cache TTL
+    maintenance_retry_after_seconds: int = 30  # Retry-After on the maintenance 503
+    # WS-3c: bounded pre-op quiesce before run_repair (spec sec 5.3) -- covers
+    # ordinary in-flight flushes (_DRAIN_POLL_INTERVAL is 0.05s); a flush that
+    # outlives this is a residual, DETECTED risk (constraint create fails loud).
+    maintenance_quiesce_seconds: float = 2.0
+
     @classmethod
     def settings_customise_sources(
         cls,

context_intelligence_server/handlers/data_layer_2/content_block.py

@@ -50,8 +50,11 @@ async def __call__(self, event: str, data: dict[str, Any]) -> HookResult:
 
         block_index = data.get("block_index")
         iteration_id = self.services.data_layer_2.active_iteration_id
-        # ID format is "{session_id}::iteration::{n}"; [-1] extracts the iteration number.
-        # If the cursor format ever changes, this extraction must be updated to match.
+        # active_iteration_id's trailing "::"-segment is always the plain iteration
+        # number, whether the cursor is the bare "{session_id}::iteration::{n}" shape
+        # or the run-scoped "{session_id}::orch_run::{ts}::iteration::{n}" shape
+        # (P2.1 fix, see IterationHandler) -- [-1] extracts it either way. If the
+        # cursor format ever changes this extraction must be updated to match.
         iteration_n = iteration_id.split("::")[-1] if iteration_id else "0"
         block_node_id = f"{session_id}::block::{iteration_n}::{block_index}"
 

context_intelligence_server/handlers/data_layer_2/iteration.py

@@ -36,6 +36,18 @@ class IterationHandler:
     def __init__(self, services: HookStateService) -> None:
         self.services = services
 
+    def _current_iteration_scope(self) -> str:
+        """The additive 'run' | 'unscoped' discriminator (D6), sourced from the
+        SAME cursor field (``execution_start_ts``) used to decide the node_id
+        shape. A single source of truth so all three upsert_node call sites
+        (provider:request, llm:request, llm:response) always agree.
+        """
+        return (
+            "run"
+            if self.services.data_layer_2.execution_start_ts is not None
+            else "unscoped"
+        )
+
     async def __call__(self, event: str, data: dict[str, Any]) -> HookResult:
         """Dispatch to the appropriate sub-handler.
 
@@ -64,22 +76,56 @@ async def _handle_provider_request(
     ) -> None:
         """Create Iteration node and set active_iteration_id cursor.
 
-        - Computes iteration_id as '{session_id}::iteration::{iteration_number}'
+        - Computes iteration_id as run-scoped:
+          '{session_id}::orch_run::{execution_start_ts}::iteration::{iteration_number}'
+          when an orchestrator run is active (execution_start_ts cursor set), using the
+          SAME disambiguator OrchestratorRun uses for its own node_id (P2.1 / I5 fix).
+          Falls back to the bare '{session_id}::iteration::{iteration_number}' shape when
+          no run is active -- this bare shape is also what historical (pre-fix) Iteration
+          nodes look like, making them detectable as suspect. See CHANGELOG.md.
         - Sets active_iteration_id cursor on DataLayer2State
         - Creates Iteration:SST_EVENT node with session_id, iteration_number, started_at
         - Conditionally creates E06: OrchestratorRun -[:HAS_PART {sst_semantic: 'CONTAINS'}]->
           Iteration when execution_start_ts cursor is set
+
+        Without run-scoping, iteration_number alone (a counter scoped to the whole
+        session, not the run) can repeat across orchestrator runs -- e.g. after a
+        drainer restart/replay resets the in-memory counter -- causing distinct runs'
+        Iteration nodes to MERGE onto the same node_id (I5) and their usage figures to
+        clobber each other (I3, usage_cache_write in particular).
         """
         # Increment counter to get the next iteration number
         self.services.data_layer_2.iteration_count += 1
         iteration_number = self.services.data_layer_2.iteration_count
 
         timestamp: str = data.get("timestamp", "")
-        iteration_id = f"{session_id}::iteration::{iteration_number}"
+
+        execution_start_ts = self.services.data_layer_2.execution_start_ts
+        orch_run_id: str | None = None
+        if execution_start_ts is not None:
+            orch_run_id = f"{session_id}::orch_run::{execution_start_ts}"
+            iteration_id = f"{orch_run_id}::iteration::{iteration_number}"
+        else:
+            iteration_id = f"{session_id}::iteration::{iteration_number}"
 
         # Set cursor so llm:request and llm:response can find this iteration
         self.services.data_layer_2.active_iteration_id = iteration_id
 
+        # D6: additive, queryable discriminator between a run-scoped iteration
+        # and a legitimate loop-basic session with no active orchestrator run.
+        # Bare shape != state-lost; see the module/method docstrings.
+        iteration_scope = self._current_iteration_scope()
+        if iteration_scope == "unscoped":
+            # INFO (not WARNING, spec §10.4): a loop-basic session with no
+            # execution:start is a normal case, not an alert-worthy anomaly.
+            logger.info(
+                "unscoped_iteration_emitted session=%s iteration_number=%d iteration_id=%s",
+                session_id,
+                iteration_number,
+                iteration_id,
+                extra={"session_id": session_id},
+            )
+
         # Create the Iteration node
         await self.services.graph.upsert_node(
             iteration_id,
@@ -88,13 +134,12 @@ async def _handle_provider_request(
                 "session_id": session_id,
                 "iteration_number": iteration_number,
                 "started_at": timestamp,
+                "iteration_scope": iteration_scope,
             },
         )
 
         # E06 (conditional): OrchestratorRun -[:HAS_PART {sst_semantic: 'CONTAINS'}]-> Iteration
-        execution_start_ts = self.services.data_layer_2.execution_start_ts
-        if execution_start_ts is not None:
-            orch_run_id = f"{session_id}::orch_run::{execution_start_ts}"
+        if orch_run_id is not None:
             await self.services.graph.upsert_edge(
                 orch_run_id,
                 iteration_id,
@@ -127,6 +172,12 @@ async def _handle_llm_request(self, data: dict[str, Any]) -> None:
                 "model": data.get("model"),
                 "message_count": data.get("message_count"),
                 "has_system": data.get("has_system"),
+                # BLOCKER-2: stamp independently of provider:request's own
+                # write -- an Iteration node must never be created/updated
+                # without a scope value, even if this write is the first one
+                # to ever reach the node (e.g. a dead-lettered provider:request
+                # whose cursor mutation nonetheless survived).
+                "iteration_scope": self._current_iteration_scope(),
             },
         )
 
@@ -157,6 +208,9 @@ async def _handle_llm_response(self, data: dict[str, Any]) -> None:
                 "usage_input": usage.get("input_tokens"),
                 "usage_output": usage.get("output_tokens"),
                 "usage_cache_write": usage.get("cache_creation_input_tokens"),
+                # BLOCKER-2: see _handle_llm_request -- same completeness
+                # rationale applies to this, the third of the three sites.
+                "iteration_scope": self._current_iteration_scope(),
             },
         )