diff --git a/CHANGELOG.md b/CHANGELOG.md index 4fe7583..efff75b 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/context_intelligence_server/main.py b/context_intelligence_server/main.py index dfb9cf3..978ae00 100644 --- a/context_intelligence_server/main.py +++ b/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. diff --git a/context_intelligence_server/neo4j_store.py b/context_intelligence_server/neo4j_store.py index 6dad68a..803da74 100644 --- a/context_intelligence_server/neo4j_store.py +++ b/context_intelligence_server/neo4j_store.py @@ -15,7 +15,7 @@ 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 @@ -23,6 +23,7 @@ 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*. diff --git a/context_intelligence_server/status.py b/context_intelligence_server/status.py index 19f8fbc..d2370d0 100644 --- a/context_intelligence_server/status.py +++ b/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 diff --git a/pyproject.toml b/pyproject.toml index cc9105d..474c4b5 100644 --- a/pyproject.toml +++ b/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 = [ diff --git a/tests/neo4j/test_schema_version_baseline.py b/tests/neo4j/test_schema_version_baseline.py new file mode 100644 index 0000000..d26f5b3 --- /dev/null +++ b/tests/neo4j/test_schema_version_baseline.py @@ -0,0 +1,198 @@ +"""Neo4j integration proof for the SchemaMeta baseline singleton. + +BASELINE DATA POINTS ONLY: this proves ``ensure_schema_version_baseline``'s +``:SchemaMeta {id: 'singleton'}`` write is create-if-absent (``ON CREATE SET`` +only, no ``ON MATCH SET``) and O(1) -- calling it twice must leave exactly one +node in place with an UNCHANGED `last_updated`, proving the second call did +not clobber it. It also proves the uniqueness constraint on +``(:SchemaMeta).id`` is actually created, and that the constraint is what +makes the singleton race-free under real concurrency (N concurrent callers +still leave exactly one node) -- the whole point of this hardening +follow-up: moving the write off the per-worker-flush path onto a +single-writer startup call, backed by a uniqueness constraint so even a +violation of that single-writer invariant could never create a duplicate. + +Also proves ``ensure_neo4j_schema`` itself no longer writes SchemaMeta at +all -- that responsibility now belongs exclusively to +``ensure_schema_version_baseline``, called once from the lifespan startup +handler, never from the per-flush / doctor-repair paths that call +``ensure_neo4j_schema``. + +No comparison/upgrade/migration logic is exercised here; there is none to +exercise -- that is the point of this test. + +Run: uv run pytest tests/neo4j/test_schema_version_baseline.py -v -m neo4j +""" + +from __future__ import annotations + +import asyncio +from typing import Any + +import pytest +from context_intelligence_server.neo4j_store import ( + ensure_neo4j_schema, + ensure_schema_version_baseline, +) +from context_intelligence_server.status import SCHEMA_VERSION +from neo4j import AsyncGraphDatabase + + +async def _schema_meta_rows(driver: Any) -> list[Any]: + """Return all :SchemaMeta{id:'singleton'} rows (schema_version, last_updated).""" + async with driver.session() as session: + result = await session.run( + "MATCH (m:SchemaMeta {id: 'singleton'}) " + "RETURN m.schema_version AS schema_version, " + "m.last_updated AS last_updated" + ) + return [record async for record in result] + + +@pytest.mark.neo4j +class TestSchemaMetaBaselineIdempotence: + """``ensure_schema_version_baseline`` writes the singleton create-if-absent only.""" + + async def test_second_call_does_not_clobber_first( + self, neo4j_container: dict[str, Any] + ) -> None: + auth = (neo4j_container["user"], neo4j_container["password"]) + bolt = neo4j_container["bolt_url"] + + driver = AsyncGraphDatabase.driver(bolt, auth=auth) + try: + # First call: creates the singleton (and the uniqueness constraint). + await ensure_schema_version_baseline(driver) + + rows = await _schema_meta_rows(driver) + assert len(rows) == 1, ( + f"expected exactly one :SchemaMeta singleton after first call, " + f"got {len(rows)}" + ) + assert rows[0]["schema_version"] == SCHEMA_VERSION + first_last_updated = rows[0]["last_updated"] + assert first_last_updated is not None + + # Second call: must be a no-op on the existing node (ON CREATE only). + await ensure_schema_version_baseline(driver) + + rows_after = await _schema_meta_rows(driver) + assert len(rows_after) == 1, ( + f"expected exactly one :SchemaMeta node after second call " + f"(no duplicate created), got {len(rows_after)}" + ) + assert rows_after[0]["last_updated"] == first_last_updated, ( + "last_updated changed on the second call -- ON MATCH SET must " + "not be present; the singleton must be left untouched once it " + "exists" + ) + assert rows_after[0]["schema_version"] == SCHEMA_VERSION + finally: + await driver.close() + + async def test_uniqueness_constraint_exists( + self, neo4j_container: dict[str, Any] + ) -> None: + """The (:SchemaMeta).id uniqueness constraint is created and enforced.""" + auth = (neo4j_container["user"], neo4j_container["password"]) + bolt = neo4j_container["bolt_url"] + + driver = AsyncGraphDatabase.driver(bolt, auth=auth) + try: + await ensure_schema_version_baseline(driver) + + async with driver.session() as session: + result = await session.run("SHOW CONSTRAINTS") + constraints = [record async for record in result] + + schema_meta_constraints = [ + c + for c in constraints + if "SchemaMeta" in (c.get("labelsOrTypes") or []) + and "id" in (c.get("properties") or []) + ] + assert schema_meta_constraints, ( + "expected a uniqueness constraint on (:SchemaMeta).id to exist " + f"after ensure_schema_version_baseline; SHOW CONSTRAINTS returned: " + f"{constraints}" + ) + + # Belt-and-suspenders: attempting to create a second singleton node + # directly (bypassing MERGE) must be rejected by the constraint. + with pytest.raises(Exception): # noqa: B017 - Neo4jError subtype + async with driver.session() as session: + await session.run("CREATE (m:SchemaMeta {id: 'singleton'})") + finally: + await driver.close() + + async def test_concurrent_calls_create_exactly_one_node( + self, neo4j_container: dict[str, Any] + ) -> None: + """N concurrent baseline calls against the same DB leave exactly one node. + + This is the whole point of the uniqueness-constraint hardening: without + it, concurrent MERGEs on a fresh singleton key can each pass the + existence check and create divergent duplicate nodes. Fire many + concurrent calls (each on its own driver, mirroring independent + SessionWorker stores) and assert the constraint prevents any + duplication. + """ + auth = (neo4j_container["user"], neo4j_container["password"]) + bolt = neo4j_container["bolt_url"] + + n_concurrent = 20 + drivers = [ + AsyncGraphDatabase.driver(bolt, auth=auth) for _ in range(n_concurrent) + ] + try: + await asyncio.gather(*(ensure_schema_version_baseline(d) for d in drivers)) + + rows = await _schema_meta_rows(drivers[0]) + assert len(rows) == 1, ( + f"expected exactly one :SchemaMeta singleton after " + f"{n_concurrent} concurrent calls, got {len(rows)} -- the " + "uniqueness constraint should make concurrent creation race-free" + ) + assert rows[0]["schema_version"] == SCHEMA_VERSION + finally: + for d in drivers: + await d.close() + + +@pytest.mark.neo4j +class TestEnsureNeo4jSchemaNoLongerWritesSchemaMeta: + """``ensure_neo4j_schema`` must not touch :SchemaMeta at all (hardening follow-up). + + That responsibility moved exclusively to ``ensure_schema_version_baseline``, + called once from the lifespan startup handler -- NOT from the per-flush / + doctor-repair paths that call ``ensure_neo4j_schema``. If ``ensure_neo4j_schema`` + still created the singleton, it would fire redundantly (and concurrently) on + every SessionWorker's first flush. + """ + + async def test_ensure_neo4j_schema_does_not_create_schema_meta( + self, neo4j_container: dict[str, Any] + ) -> None: + auth = (neo4j_container["user"], neo4j_container["password"]) + bolt = neo4j_container["bolt_url"] + + driver = AsyncGraphDatabase.driver(bolt, auth=auth) + try: + # Remove any pre-existing singleton so this test is unambiguous + # regardless of what earlier tests in this (session-scoped) + # container have already written. + async with driver.session() as session: + await session.run( + "MATCH (m:SchemaMeta {id: 'singleton'}) DETACH DELETE m" + ) + + await ensure_neo4j_schema(driver) + + rows = await _schema_meta_rows(driver) + assert rows == [], ( + "ensure_neo4j_schema must not write the :SchemaMeta singleton " + f"-- that is ensure_schema_version_baseline's job now, but " + f"found {len(rows)} node(s)" + ) + finally: + await driver.close() diff --git a/tests/test_main.py b/tests/test_main.py index 4db529f..3f8e901 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -1414,6 +1414,89 @@ async def test_status_includes_neo4j_connected_false_when_no_driver( assert data["neo4j_connected"] is False +# --------------------------------------------------------------------------- +# /status schema-version drift reporting +# +# /status reports the compiled model version (SCHEMA_VERSION) alongside the +# graph's stored value and a drift flag: in-sync -> True, mismatch -> False, +# unknown (graph unreachable / baseline never written / no driver) -> None. +# read_graph_schema_version is patched so the reporting logic is exercised +# without a real graph. +# --------------------------------------------------------------------------- + + +async def test_status_schema_version_in_sync( + client: httpx.AsyncClient, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """graph version == SCHEMA_VERSION -> schema_version_current is True.""" + from context_intelligence_server.status import SCHEMA_VERSION + + monkeypatch.setattr( + main_module.app.state, "neo4j_driver", AsyncMock(), raising=False + ) + monkeypatch.setattr( + main_module, + "read_graph_schema_version", + AsyncMock(return_value=SCHEMA_VERSION), + ) + + response = await client.get("/status") + assert response.status_code == 200 + data = response.json() + assert data["schema_version"] == SCHEMA_VERSION + assert data["graph_schema_version"] == SCHEMA_VERSION + assert data["schema_version_current"] is True + + +async def test_status_schema_version_drift( + client: httpx.AsyncClient, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """graph version != SCHEMA_VERSION -> schema_version_current is False.""" + from context_intelligence_server.status import SCHEMA_VERSION + + monkeypatch.setattr( + main_module.app.state, "neo4j_driver", AsyncMock(), raising=False + ) + monkeypatch.setattr( + main_module, + "read_graph_schema_version", + AsyncMock(return_value=SCHEMA_VERSION + 1), + ) + + response = await client.get("/status") + assert response.status_code == 200 + data = response.json() + assert data["schema_version"] == SCHEMA_VERSION + assert data["graph_schema_version"] == SCHEMA_VERSION + 1 + assert data["schema_version_current"] is False + + +async def test_status_schema_version_unknown_when_absent( + client: httpx.AsyncClient, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """graph version None (unreachable / never baselined) -> current is None.""" + from context_intelligence_server.status import SCHEMA_VERSION + + monkeypatch.setattr( + main_module.app.state, "neo4j_driver", AsyncMock(), raising=False + ) + monkeypatch.setattr( + main_module, + "read_graph_schema_version", + AsyncMock(return_value=None), + ) + + response = await client.get("/status") + assert response.status_code == 200 + data = response.json() + assert data["schema_version"] == SCHEMA_VERSION + assert data["graph_schema_version"] is None + assert data["schema_version_current"] is None + + # --------------------------------------------------------------------------- # Concern B (council review) -- /status neo4j_query_connected field tests # --------------------------------------------------------------------------- diff --git a/uv.lock b/uv.lock index 439c0b4..7e613ea 100644 --- a/uv.lock +++ b/uv.lock @@ -233,7 +233,7 @@ wheels = [ [[package]] name = "context-intelligence-server" -version = "7.1.0" +version = "7.2.0" source = { editable = "." } dependencies = [ { name = "aiofiles" },