diff --git a/.env.example b/.env.example index 54ca1f4..1428919 100644 --- a/.env.example +++ b/.env.example @@ -18,8 +18,12 @@ NURON_AI_DB_HOST=127.0.0.1 NURON_AI_DB_PORT=5432 # --- Neo4j ------------------------------------------------------------------- -# Community edition has exactly one user, `neo4j`. +# Community edition has exactly one user, `neo4j`, and exactly one database -- no +# NEO4J_USERNAME/NEO4J_DATABASE var, nothing to configure. NEO4J_PASSWORD=changeme +# Host-published bolt port (see docker-compose.yml) -- nuron-ai isn't containerized yet, +# so it talks to Neo4j over the host network, not the `nuron` compose network. +NEO4J_URI=bolt://127.0.0.1:7687 # --- RustFS ------------------------------------------------------------------ RUSTFS_ACCESS_KEY=changeme diff --git a/docker-compose.yml b/docker-compose.yml index ee3919f..60150c1 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -62,9 +62,10 @@ services: NEO4J_AUTH: neo4j/${NEO4J_PASSWORD:?NEO4J_PASSWORD is required} volumes: - neo4j_data:/data - # ports: - # - "127.0.0.1:7474:7474" - # - "127.0.0.1:7687:7687" + ports: + # ponytail: host-published bolt until nuron-ai joins the `nuron` network (NU-005). HTTP + # browser UI (7474) stays commented out -- nothing here needs it. + - "127.0.0.1:7687:7687" networks: - nuron healthcheck: diff --git a/services/nuron-ai/pyproject.toml b/services/nuron-ai/pyproject.toml index 1a6b55d..accadb5 100644 --- a/services/nuron-ai/pyproject.toml +++ b/services/nuron-ai/pyproject.toml @@ -12,6 +12,7 @@ dependencies = [ "fsspec>=2026.7.0", "llama-cloud>=1.0", "llama-index-core==0.14.23", + "llama-index-graph-stores-neo4j==0.8.0", "llama-index-llms-openai-like==0.8.0", "openai>=2.54.0,<3", "psycopg[binary]>=3.3.5", @@ -30,7 +31,7 @@ dev = [ [[tool.mypy.overrides]] # These libraries ship no py.typed marker; nothing to check against. -module = ["defusedxml.*", "fsspec.*", "llama_index.llms.openai_like.*"] +module = ["defusedxml.*", "fsspec.*", "llama_index.llms.openai_like.*", "llama_index.graph_stores.neo4j.*", "neo4j.*"] ignore_missing_imports = true diff --git a/services/nuron-ai/src/nuron_ai/persist.py b/services/nuron-ai/src/nuron_ai/persist.py new file mode 100644 index 0000000..2bc6e6b --- /dev/null +++ b/services/nuron-ai/src/nuron_ai/persist.py @@ -0,0 +1,541 @@ +"""Persist compiled triples to Neo4j: alias matching, delta apply, refcounted provenance (NU-009). + +Diffs a document's compiled graph against the live Neo4j state -- its own prior contribution +(for drop_ref) and whatever already exists under its proposed keys (for update/unchanged, so an +auto-joined entity another document already wrote is never re-embedded) -- then applies +plan_delta's classification with a refcounted provenance set per node (docs/tracer-bullet-01.md +"persist", ADR-0003, ADR-0004). Compiler.py extracts and enriches; this module is what actually +calls insert/upsert against a live graph store. + +**Deliberately does not go through `PropertyGraphIndex(embed_kg_nodes=True)`** (the construction +ADR-0003 shows): that index's own `insert_nodes()` skips a node entirely once its id already +exists in the store, with no update path -- exactly wrong for a re-approval, which must still +overwrite a changed node's content (A6). Driving `Neo4jPropertyGraphStore.upsert_nodes()` directly +with our own plan_delta-classified embedding is what makes "leave unchanged alone, update in +place" possible at all; ADR-0003's point -- embeddings live on the graph node, one store, one +write -- holds either way. + +**Relations are re-upserted, never diffed or dropped.** A relationship-only edit (A6's own +trigger case) leaves a prior version's edges in place alongside the new ones. plan_delta has no +edge-level concept to apply (core.py's contract is nodes only), and edges carry no provenance ref +of their own in this schema -- ponytail: stale edges accumulate, revisit with an edge provenance +scheme if a real re-approval workload needs relationships to disappear too. +""" + +import logging +import os +import time +import uuid +from collections.abc import Callable, Sequence +from typing import Any + +import psycopg +from llama_index.core.graph_stores.types import EntityNode, Relation +from llama_index.graph_stores.neo4j import Neo4jPropertyGraphStore +from openai import OpenAI # nosemgrep +from psycopg import sql + +from nuron_ai import db +from nuron_ai.core import NodeDelta, plan_delta, resolve_key + +logger = logging.getLogger(__name__) + +_LEASE_SECONDS = 600.0 +_MAX_ATTEMPTS = 5 +_RETRY_DELAY_SECONDS = 60.0 +_POLL_DELAY_SECONDS = 5.0 + +# Structural type for the configured embedding model -- lets tests use a stub with no live endpoint. +Embedder = Callable[[str], list[float]] + +_PROVENANCE_REFS_PROPERTY = "provenance_refs" +_EMBEDDING_MODEL_PROPERTY = "embedding_model_id" +_EMBEDDING_DIMENSIONS_PROPERTY = "embedding_dimensions" +_DISPLAY_NAME_PROPERTY = "display_name" +# Bookkeeping/identity properties this module owns -- never part of a node's compared content. +_RESERVED_PROPERTIES = frozenset( + { + _PROVENANCE_REFS_PROPERTY, + _EMBEDDING_MODEL_PROPERTY, + _EMBEDDING_DIMENSIONS_PROPERTY, + _DISPLAY_NAME_PROPERTY, + "name", + } +) + + +class EmbeddingConfigurationMismatchError(ValueError): + """An existing node's embedding model or dimension count differs from config -- terminal until a re-embedding migration (ADR-0004).""" + + +def build_store_from_env() -> Neo4jPropertyGraphStore: + """Connects to Neo4j via NEO4J_URI/NEO4J_PASSWORD -- community edition has exactly one user, `neo4j`.""" + return Neo4jPropertyGraphStore( + username="neo4j", + password=os.environ["NEO4J_PASSWORD"], + url=os.environ["NEO4J_URI"], + ) + + +def build_embedder_from_env() -> tuple[Embedder, str, int]: + """Builds an OpenAI-compatible embedder from OPENAI_*/EMBEDDING_DIMENSIONS; returns it with the model id and dimension count ADR-0004 requires stamped on every node.""" + # Guardrails suppressed: this client only embeds. Any text generation or summarization + # (planned) must go through a Guardrails-wrapped client -- generated output needs validation. + client = OpenAI(base_url=os.environ["OPENAI_BASE_URL"], api_key=os.environ["OPENAI_API_KEY"]) # nosemgrep + model = os.environ["OPENAI_EMBEDDING_MODEL"] + dimensions = int(os.environ["EMBEDDING_DIMENSIONS"]) + + def embed(text: str) -> list[float]: + """Calls the configured OpenAI-compatible embeddings endpoint for one text.""" + response = client.embeddings.create(model=model, input=text, dimensions=dimensions) + return response.data[0].embedding + + return embed, model, dimensions + + +def _content(label: str, display_name: str, extra_properties: dict[str, Any]) -> dict[str, Any]: + """Builds one node's comparable content: label, display name, and its non-bookkeeping properties.""" + content: dict[str, Any] = {"label": label, "display_name": display_name} + for key, value in extra_properties.items(): + content[key] = value + return content + + +def _content_from_compiled_node(node: dict[str, Any]) -> dict[str, Any]: + """Strips the per-document provenance dict from a compiled_graph node before comparison. + + Also drops None-valued properties (compiler.py always sets author/author_source/timestamp on + a DECISION node, even to None when unmatched): Neo4j has no null property value, so a + round-tripped node never has that key at all (`upsert_nodes`'s own `apoc.map.clean` strips it + on write, `get()`'s `remove_empty_values` strips it on read). Comparing against a dict that + still carries `{"author": None}` would never equal the round-tripped one, marking every such + node "update" forever and re-embedding it on every persist -- the exact spend this module + exists to avoid. + """ + extra: dict[str, Any] = {} + for key, value in node["properties"].items(): + if key == "provenance" or value is None: + continue + extra[key] = value + return _content(node["label"], node["name"], extra) + + +def _content_from_existing_node(entity: EntityNode) -> dict[str, Any]: + """Strips this module's own bookkeeping properties from a node read back from Neo4j.""" + extra: dict[str, Any] = {} + for key, value in entity.properties.items(): + if key in _RESERVED_PROPERTIES: + continue + extra[key] = value + display_name = entity.properties.get(_DISPLAY_NAME_PROPERTY, entity.name) + return _content(entity.label, display_name, extra) + + +def _embedding_text(label: str, display_name: str) -> str: + """Deterministic text embedded for a node -- retrieval quality is A1's concern, not this one's.""" + return f"{display_name} ({label})" + + +def _node_properties(content: dict[str, Any], model_id: str, dimensions: int) -> dict[str, Any]: + """Builds a node's content properties plus embedding bookkeeping -- never the provenance set. + + Provenance is mutated separately, atomically, in Neo4j itself (see `_add_ref_atomic`/ + `_release_ref_atomic`): `upsert_nodes`'s own Cypher does `SET e += ...`, a merge, so it can + only ever add or overwrite keys, never clear one on a concurrent writer's behalf -- exactly + wrong for a value multiple documents contribute refs to independently. + """ + properties: dict[str, Any] = {_DISPLAY_NAME_PROPERTY: content["display_name"]} + for key, value in content.items(): + if key in {"label", "display_name"}: + continue + properties[key] = value + properties[_EMBEDDING_MODEL_PROPERTY] = model_id + properties[_EMBEDDING_DIMENSIONS_PROPERTY] = dimensions + return properties + + +# Both provenance queries recompute the ref set from whatever is actually stored at the moment +# they run, inside one Neo4j transaction each (`execute_query` auto-commits per call) -- never +# from a Python-held snapshot. Two documents racing to touch the same auto-joined node therefore +# each apply their own add/remove against the graph's current state, not against a stale read; +# core.py's release_ref/plan_delta already proved the *rule* (remove one ref, the node dies only +# when the set is empty) -- this is that same rule, just evaluated where the mutation actually +# happens so it can't be lost to a lost update. +_RELEASE_REF_QUERY = """ + MATCH (n {id: $node_key}) + WITH n, [x IN coalesce(n.provenance_refs, []) WHERE x <> $ref] AS remaining + SET n.provenance_refs = remaining + WITH n, remaining + WHERE size(remaining) = 0 + DETACH DELETE n + """ + +_ADD_REF_QUERY = """ + MATCH (n {id: $node_key}) + WITH n, [x IN coalesce(n.provenance_refs, []) WHERE x <> $ref_to_remove] AS without_old + WITH n, CASE WHEN $ref_to_add IN without_old THEN without_old ELSE without_old + $ref_to_add END AS refs + SET n.provenance_refs = refs + """ + +# Dynamic property access (`n[$key]`) parameterizes the property *name*, not just its value -- +# property keys here can trace back to LLM-extracted entity properties (compiler.py's +# `dict(node.properties)`), so they are not a fixed, trusted vocabulary. String-formatting a key +# straight into `REMOVE n.` would be a Cypher injection surface; this keeps every key bound. +_REMOVE_STALE_PROPERTIES_QUERY = """ + MATCH (n {id: $node_key}) + UNWIND $stale_keys AS stale_key + REMOVE n[stale_key] + """ + + +def _release_ref_atomic(store: Neo4jPropertyGraphStore, node_key: str, ref: str) -> None: + """Atomically drops one provenance ref, deleting the node in the same query if it was the last.""" + store.structured_query(_RELEASE_REF_QUERY, param_map={"node_key": node_key, "ref": ref}) + + +def _add_ref_atomic( + store: Neo4jPropertyGraphStore, node_key: str, ref_to_add: str, ref_to_remove: str | None +) -> None: + """Atomically adds one provenance ref, optionally dropping another (a re-approval's old ref).""" + store.structured_query( + _ADD_REF_QUERY, + param_map={"node_key": node_key, "ref_to_add": ref_to_add, "ref_to_remove": ref_to_remove or ""}, + ) + + +def _remove_stale_properties(store: Neo4jPropertyGraphStore, node_key: str, stale_keys: frozenset[str]) -> None: + """Clears properties the new content no longer carries -- `upsert_nodes`'s own SET += never would.""" + if not stale_keys: + return + store.structured_query( + _REMOVE_STALE_PROPERTIES_QUERY, param_map={"node_key": node_key, "stale_keys": sorted(stale_keys)} + ) + + +def _apply_node_delta( + store: Neo4jPropertyGraphStore, + embed: Embedder, + model_id: str, + dimensions: int, + *, + delta: NodeDelta, + content: dict[str, Any] | None, + existing: EntityNode | None, + content_hash: str, + previous_content_hash: str | None, + is_previous_contribution: bool, +) -> None: + """Applies one node's classification: create, update in place, leave alone, or release a ref.""" + if delta.action == "drop_ref": + if existing is None: + logger.warning("drop_ref for %s but the node no longer exists in Neo4j", delta.node_key) + return + old_refs = frozenset(existing.properties.get(_PROVENANCE_REFS_PROPERTY, [])) + if previous_content_hash is None or previous_content_hash not in old_refs: + # Already released by an earlier attempt at this same persist (the worker crashed or + # lost its lease between this Neo4j write and the Postgres bookkeeping that follows + # it) -- retrying is expected to be a no-op here, not a second release (tracer-bullet-01 + # "the Neo4j write... is self-correcting"). This is a snapshot-based pre-check only -- + # the actual release below is still atomic against live state either way. + return + _release_ref_atomic(store, delta.node_key, previous_content_hash) + return + + # add/update/unchanged: plan_delta only ever emits these actions for a key it read out of + # `current_content`, so `content` is always present here -- trust that contract. + assert content is not None + + ref_to_remove = previous_content_hash if is_previous_contribution else None + + if delta.action == "unchanged": + old_refs = frozenset(existing.properties.get(_PROVENANCE_REFS_PROPERTY, [])) if existing else frozenset() + already_has_ref = content_hash in old_refs and (ref_to_remove is None or ref_to_remove not in old_refs) + if already_has_ref: + return # true no-op: same content, same contributor -- nothing to write (A5). Snapshot-based + # fast path only; a redundant or (rarely) skipped write here never loses a ref, since + # the write itself, when it does happen, is the atomic query below either way. + _add_ref_atomic(store, delta.node_key, content_hash, ref_to_remove) + return + + # add or update: (re)write content and (re)compute the embedding -- the only branch that + # spends an embedding call -- then reconcile the provenance ref and any properties the new + # content dropped, each as its own atomic step. + embedding_text = _embedding_text(content["label"], content["display_name"]) + node = EntityNode( + name=delta.node_key, + label=content["label"], + properties=_node_properties(content, model_id, dimensions), + embedding=embed(embedding_text), + ) + store.upsert_nodes([node]) + + if existing is not None: + stale_keys = frozenset(_content_from_existing_node(existing)) - frozenset(content) + _remove_stale_properties(store, delta.node_key, stale_keys) + + _add_ref_atomic(store, delta.node_key, content_hash, ref_to_remove) + + +def _apply_relations( + store: Neo4jPropertyGraphStore, + relations: Sequence[dict[str, Any]], + resolved_by_natural_key: dict[str, str], +) -> None: + """Re-upserts every relation this document contributes, resolved through the same alias keys as its nodes.""" + resolved_relations: list[Relation] = [] + for relation in relations: + source_key = resolved_by_natural_key.get(relation["source_key"]) + target_key = resolved_by_natural_key.get(relation["target_key"]) + if source_key is None or target_key is None: + logger.warning("relation %s references a node key missing from this compile", relation) + continue + resolved_relations.append( + Relation( + label=relation["label"], + source_id=source_key, + target_id=target_key, + properties=dict(relation["properties"]), + ) + ) + if resolved_relations: + store.upsert_relations(resolved_relations) + + +def persist_compiled_graph( + store: Neo4jPropertyGraphStore, + embed: Embedder, + model_id: str, + dimensions: int, + *, + content_hash: str, + aliases: dict[tuple[str, str], str], + previous_content_hash: str | None, + previous_node_keys: frozenset[str], + compiled_graph: dict[str, Any], +) -> frozenset[str]: + """Applies one document's compiled graph to Neo4j; returns the node keys it now contributes to.""" + current_content: dict[str, dict[str, Any]] = {} + resolved_by_natural_key: dict[str, str] = {} + for node in compiled_graph["nodes"]: + resolved_key = resolve_key(node["name"], node["label"], aliases) + resolved_by_natural_key[node["key"]] = resolved_key + current_content[resolved_key] = _content_from_compiled_node(node) + + lookup_keys = set(current_content) | set(previous_node_keys) + existing_by_key: dict[str, EntityNode] = {} + if lookup_keys: + for entity in store.get(ids=list(lookup_keys)): + existing_by_key[entity.id] = entity + + for entity in existing_by_key.values(): + existing_model_id = entity.properties.get(_EMBEDDING_MODEL_PROPERTY) + if existing_model_id is not None and existing_model_id != model_id: + raise EmbeddingConfigurationMismatchError( + f"node {entity.id!r} was embedded with model {existing_model_id!r}, not the " + f"configured {model_id!r} -- re-embedding is a migration, not a write (ADR-0004)" + ) + existing_dimensions = entity.properties.get(_EMBEDDING_DIMENSIONS_PROPERTY) + if existing_dimensions is not None and existing_dimensions != dimensions: + raise EmbeddingConfigurationMismatchError( + f"node {entity.id!r} was embedded at {existing_dimensions} dimensions, not the " + f"configured {dimensions} -- re-embedding is a migration, not a write (ADR-0004)" + ) + + prior_content: dict[str, dict[str, Any]] = {} + for key, entity in existing_by_key.items(): + prior_content[key] = _content_from_existing_node(entity) + + deltas = plan_delta(prior_content, current_content) + for delta in deltas: + _apply_node_delta( + store, + embed, + model_id, + dimensions, + delta=delta, + content=current_content.get(delta.node_key), + existing=existing_by_key.get(delta.node_key), + content_hash=content_hash, + previous_content_hash=previous_content_hash, + is_previous_contribution=delta.node_key in previous_node_keys, + ) + + _apply_relations(store, compiled_graph["relations"], resolved_by_natural_key) + + return frozenset(current_content) + + +_RETRY_SET_SQL = sql.SQL( + """ + attempt_count = attempt_count + 1, + next_attempt_at = now() + %(retry_delay_seconds)s * interval '1 second', + state = CASE WHEN attempt_count + 1 >= %(max_attempts)s + THEN 'failed'::nuron_ai.pipeline_state + ELSE state END + """ +) + +_RETRY_PARAMS: dict[str, Any] = {"retry_delay_seconds": _RETRY_DELAY_SECONDS, "max_attempts": _MAX_ATTEMPTS} + +_PERSISTED_SET_SQL = sql.SQL("state = 'persisted', attempt_count = 0, next_attempt_at = NULL") + +_FAILED_SET_SQL = sql.SQL("state = 'failed'") + +_AWAITING_MERGE_CONFIRM_SET_SQL = sql.SQL( + "state = 'awaiting_merge_confirm', attempt_count = 0, next_attempt_at = NULL" +) + + +def merge_gate_pending(conn: psycopg.Connection, worker_id: str) -> bool: + """Claims one compiled row and advances it to awaiting_merge_confirm.""" + # ponytail: no merge-candidate detection exists yet (NU-010), so every compiled row is an + # auto-join needing no human confirmation (CONTEXT.md "Auto-join") and passes straight + # through. NU-010 replaces this pass-through: rows with candidates stay blocked here. + claimed = db.claim(conn, worker_id, "compiled", lease_seconds=_LEASE_SECONDS) + if claimed is None: + return False + + digest, *_rest, lease_token = claimed + db.release(conn, digest, worker_id, lease_token, _AWAITING_MERGE_CONFIRM_SET_SQL, {}) + return True + +# Versions are numbered per original_filename, not per content_hash (review.py's approve(): +# "Version lineage follows original_filename... a re-ingested file lands under a new content_hash +# every time"). rs_current.version - 1 is therefore the row this content_hash's own prior +# ingestion produced, joined back through original_filename to find its content_hash. +_PREVIOUS_CONTENT_HASH_QUERY = """ + SELECT rs_prev.content_hash + FROM nuron_ai.reviewed_sources rs_current + JOIN nuron_ai.documents d_current ON d_current.content_hash = rs_current.content_hash + JOIN nuron_ai.reviewed_sources rs_prev ON rs_prev.version = rs_current.version - 1 + JOIN nuron_ai.documents d_prev ON d_prev.content_hash = rs_prev.content_hash + AND d_prev.original_filename = d_current.original_filename + WHERE rs_current.content_hash = %(content_hash)s + ORDER BY rs_current.version DESC + LIMIT 1 + """ + + +def persist_pending( + conn: psycopg.Connection, + store: Neo4jPropertyGraphStore, + embed: Embedder, + model_id: str, + dimensions: int, + worker_id: str, +) -> bool: + """Claims one awaiting_merge_confirm row and persists its compiled graph to Neo4j.""" + # ponytail: claims unconditionally -- rows only reach awaiting_merge_confirm via + # merge_gate_pending, which is where NU-010's real merge-candidate blocking must land. Until + # a human-confirm path exists, every row here is an auto-join (CONTEXT.md "Auto-join"). + claimed = db.claim(conn, worker_id, "awaiting_merge_confirm", lease_seconds=_LEASE_SECONDS) + if claimed is None: + return False + + digest, original_filename, *_rest, lease_token = claimed + + graph_row = conn.execute( + "SELECT compiled_graph FROM nuron_ai.documents WHERE content_hash = %(content_hash)s", + {"content_hash": digest}, + ).fetchone() + reviewed_source_row = conn.execute( + "SELECT id FROM nuron_ai.reviewed_sources WHERE content_hash = %(content_hash)s " + "ORDER BY version DESC LIMIT 1", + {"content_hash": digest}, + ).fetchone() + previous_row = conn.execute(_PREVIOUS_CONTENT_HASH_QUERY, {"content_hash": digest}).fetchone() + alias_rows = conn.execute( + "SELECT alias_key, label, survivor_key FROM nuron_ai.entity_aliases" + ).fetchall() + conn.commit() + + if graph_row is None or reviewed_source_row is None: + logger.error( + "awaiting_merge_confirm row %s has no compiled_graph or reviewed_sources row -- " + "should be impossible", + digest, + ) + db.release(conn, digest, worker_id, lease_token, _RETRY_SET_SQL, _RETRY_PARAMS) + return True + compiled_graph = graph_row[0] + reviewed_source_id = reviewed_source_row[0] + previous_content_hash = previous_row[0] if previous_row is not None else None + + previous_node_keys: frozenset[str] = frozenset() + if previous_content_hash is not None: + provenance_rows = conn.execute( + "SELECT node_key FROM nuron_ai.node_provenance WHERE content_hash = %(content_hash)s", + {"content_hash": previous_content_hash}, + ).fetchall() + conn.commit() + previous_node_keys = frozenset(row[0] for row in provenance_rows) + + aliases: dict[tuple[str, str], str] = {} + for alias_key, label, survivor_key in alias_rows: + aliases[(alias_key, label)] = survivor_key + + try: + current_keys = persist_compiled_graph( + store, + embed, + model_id, + dimensions, + content_hash=digest, + aliases=aliases, + previous_content_hash=previous_content_hash, + previous_node_keys=previous_node_keys, + compiled_graph=compiled_graph, + ) + except EmbeddingConfigurationMismatchError as err: + logger.warning("persist failed permanently for %s (%s): %s", digest, original_filename, err) + db.release(conn, digest, worker_id, lease_token, _FAILED_SET_SQL, {}) + return True + except Exception as err: + logger.warning("persist failed for %s (%s): %s", digest, original_filename, err) + db.release(conn, digest, worker_id, lease_token, _RETRY_SET_SQL, _RETRY_PARAMS) + return True + + if previous_content_hash is not None: + conn.execute( + "DELETE FROM nuron_ai.node_provenance WHERE content_hash = %(content_hash)s", + {"content_hash": previous_content_hash}, + ) + for node_key in current_keys: + conn.execute( + """ + INSERT INTO nuron_ai.node_provenance (node_key, content_hash, reviewed_source_id) + VALUES (%(node_key)s, %(content_hash)s, %(reviewed_source_id)s) + ON CONFLICT (node_key, content_hash) DO NOTHING + """, + {"node_key": node_key, "content_hash": digest, "reviewed_source_id": reviewed_source_id}, + ) + conn.commit() + + db.release(conn, digest, worker_id, lease_token, _PERSISTED_SET_SQL, {}) + return True + + +def main() -> None: + """Runs the Persist worker forever: gates compiled rows, then persists awaiting_merge_confirm rows.""" + logging.basicConfig(level=logging.INFO) + worker_id = uuid.uuid4().hex + store = build_store_from_env() + embed, model_id, dimensions = build_embedder_from_env() + + while True: + try: + with db.from_env() as conn: + while True: + gated = merge_gate_pending(conn, worker_id) + persisted = persist_pending(conn, store, embed, model_id, dimensions, worker_id) + if not (gated or persisted): + time.sleep(_POLL_DELAY_SECONDS) + except Exception: + logger.exception( + "persist worker failed; retrying after %.0f seconds", _RETRY_DELAY_SECONDS + ) + time.sleep(_RETRY_DELAY_SECONDS) + + +if __name__ == "__main__": + main() diff --git a/services/nuron-ai/tests/test_persist.py b/services/nuron-ai/tests/test_persist.py new file mode 100644 index 0000000..204689b --- /dev/null +++ b/services/nuron-ai/tests/test_persist.py @@ -0,0 +1,649 @@ +"""Tests for persist.py: Neo4j alias matching, delta apply, refcounted provenance (NU-009).""" + +from typing import Any +from unittest.mock import MagicMock + +import psycopg +import pytest +from llama_index.core.graph_stores.types import EntityNode + +from nuron_ai import persist +from nuron_ai.core import natural_key +from nuron_ai.persist import ( + EmbeddingConfigurationMismatchError, + merge_gate_pending, + persist_compiled_graph, + persist_pending, +) + +_MODEL_ID = "test-model" +_DIMENSIONS = 3 + + +def _node(label: str, name: str, extra_properties: dict[str, Any] | None = None) -> dict[str, Any]: + """Builds a compiled_graph node dict as compiler.py's enrich_and_serialize would.""" + properties: dict[str, Any] = {"provenance": {"content_hash": "a" * 64, "reviewed_source_id": 1}} + if extra_properties: + properties.update(extra_properties) + return {"key": natural_key(name, label), "label": label, "name": name, "properties": properties} + + +def _graph(nodes: list[dict[str, Any]], relations: list[dict[str, Any]] | None = None) -> dict[str, Any]: + """Builds a compiled_graph dict as compiler.py's compile_graph would return it.""" + return {"nodes": nodes, "relations": relations or [], "decisions": []} + + +def _existing_entity( + node_key: str, + label: str, + *, + display_name: str, + refs: list[str], + model_id: str = _MODEL_ID, + dimensions: int = _DIMENSIONS, + extra_properties: dict[str, Any] | None = None, +) -> EntityNode: + """Builds an EntityNode as Neo4jPropertyGraphStore.get() would return it (id-keyed by name).""" + properties: dict[str, Any] = { + "display_name": display_name, + "provenance_refs": refs, + "embedding_model_id": model_id, + "embedding_dimensions": dimensions, + } + if extra_properties: + properties.update(extra_properties) + return EntityNode(name=node_key, label=label, properties=properties) + + +def _embed_that_must_not_run(text: str) -> list[float]: + """An Embedder stub for tests where a call means the "don't re-embed" guarantee broke.""" + raise AssertionError(f"this node must not be re-embedded, but embed() was called with {text!r}") + + +def _structured_query_calls(store: MagicMock, marker: str) -> list[Any]: + """Returns every store.structured_query() call whose Cypher text contains marker.""" + matches = [] + for one_call in store.structured_query.call_args_list: + if marker in str(one_call.args[0]): + matches.append(one_call) + return matches + + +def _persist( + store: Any, + *, + content_hash: str, + aliases: dict[tuple[str, str], str] | None = None, + previous_content_hash: str | None = None, + previous_node_keys: frozenset[str] = frozenset(), + compiled_graph: dict[str, Any], + embed: persist.Embedder = _embed_that_must_not_run, +) -> frozenset[str]: + """Calls persist_compiled_graph with this test module's fixed model id/dimensions.""" + return persist_compiled_graph( + store, + embed, + _MODEL_ID, + _DIMENSIONS, + content_hash=content_hash, + aliases=aliases or {}, + previous_content_hash=previous_content_hash, + previous_node_keys=previous_node_keys, + compiled_graph=compiled_graph, + ) + + +# -- persist_compiled_graph: add --------------------------------------------------- + + +def test_persist_adds_a_brand_new_node_and_embeds_it() -> None: + store = MagicMock() + store.get.return_value = [] + embed_calls: list[str] = [] + + def embed(text: str) -> list[float]: + embed_calls.append(text) + return [0.1, 0.2] + + current_keys = _persist( + store, + content_hash="a" * 64, + compiled_graph=_graph([_node("ENTITY", "session store")]), + embed=embed, + ) + + assert current_keys == frozenset({"session store:ENTITY"}) + assert embed_calls == ["session store (ENTITY)"] + [written] = store.upsert_nodes.call_args.args[0] + assert written.id == "session store:ENTITY" + assert written.label == "ENTITY" + assert written.embedding == [0.1, 0.2] + assert written.properties["embedding_model_id"] == _MODEL_ID + assert written.properties["embedding_dimensions"] == _DIMENSIONS + assert written.properties["display_name"] == "session store" + # provenance is never part of the content upsert -- it's added atomically, separately, so a + # concurrent writer touching the same node's ref set can never be clobbered by this write. + assert "provenance_refs" not in written.properties + [add_ref_call] = _structured_query_calls(store, "SET n.provenance_refs = refs") + assert add_ref_call.kwargs["param_map"] == { + "node_key": "session store:ENTITY", + "ref_to_add": "a" * 64, + "ref_to_remove": "", + } + + +# -- persist_compiled_graph: update (A6) ------------------------------------------- + + +def test_persist_updates_changed_content_and_swaps_the_provenance_ref() -> None: + prior_hash = "a" * 64 + new_hash = "b" * 64 + existing = _existing_entity( + "drop session store:DECISION", + "DECISION", + display_name="drop session store", + refs=[prior_hash], + extra_properties={"author": "Basuru", "timestamp": "2026-05-14"}, + ) + store = MagicMock() + store.get.return_value = [existing] + + current_keys = _persist( + store, + content_hash=new_hash, + previous_content_hash=prior_hash, + previous_node_keys=frozenset({"drop session store:DECISION"}), + compiled_graph=_graph( + [_node("DECISION", "drop session store", {"author": "Basuru", "timestamp": "2026-05-15"})] + ), + embed=lambda text: [0.9], + ) + + assert current_keys == frozenset({"drop session store:DECISION"}) + [written] = store.upsert_nodes.call_args.args[0] + assert written.properties["timestamp"] == "2026-05-15" + assert written.embedding == [0.9] + [add_ref_call] = _structured_query_calls(store, "SET n.provenance_refs = refs") + assert add_ref_call.kwargs["param_map"] == { + "node_key": "drop session store:DECISION", + "ref_to_add": new_hash, + "ref_to_remove": prior_hash, + } + # author/timestamp are present on both versions -- nothing to clear. + assert _structured_query_calls(store, "REMOVE n[stale_key]") == [] + + +def test_persist_removes_a_property_the_new_version_no_longer_carries() -> None: + # upsert_nodes's own Cypher is a merge (`SET e += ...`): it would leave a stale `author` on + # the node forever if we relied on it alone once the new content drops that property. + prior_hash = "a" * 64 + new_hash = "b" * 64 + existing = _existing_entity( + "undated decision:DECISION", + "DECISION", + display_name="undated decision", + refs=[prior_hash], + extra_properties={"author": "Basuru", "timestamp": "2026-05-14"}, + ) + store = MagicMock() + store.get.return_value = [existing] + + _persist( + store, + content_hash=new_hash, + previous_content_hash=prior_hash, + previous_node_keys=frozenset({"undated decision:DECISION"}), + # the reviewer removed the author/date on re-approval -- compiler.py still emits the + # keys, just set to None, which _content_from_compiled_node drops entirely. + compiled_graph=_graph([_node("DECISION", "undated decision", {"author": None, "timestamp": None})]), + embed=lambda text: [0.5], + ) + + [remove_call] = _structured_query_calls(store, "REMOVE n[stale_key]") + assert remove_call.kwargs["param_map"] == { + "node_key": "undated decision:DECISION", + "stale_keys": ["author", "timestamp"], + } + + +def test_persist_treats_a_none_valued_property_as_absent_so_it_still_counts_as_unchanged() -> None: + # compiler.py always sets author/author_source/timestamp on a DECISION, even to None when + # the reviewed source had no attribution -- Neo4j has no null property, so a round-tripped + # node never carries that key. Comparing against a dict that still has {"author": None} + # would wrongly call this "update" and re-embed it on every single persist. + existing = _existing_entity( + "undated decision:DECISION", "DECISION", display_name="undated decision", refs=["a" * 64] + ) + store = MagicMock() + store.get.return_value = [existing] + + current_keys = _persist( + store, + content_hash="a" * 64, + compiled_graph=_graph( + [_node("DECISION", "undated decision", {"author": None, "author_source": None, "timestamp": None})] + ), + ) + + assert current_keys == frozenset({"undated decision:DECISION"}) + store.upsert_nodes.assert_not_called() # true no-op: unchanged, and the ref is already there. + store.structured_query.assert_not_called() + + +# -- persist_compiled_graph: unchanged, cross-document auto-join (A9) ------------- + + +def test_persist_leaves_an_auto_joined_node_unchanged_but_adds_the_new_ref() -> None: + hash_a = "a" * 64 + hash_b = "b" * 64 + existing = _existing_entity( + "session store:ENTITY", "ENTITY", display_name="session store", refs=[hash_a] + ) + store = MagicMock() + store.get.return_value = [existing] + + current_keys = _persist( + store, + content_hash=hash_b, + compiled_graph=_graph([_node("ENTITY", "session store")]), + ) + + assert current_keys == frozenset({"session store:ENTITY"}) + # unchanged: no content rewrite at all, only the provenance set changes. + store.upsert_nodes.assert_not_called() + [add_ref_call] = _structured_query_calls(store, "SET n.provenance_refs = refs") + assert add_ref_call.kwargs["param_map"] == { + "node_key": "session store:ENTITY", + "ref_to_add": hash_b, + "ref_to_remove": "", + } + + +def test_persist_is_a_true_no_op_when_the_ref_already_present() -> None: + hash_a = "a" * 64 + existing = _existing_entity( + "session store:ENTITY", "ENTITY", display_name="session store", refs=[hash_a] + ) + store = MagicMock() + store.get.return_value = [existing] + + _persist(store, content_hash=hash_a, compiled_graph=_graph([_node("ENTITY", "session store")])) + + store.upsert_nodes.assert_not_called() + store.structured_query.assert_not_called() + + +# -- persist_compiled_graph: drop_ref / refcounting (A7) --------------------------- + + +def test_persist_deletes_a_node_when_its_last_ref_is_released() -> None: + prior_hash = "a" * 64 + existing = _existing_entity( + "session store:ENTITY", "ENTITY", display_name="session store", refs=[prior_hash] + ) + store = MagicMock() + store.get.return_value = [existing] + + _persist( + store, + content_hash="c" * 64, + previous_content_hash=prior_hash, + previous_node_keys=frozenset({"session store:ENTITY"}), + compiled_graph=_graph([]), # this version no longer contributes the entity + ) + + # The conditional delete-when-empty is inside the atomic Cypher itself (verified directly + # against a live Neo4j instance -- see the docstring on _RELEASE_REF_QUERY); a mock can only + # confirm the release was routed correctly, not which branch the database took. + [release_call] = _structured_query_calls(store, "DETACH DELETE n") + assert release_call.kwargs["param_map"] == {"node_key": "session store:ENTITY", "ref": prior_hash} + store.upsert_nodes.assert_not_called() + store.delete.assert_not_called() + + +def test_persist_survives_ref_release_when_another_document_still_holds_a_ref() -> None: + prior_hash = "a" * 64 + other_hash = "b" * 64 + existing = _existing_entity( + "session store:ENTITY", + "ENTITY", + display_name="session store", + refs=[prior_hash, other_hash], + ) + store = MagicMock() + store.get.return_value = [existing] + + _persist( + store, + content_hash="c" * 64, + previous_content_hash=prior_hash, + previous_node_keys=frozenset({"session store:ENTITY"}), + compiled_graph=_graph([]), + ) + + store.delete.assert_not_called() + store.upsert_nodes.assert_not_called() + [release_call] = _structured_query_calls(store, "DETACH DELETE n") + assert release_call.kwargs["param_map"] == {"node_key": "session store:ENTITY", "ref": prior_hash} + + +def test_persist_drop_ref_is_idempotent_when_a_prior_attempt_already_released_it() -> None: + # The worker can crash (or lose its lease) after this Neo4j write lands but before the + # Postgres bookkeeping that follows it commits -- a retry must not re-release a ref that's + # already gone (tracer-bullet-01.md: the Neo4j write is "self-correcting" against live state). + other_hash = "b" * 64 + existing = _existing_entity( + "session store:ENTITY", "ENTITY", display_name="session store", refs=[other_hash] + ) + store = MagicMock() + store.get.return_value = [existing] + + _persist( + store, + content_hash="c" * 64, + previous_content_hash="a" * 64, # already absent from existing's refs + previous_node_keys=frozenset({"session store:ENTITY"}), + compiled_graph=_graph([]), + ) + + store.delete.assert_not_called() + store.upsert_nodes.assert_not_called() + store.structured_query.assert_not_called() + + +# -- ADR-0004: pinned embedding model/dimensions ----------------------------------- + + +def test_persist_rejects_a_node_embedded_with_a_different_model_id() -> None: + existing = _existing_entity( + "session store:ENTITY", + "ENTITY", + display_name="session store", + refs=["a" * 64], + model_id="golden-test-double-v1", + ) + store = MagicMock() + store.get.return_value = [existing] + + with pytest.raises(EmbeddingConfigurationMismatchError, match="golden-test-double-v1"): + _persist( + store, + content_hash="b" * 64, + compiled_graph=_graph([_node("ENTITY", "session store")]), + ) + store.upsert_nodes.assert_not_called() + + +def test_persist_rejects_a_node_embedded_at_a_different_dimension_count() -> None: + existing = _existing_entity( + "session store:ENTITY", + "ENTITY", + display_name="session store", + refs=["a" * 64], + dimensions=1536, + ) + store = MagicMock() + store.get.return_value = [existing] + + with pytest.raises(EmbeddingConfigurationMismatchError, match="1536"): + _persist( + store, + content_hash="b" * 64, + compiled_graph=_graph([_node("ENTITY", "session store")]), + ) + store.upsert_nodes.assert_not_called() + + +# -- resolve_key / aliases ---------------------------------------------------------- + + +def test_persist_matches_through_a_confirmed_alias_survivor() -> None: + survivor = "session store:ENTITY" + aliases = {("sessions table", "ENTITY"): survivor} + # Same display name as the document below contributes -- this test is about resolve_key + # finding the survivor node at all, not about the update/unchanged content classification. + existing = _existing_entity(survivor, "ENTITY", display_name="Sessions Table", refs=["a" * 64]) + store = MagicMock() + store.get.return_value = [existing] + + current_keys = _persist( + store, + content_hash="a" * 64, + aliases=aliases, + compiled_graph=_graph([_node("ENTITY", "Sessions Table")]), + ) + + assert current_keys == frozenset({survivor}) + store.get.assert_called_once() + assert set(store.get.call_args.kwargs["ids"]) == {survivor} + + +# -- relations ----------------------------------------------------------------------- + + +def test_persist_upserts_relations_resolved_through_the_same_keys_as_their_nodes() -> None: + store = MagicMock() + store.get.return_value = [] + decision = _node("DECISION", "drop session store") + entity = _node("ENTITY", "rate limiter") + relation = { + "label": "AFFECTS", + "source_key": decision["key"], + "target_key": entity["key"], + "properties": {}, + } + + _persist( + store, + content_hash="a" * 64, + compiled_graph=_graph([decision, entity], [relation]), + embed=lambda text: [0.0], + ) + + [written] = store.upsert_relations.call_args.args[0] + assert written.source_id == "drop session store:DECISION" + assert written.target_id == "rate limiter:ENTITY" + + +def test_persist_skips_a_relation_referencing_a_node_missing_from_this_compile( + caplog: pytest.LogCaptureFixture, +) -> None: + store = MagicMock() + store.get.return_value = [] + decision = _node("DECISION", "drop session store") + relation = { + "label": "AFFECTS", + "source_key": decision["key"], + "target_key": "nonexistent:ENTITY", + "properties": {}, + } + + with caplog.at_level("WARNING"): + _persist( + store, + content_hash="a" * 64, + compiled_graph=_graph([decision], [relation]), + embed=lambda text: [0.0], + ) + + store.upsert_relations.assert_not_called() + assert "missing from this compile" in caplog.text + + +# -- persist_pending: claim / release semantics, mocked Postgres ------------------ + + +def _claimed_row(digest: str, filename: str, *, lease_token: int = 3) -> tuple[Any, ...]: + """A db.claim RETURNING tuple: hash, filename, title, author, author_source, date, tags, body, lease_token.""" + return (digest, filename, None, None, None, None, [], "body", lease_token) + + +def _make_conn( + *, + digest: str, + filename: str, + compiled_graph: dict[str, Any] | None, + previous_content_hash: str | None = None, + previous_reviewed_source_id: int = 7, + previous_node_keys: list[str] | None = None, +) -> MagicMock: + """A mocked psycopg.Connection dispatching persist_pending's queries by SQL substring.""" + conn = MagicMock(spec=psycopg.Connection) + + def execute(query: object, params: object = None) -> MagicMock: + text = str(query) + result = MagicMock() + result.rowcount = 1 + if "SET claimed_by" in text: + result.fetchone.return_value = _claimed_row(digest, filename) + elif "SELECT compiled_graph" in text: + result.fetchone.return_value = None if compiled_graph is None else (compiled_graph,) + elif "rs_prev.content_hash" in text: + result.fetchone.return_value = None if previous_content_hash is None else (previous_content_hash,) + elif "SELECT node_key FROM nuron_ai.node_provenance" in text: + result.fetchall.return_value = [(key,) for key in (previous_node_keys or [])] + elif "FROM nuron_ai.reviewed_sources" in text: + result.fetchone.return_value = (previous_reviewed_source_id,) + elif "entity_aliases" in text: + result.fetchall.return_value = [] + return result + + conn.execute.side_effect = execute + return conn + + +def _calls_matching(conn: MagicMock, substring: str) -> list[Any]: + """Returns every conn.execute() call whose SQL text contains substring.""" + matches = [] + for one_call in conn.execute.call_args_list: + if substring in str(one_call.args[0]): + matches.append(one_call) + return matches + + +def test_persist_pending_returns_false_when_the_queue_is_empty() -> None: + conn = MagicMock(spec=psycopg.Connection) + conn.execute.return_value.fetchone.return_value = None + + assert persist_pending(conn, MagicMock(), _embed_that_must_not_run, _MODEL_ID, _DIMENSIONS, "worker-1") is False + + +def test_persist_pending_persists_a_first_version_and_transitions_to_persisted() -> None: + digest = "a" * 64 + compiled_graph = _graph([_node("ENTITY", "session store")]) + conn = _make_conn(digest=digest, filename="b.md", compiled_graph=compiled_graph) + store = MagicMock() + store.get.return_value = [] + + assert persist_pending(conn, store, lambda text: [0.0], _MODEL_ID, _DIMENSIONS, "worker-1") is True + + insert_calls = _calls_matching(conn, "INSERT INTO nuron_ai.node_provenance") + assert len(insert_calls) == 1 + assert insert_calls[0].args[1] == { + "node_key": "session store:ENTITY", + "content_hash": digest, + "reviewed_source_id": 7, + } + assert _calls_matching(conn, "DELETE FROM nuron_ai.node_provenance") == [] + release_call = conn.execute.call_args_list[-1] + assert "state = 'persisted'" in str(release_call.args[0]) + + +def test_persist_pending_releases_the_old_versions_provenance_on_reapproval() -> None: + digest = "b" * 64 + prior_hash = "a" * 64 + compiled_graph = _graph([_node("ENTITY", "session store")]) + conn = _make_conn( + digest=digest, + filename="a.md", + compiled_graph=compiled_graph, + previous_content_hash=prior_hash, + previous_reviewed_source_id=9, + previous_node_keys=["session store:ENTITY"], + ) + store = MagicMock() + store.get.return_value = [ + _existing_entity("session store:ENTITY", "ENTITY", display_name="session store", refs=[prior_hash]) + ] + + assert persist_pending(conn, store, lambda text: [0.0], _MODEL_ID, _DIMENSIONS, "worker-1") is True + + delete_calls = _calls_matching(conn, "DELETE FROM nuron_ai.node_provenance") + assert delete_calls[0].args[1] == {"content_hash": prior_hash} + + +def test_persist_pending_retries_when_persist_compiled_graph_fails(monkeypatch: pytest.MonkeyPatch) -> None: + digest = "a" * 64 + compiled_graph = _graph([_node("ENTITY", "session store")]) + conn = _make_conn(digest=digest, filename="b.md", compiled_graph=compiled_graph) + monkeypatch.setattr( + persist, + "persist_compiled_graph", + MagicMock(side_effect=ConnectionError("neo4j unreachable")), + ) + + assert persist_pending(conn, MagicMock(), _embed_that_must_not_run, _MODEL_ID, _DIMENSIONS, "worker-1") is True + + release_call = conn.execute.call_args_list[-1] + assert "attempt_count = attempt_count + 1" in str(release_call.args[0]) + + +def test_persist_pending_retries_when_the_compiled_graph_row_is_missing( + caplog: pytest.LogCaptureFixture, +) -> None: + digest = "a" * 64 + conn = _make_conn(digest=digest, filename="b.md", compiled_graph=None) + + with caplog.at_level("ERROR"): + result = persist_pending(conn, MagicMock(), _embed_that_must_not_run, _MODEL_ID, _DIMENSIONS, "worker-1") + assert result is True + + assert any(record.levelname == "ERROR" and digest in record.message for record in caplog.records) + release_call = conn.execute.call_args_list[-1] + assert "attempt_count = attempt_count + 1" in str(release_call.args[0]) + + +def test_persist_pending_fails_terminally_on_an_embedding_configuration_mismatch( + caplog: pytest.LogCaptureFixture, +) -> None: + digest = "b" * 64 + compiled_graph = _graph([_node("ENTITY", "session store")]) + conn = _make_conn(digest=digest, filename="b.md", compiled_graph=compiled_graph) + store = MagicMock() + store.get.return_value = [ + _existing_entity( + "session store:ENTITY", "ENTITY", display_name="session store", refs=["a" * 64], dimensions=1536 + ) + ] + + with caplog.at_level("WARNING"): + assert persist_pending(conn, store, _embed_that_must_not_run, _MODEL_ID, _DIMENSIONS, "worker-1") is True + + release_sql = str(conn.execute.call_args_list[-1].args[0]) + assert "state = 'failed'" in release_sql + assert "attempt_count" not in release_sql + assert any(record.levelname == "WARNING" and digest in record.message for record in caplog.records) + store.upsert_nodes.assert_not_called() + + +# -- merge_gate_pending: compiled -> awaiting_merge_confirm ---------------------- + + +def test_merge_gate_pending_returns_false_when_nothing_is_compiled() -> None: + conn = MagicMock(spec=psycopg.Connection) + conn.execute.return_value.fetchone.return_value = None + + assert merge_gate_pending(conn, "worker-1") is False + + +def test_merge_gate_pending_claims_compiled_and_releases_to_awaiting_merge_confirm() -> None: + conn = MagicMock(spec=psycopg.Connection) + conn.execute.return_value.fetchone.return_value = _claimed_row("c" * 64, "c.md") + conn.execute.return_value.rowcount = 1 + + assert merge_gate_pending(conn, "worker-1") is True + + claim_call, release_call = conn.execute.call_args_list + assert claim_call.args[1]["state"] == "compiled" + assert "state = 'awaiting_merge_confirm'" in str(release_call.args[0]) + assert release_call.args[1]["content_hash"] == "c" * 64 diff --git a/services/nuron-ai/uv.lock b/services/nuron-ai/uv.lock index 096ad44..a21d168 100644 --- a/services/nuron-ai/uv.lock +++ b/services/nuron-ai/uv.lock @@ -1058,6 +1058,19 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/ec/d5/05d61f34c01c6578fb758d0a3ddef58d36c6ffa9a9f84a5c9a16262ad94d/llama_index_core-0.14.23-py3-none-any.whl", hash = "sha256:6a54d267826732a8507f81df40785b107f7592af20f451a39a59005147caf84c", size = 11924908, upload-time = "2026-06-24T19:35:52.833Z" }, ] +[[package]] +name = "llama-index-graph-stores-neo4j" +version = "0.8.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "llama-index-core" }, + { name = "neo4j" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/6e/be/9c3a82e05277e2e53977af975be4b5da256d3925e092cc928432ecbc3400/llama_index_graph_stores_neo4j-0.8.0.tar.gz", hash = "sha256:4046b32ffd09fe6cc9c385ef1193aa9ba41aba369dc358abb835b5a3a3e1e7c0", size = 17380, upload-time = "2026-08-31T15:54:08.754Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/8f/bf/65762164a0b9b06627145b22fa5ba5eb9bfd2598ff2d3b33eff6d0dbf3c6/llama_index_graph_stores_neo4j-0.8.0-py3-none-any.whl", hash = "sha256:367aaab7296f684f5354e22bc837307293b7581b23f83fb3abc8ef9ed8681ab2", size = 18589, upload-time = "2026-08-31T15:54:07.959Z" }, +] + [[package]] name = "llama-index-instrumentation" version = "0.6.0" @@ -1348,6 +1361,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/79/7b/2c79738432f5c924bef5071f933bcc9efd0473bac3b4aa584a6f7c1c8df8/mypy_extensions-1.1.0-py3-none-any.whl", hash = "sha256:1be4cccdb0f2482337c4743e60421de3a356cd97508abadd57d47403e94f5505", size = 4963, upload-time = "2025-04-22T14:54:22.983Z" }, ] +[[package]] +name = "neo4j" +version = "5.28.6" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "pytz" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/77/ea/a17b0df723422cb3124bd45e6570d9677eafa38ffa207c3d12f601cbb90e/neo4j-5.28.6.tar.gz", hash = "sha256:224020cb649517cba1b76bf94129ccf84de30e7005932b5ab0cdd9cb566d55b2", size = 233391, upload-time = "2026-09-15T08:47:07.57Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/cd/9c/ec2a2f4b9fbb6c93d51c785a8c4e4cdca280009b7ad0821ff145e6cb02a5/neo4j-5.28.6-py3-none-any.whl", hash = "sha256:5454b51e0a1de870e7d0cc233b8897bb8b3a273fe32458ccbcde9d5a604f7437", size = 313892, upload-time = "2026-09-15T08:47:04.604Z" }, +] + [[package]] name = "nest-asyncio" version = "1.6.0" @@ -1464,6 +1489,7 @@ dependencies = [ { name = "fsspec" }, { name = "llama-cloud" }, { name = "llama-index-core" }, + { name = "llama-index-graph-stores-neo4j" }, { name = "llama-index-llms-openai-like" }, { name = "openai" }, { name = "psycopg", extra = ["binary"] }, @@ -1482,6 +1508,7 @@ requires-dist = [ { name = "fsspec", specifier = ">=2026.7.0" }, { name = "llama-cloud", specifier = ">=1.0" }, { name = "llama-index-core", specifier = "==0.14.23" }, + { name = "llama-index-graph-stores-neo4j", specifier = "==0.8.0" }, { name = "llama-index-llms-openai-like", specifier = "==0.8.0" }, { name = "openai", specifier = ">=2.54.0,<3" }, { name = "psycopg", extras = ["binary"], specifier = ">=3.3.5" }, @@ -1899,6 +1926,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/ec/57/56b9bcc3c9c6a792fcbaf139543cee77261f3651ca9da0c93f5c1221264b/python_dateutil-2.9.0.post0-py2.py3-none-any.whl", hash = "sha256:a8b2bc7bffae282281c8140a97d3aa9c14da0b136dfe83f850eea9a5f7470427", size = 229892, upload-time = "2024-03-01T18:36:18.57Z" }, ] +[[package]] +name = "pytz" +version = "2026.4" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/b0/ed/fa23d28713004418bbf2407127f80f385bfb0bf4e677bef02c25c27b00ee/pytz-2026.4.tar.gz", hash = "sha256:464303645bafafd72418898368b2429458f709cf1eb6a15372fbcc396b64da63", size = 318444, upload-time = "2026-09-24T03:58:40.443Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/d5/11/e3df59614bebe0ca50c905442603eda64ae7ac19811d4e4f74aa2d51ca85/pytz-2026.4-py2.py3-none-any.whl", hash = "sha256:9d514388fbc89ca0833203464272ac485b8828568ab73532f5020f17e892a0ff", size = 506747, upload-time = "2026-09-24T03:58:38.629Z" }, +] + [[package]] name = "pyyaml" version = "6.0.3"