PR3: Schema-version observability

CHANGELOG.md

@@ -5,6 +5,20 @@ 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.2.0]
+
+### Added
+
+- **Schema-version observability.** The server now records the graph data-model
+  version it wrote and reports drift on `/status`. A `SCHEMA_VERSION` constant
+  is the compiled model version; at startup the server writes a single
+  `:SchemaMeta{id:'singleton'}.schema_version` baseline (create-if-absent, off
+  the per-worker flush path). `GET /status` returns three fields:
+  `schema_version` (compiled), `graph_schema_version` (stored, or `null` when
+  the graph is unreachable or never baselined), and `schema_version_current`
+  (`true` in sync, `false` on mismatch, `null` when unknown). Advisory only —
+  reported, never gated or migrated; `/status` never raises on a read failure.
+
 ## [7.1.0]
 
 ### Added

context_intelligence_server/main.py

@@ -51,12 +51,18 @@
     build_bounded_neo4j_driver,
     count_untagged_nodes,
     ensure_neo4j_schema,
+    ensure_schema_version_baseline,
+    read_graph_schema_version,
 )
 from context_intelligence_server.registry import SessionRegistry
 from context_intelligence_server.routers.admin import router as admin_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.status import boot_state, build_status_response
+from context_intelligence_server.status import (
+    SCHEMA_VERSION,
+    boot_state,
+    build_status_response,
+)
 from context_intelligence_server.writer_lease import (
     WriterLeaseConflict,
     shutdown_lease_io,
@@ -325,6 +331,11 @@ async def _ensure_schema_ready() -> None:
         )
     app.state.schema_ready = True
     logger.info("lifespan_startup: Neo4j schema initialized")
+    # Record the graph's schema-version baseline once, AFTER the rest of the
+    # schema is established. Single-writer, startup-only: kept off the per-worker
+    # flush path. Never raises -- a transient failure leaves the marker absent,
+    # which /status surfaces as an unknown (never a false "in sync").
+    await ensure_schema_version_baseline(app.state.neo4j_driver)
 
 
 async def _crash_recovery_sweep_loop(interval: int, respawn_limit: int) -> None:
@@ -1040,11 +1051,33 @@ async def get_status(request: Request) -> dict[str, Any]:
         # Same contract: aggregate integers only, cheap (stat-only,
         # short-TTL cached) even with a huge spool.
         response["spool"] = await registry.queue_manager.spool_stats()
+        # Schema-version drift: the compiled model version vs the graph's stored
+        # :SchemaMeta version. read_graph_schema_version returns None when the
+        # graph is unreachable or the baseline was never written; an absent
+        # driver is treated the same way -- drift=None (unknown), never a false
+        # "in sync" and never a 500 (/status must never raise). Gated with the
+        # disk reads above so /status stays graph-read-free while booting.
+        _schema_driver = getattr(request.app.state, "neo4j_driver", None)
+        graph_schema_version = (
+            await read_graph_schema_version(_schema_driver)
+            if _schema_driver is not None
+            else None
+        )
+        response["schema_version"] = SCHEMA_VERSION
+        response["graph_schema_version"] = graph_schema_version
+        response["schema_version_current"] = (
+            None
+            if graph_schema_version is None
+            else graph_schema_version == SCHEMA_VERSION
+        )
     else:
         # While booting, /status performs zero disk reads. metrics/spool stay
         # present but null, so an absent key is never confused with a version skew.
         response["metrics"] = None
         response["spool"] = None
+        response["schema_version"] = SCHEMA_VERSION
+        response["graph_schema_version"] = None
+        response["schema_version_current"] = None
         response["status_detail"] = {"reason": "booting"}
     # Surface auth mode/admin capability so operators can confirm admin is
     # enabled without tailing logs -- boolean flags only, no credentials.

context_intelligence_server/neo4j_store.py

@@ -15,14 +15,15 @@
 import logging
 import re
 from collections.abc import Generator
-from datetime import datetime
+from datetime import UTC, datetime
 from typing import Any, LiteralString, cast
 
 from neo4j import AsyncGraphDatabase
 from neo4j import unit_of_work as _unit_of_work
 from neo4j.exceptions import DriverError, Neo4jError
 
 from context_intelligence_server.config import Neo4jClientConfig
+from context_intelligence_server.status import SCHEMA_VERSION
 
 _LOG = logging.getLogger(__name__)
 
@@ -927,6 +928,115 @@ async def _create_constraint(
         return fully_established
 
 
+async def ensure_schema_version_baseline(
+    driver: Any,
+    *,
+    database: str = "neo4j",
+) -> None:
+    """Create the :SchemaMeta uniqueness constraint and baseline singleton.
+
+    Baseline only -- create-if-absent, no comparison/migration. Call this
+    exactly once, from the lifespan startup handler, AFTER ``ensure_neo4j_schema``
+    has established the rest of the schema (indexes + uniqueness constraints).
+
+    Kept out of ``ensure_neo4j_schema`` on purpose: that runs on every
+    ``Neo4jGraphStore``'s first flush (once per worker, concurrently, on cold
+    start) and from ``run_repair``/``doctor --fix``. A SchemaMeta baseline write
+    is a single-writer, startup-only concern that must not fire per worker.
+
+    Ordering matters: the ``(:SchemaMeta).id`` uniqueness constraint is created
+    FIRST, then the singleton MERGE -- without the constraint, two concurrent
+    MERGEs on a fresh database can each create a ``{id: 'singleton'}`` node.
+
+    ``ON CREATE SET`` only: an existing node is left untouched. Reconciling a
+    stored ``schema_version`` against the running server's value is deferred;
+    the read path (``read_graph_schema_version``) and this write path stay
+    structurally separate so comparison/upgrade logic cannot creep in here.
+
+    O(1): one constraint DDL plus a MERGE on a fixed key -- never a scan. Any
+    ``Neo4jError``/``DriverError`` is logged and swallowed: a transient failure
+    on this passive data point must never crash boot.
+    """
+    try:
+        async with driver.session(database=database) as session:
+            try:
+                await session.run(
+                    "CREATE CONSTRAINT schemameta_id_unique IF NOT EXISTS "
+                    "FOR (m:SchemaMeta) REQUIRE m.id IS UNIQUE"
+                )
+            except (Neo4jError, DriverError) as exc:
+                if isinstance(exc, Neo4jError) and exc.code in _BENIGN_SCHEMA_CODES:
+                    _LOG.debug(
+                        "ensure_schema_version_baseline: SchemaMeta "
+                        "uniqueness constraint already present (benign "
+                        "concurrent-schema race, code=%s)",
+                        exc.code,
+                    )
+                else:
+                    _LOG.warning(
+                        "ensure_schema_version_baseline: could not create "
+                        "SchemaMeta uniqueness constraint; continuing "
+                        "without it: %s",
+                        exc,
+                    )
+
+            await session.run(
+                "MERGE (m:SchemaMeta {id: 'singleton'}) "
+                "ON CREATE SET m.schema_version = $schema_version, "
+                "m.last_updated = $now",
+                schema_version=SCHEMA_VERSION,
+                now=datetime.now(UTC).isoformat(),
+            )
+    except (Neo4jError, DriverError) as exc:
+        _LOG.warning(
+            "ensure_schema_version_baseline: could not write SchemaMeta "
+            "baseline singleton (connectivity error); continuing without "
+            "it: %s",
+            exc,
+        )
+
+
+async def read_graph_schema_version(
+    driver: Any,
+    *,
+    database: str = "neo4j",
+) -> int | None:
+    """Read-only: the STORED ``:SchemaMeta{id:'singleton'}.schema_version``.
+
+    Advisory drift signal only, not a guard. Read-only companion to
+    ``ensure_schema_version_baseline``, kept structurally apart from that write
+    path. It does NOT import or compare against ``SCHEMA_VERSION`` -- it only
+    reads back whatever is stored. ``GET /status`` compares the returned value
+    against ``status.SCHEMA_VERSION`` itself so a server/graph mismatch is
+    detectable; this function performs no comparison, gating, or migration.
+
+    Returns ``None`` when the singleton is absent (startup never ran the
+    baseline against this graph) or when the read fails for any reason (treated
+    as "unknown", never an error), mirroring the never-500-``/status`` contract.
+
+    O(1): a point lookup by the unique ``id`` key. Reads via ``async for``
+    (not ``.single()``) so it works against both the real async driver and the
+    test suite's mock session, which only implements async iteration.
+    """
+    try:
+        async with driver.session(database=database) as session:
+            result = await session.run(
+                "MATCH (m:SchemaMeta {id: 'singleton'}) "
+                "RETURN m.schema_version AS schema_version"
+            )
+            async for record in result:
+                value = record["schema_version"]
+                return int(value) if value is not None else None
+            return None
+    except Exception as exc:  # noqa: BLE001 - defensive: /status must never 500
+        _LOG.warning(
+            "read_graph_schema_version: could not read SchemaMeta singleton "
+            "(connectivity error?); returning None: %s",
+            exc,
+        )
+        return None
+
+
 def _serialized_row_size(value: Any) -> int:
     """Return a cheap conservative proxy for the serialized byte size of *value*.
 

context_intelligence_server/status.py

@@ -13,6 +13,11 @@
 # Resolved once at import time — never changes within a process lifetime.
 SERVER_VERSION: str = _pkg_version("context-intelligence-server")
 
+# Graph data-model version, distinct from the server release version above.
+# Bumped only when the node/edge schema changes; /status compares it against
+# the graph's stored :SchemaMeta.schema_version to surface drift.
+SCHEMA_VERSION: int = 1
+
 if TYPE_CHECKING:
     from context_intelligence_server.registry import SessionRegistry
 

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "context-intelligence-server"
-version = "7.1.0"
+version = "7.2.0"
 description = "Context Intelligence Server for Amplifier"
 requires-python = ">=3.11"
 dependencies = [