diff --git a/datamind/capabilities/graph/providers/__init__.py b/datamind/capabilities/graph/providers/__init__.py index 4118048..94f82f0 100644 --- a/datamind/capabilities/graph/providers/__init__.py +++ b/datamind/capabilities/graph/providers/__init__.py @@ -2,5 +2,6 @@ from __future__ import annotations from . import networkx_store # noqa: F401 +from . import terminusdb_store # noqa: F401 __all__: list[str] = [] diff --git a/datamind/capabilities/graph/providers/networkx_store.py b/datamind/capabilities/graph/providers/networkx_store.py index cad35bc..acdc877 100644 --- a/datamind/capabilities/graph/providers/networkx_store.py +++ b/datamind/capabilities/graph/providers/networkx_store.py @@ -59,7 +59,7 @@ def __init__( }, ) - def schema(self) -> dict[str, Any]: + async def schema(self) -> dict[str, Any]: """Observed vocabulary, not a restrictive ontology.""" from collections import Counter with self._state_lock: diff --git a/datamind/capabilities/graph/providers/terminusdb_store.py b/datamind/capabilities/graph/providers/terminusdb_store.py new file mode 100644 index 0000000..c36dd1c --- /dev/null +++ b/datamind/capabilities/graph/providers/terminusdb_store.py @@ -0,0 +1,1242 @@ +"""TerminusDB-backed graph storage with versioned reads and rollback support. + +The provider persists graph entities and edges in TerminusDB, while a local +SQLite history records the DataMind ingest operations associated with each +TerminusDB commit. +""" +from __future__ import annotations + +import asyncio +import difflib +import json +import os +import re +import threading +import sqlite3 +import shutil +import subprocess +import time +import httpx + +from filelock import FileLock +from pathlib import Path +from typing import Any, Sequence +from urllib.parse import urlsplit + +from datamind.core.logging import get_logger +from datamind.core.protocols import Edge, Entity, GraphPath, GraphTriple +from datamind.core.registry import graph_registry + +_log = get_logger("graph.terminusdb") + + +@graph_registry.register("terminusdb") +class TerminusDBGraphStore: + + def __init__( + self, + *, + dsn: str, + password: str, + persist_path: str | Path, + branch: str = "main", + ) -> None: + self._dsn = dsn + self._password = password + self._path = Path(persist_path) + self._path.mkdir(parents=True, exist_ok=True) + self._db = "kg_default" + self._team = "admin" + self._branch = branch + self._terminusdb_docker_home = None + self._init_terminusdb() + self._client = self._get_client() + self._schema() + self._dirty = False + self._mutation_revision = 0 + self._state_lock = threading.RLock() + self._persist_lock = asyncio.Lock() + self._persist_worker_lock = threading.Lock() + + self._history_path = self._path / "version_history.sqlite3" + with sqlite3.connect(self._history_path) as conn: + conn.execute("CREATE TABLE IF NOT EXISTS history (database TEXT, version INTEGER, payload TEXT, PRIMARY KEY(database, version))") + + _log.info( + "graph_loaded", + extra={ + "path": str(self._path), + }, + ) + + def _is_terminusdb_running(self) -> bool: + """Check whether TerminusDB is already available.""" + try: + response = httpx.get( + f"{self._dsn.rstrip('/')}/api/ok", + timeout=2.0, + ) + return response.is_success + except httpx.HTTPError: + return False + + def _init_terminusdb(self): + """Start a managed local TerminusDB when no configured server responds.""" + if not self._dsn: + raise ValueError( + "Missing TerminusDB DSN. Configure DATAMIND__GRAPH__DSN." + ) + if self._is_terminusdb_running(): + return + + parsed = urlsplit(self._dsn) + if parsed.scheme not in {"http", "https"} or not parsed.hostname: + raise ValueError(f"Invalid TerminusDB DSN: {self._dsn!r}") + if parsed.hostname not in {"127.0.0.1", "localhost", "::1"}: + raise RuntimeError( + f"Cannot connect to remote TerminusDB at {self._dsn}; " + "DataMind only auto-starts a server for localhost DSNs" + ) + if parsed.scheme != "http" or parsed.path not in {"", "/"}: + raise RuntimeError( + "Managed local TerminusDB requires a base DSN such as " + "http://127.0.0.1:6363" + ) + if not self._password: + raise ValueError( + "Missing TerminusDB password. Configure DATAMIND__GRAPH__PASSWORD." + ) + if any(char in self._password for char in "\r\n"): + raise ValueError("TerminusDB password must not contain newlines") + + repo_url = "https://github.com/terminusdb/terminusdb" + home = Path.home() / ".datamind" / "terminusdb" + home = home.expanduser().resolve() + home.parent.mkdir(parents=True, exist_ok=True) + + # Multiple server workers may construct the provider concurrently. + with FileLock(str(home.parent / "terminusdb-start.lock"), timeout=120): + if self._is_terminusdb_running(): + return + if not shutil.which("docker"): + if not shutil.which("docker"): + raise RuntimeError( + "Docker is not installed or not in PATH. " + "Install Docker to use the default setup, or install TerminusDB " + "from source code following https://terminusdb.com/docs/install-terminusdb-from-source-code/ " + "and make sure the `terminusdb` binary is available in PATH." + ) + if not shutil.which("git"): + raise RuntimeError("Git is not installed or not in PATH") + subprocess.run( + ["docker", "compose", "version"], + check=True, + capture_output=True, + text=True, + ) + + if not (home / ".git").exists(): + if home.exists() and any(home.iterdir()): + raise RuntimeError( + f"{home} exists but is not a TerminusDB checkout" + ) + subprocess.run( + ["git", "clone", "--depth", "1", repo_url, str(home)], + check=True, + ) + + compose_env = os.environ.copy() + compose_env.update( + { + "TERMINUSDB_SERVER_DB_PATH": str(self._path.resolve()), + "TERMINUSDB_SERVER_PORT": str(parsed.port or 6363), + "TERMINUSDB_ADMIN_PASS": self._password, + } + ) + subprocess.run( + ["docker", "compose", "up", "-d"], + cwd=home, + env=compose_env, + check=True, + ) + deadline = time.monotonic() + 60 + while time.monotonic() < deadline: + if self._is_terminusdb_running(): + _log.info("terminusdb_ready", extra={"dsn": self._dsn}) + self._terminusdb_docker_home = home + return + time.sleep(2) + raise RuntimeError( + f"TerminusDB failed to become ready within 60 seconds at {self._dsn}. " + f"Run 'cd {home} && docker compose logs' for details." + ) + + def _get_client(self): + from terminusdb_client import Client + if not self._dsn: + raise ValueError("Missing TerminusDB DSN. Please configure DATAMIND__GRAPH__DSN.") + + if not self._password: + raise ValueError("Missing TerminusDB password. Please configure DATAMIND__GRAPH__PASSWORD.") + + try: + client = Client(self._dsn) + client.connect( + team=self._team, + user=self._team, + key=self._password, + server_url=self._dsn, + ) + except Exception as e: + raise RuntimeError( + f"TerminusDB connection failed: " + f"dsn={self._dsn}, " + f"error_type={type(e).__name__}, " + f"error={e}" + ) from e + + if not client.has_database(self._db, team=self._team): + client.create_database( + self._db, + label=self._db, + description="Graph of DataMind stores in TerminusDB", + include_schema=False, + ) + + client.connect( + team=self._team, + db=self._db, + user=self._team, + key=self._password, + server_url=self._dsn, + branch=self._branch, + ) + return client + + def _normalize(self, schema: dict) -> str: + return json.dumps( + schema, + sort_keys=True, + ensure_ascii=False, + separators=(",", ":"), + ) + + def _schema(self) -> None: + schema_docs = [ + { + "@type": "Class", + "@id": "entity", + "label": "xsd:string", + "outgoing_edge": { + "@type": "Set", + "@class": "GraphEdge", + }, + }, + { + "@type": "Class", + "@id": "GraphEdge", + "subject": "entity", + "object": "entity", + "relation": "xsd:string", + "key": "xsd:string", + "weight": "xsd:double", + "source": "xsd:string", + "properties": "sys:JSON", + "incoming_edge": { + "@type": "Optional", + "@class": "entity", + }, + }, + ] + + to_sync = [] + + for schema in schema_docs: + doc_id = schema["@id"] + + if not self._has_doc(doc_id, graph_type="schema"): + to_sync.append(schema) + continue + + existing = self._client.get_document( + doc_id, + graph_type="schema", + ) + if self._normalize(existing) != self._normalize(schema): + to_sync.append(schema) + + if to_sync: + self._client.replace_document( + to_sync, + graph_type="schema", + create=True, + commit_msg="Synchronize graph schema", + ) + + for schema in schema_docs: + doc_id = schema["@id"] + if not self._has_doc(doc_id, graph_type="schema"): + raise RuntimeError( + f"TerminusDB schema registration failed: {doc_id}" + ) + + def _has_doc(self, doc_id: str, graph_type=None): + if graph_type is None: + graph_type = 'instance' + + client = self._client + client._check_connection() + + response = client._session.get( + client._documents_url(), + headers=client._default_headers, + params={ + "id": doc_id, + "graph_type": graph_type, + }, + auth=client._auth(), + ) + if response.status_code == 200: + return True + if response.status_code == 404: + try: + body = response.json() + except ValueError: + return False + error = body.get("api:error", {}) + if error.get("@type") == "api:DocumentNotFound": + return False + + response.raise_for_status() + response.raise_for_status() + return False + + async def aclose(self) -> None: + """Release this store without stopping the shared database service.""" + if not self._is_terminusdb_running() or not self._terminusdb_docker_home: + # manually close + return + + process = await asyncio.create_subprocess_exec( + "docker", + "compose", + "stop", + cwd=self._terminusdb_docker_home, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + + try: + stdout, stderr = await asyncio.wait_for( + process.communicate(), + timeout=60, + ) + except asyncio.TimeoutError: + process.kill() + await process.wait() + raise TimeoutError("Timed out stopping TerminusDB") from None + + if process.returncode != 0: + raise RuntimeError( + f"Failed to stop TerminusDB: " + f"{stderr.decode(errors='replace').strip()}" + ) + + # ------------------------------------------------------------ mutation + + def _upsert_documents(self, documents, commit_msg): + from terminusdb_client.errors import DatabaseError + + to_insert = [] + to_replace = [] + for doc in documents: + doc_id = doc["@id"] + if not self._has_doc(doc_id): + to_insert.append(doc) + continue + existing = self._client.get_document(doc_id) + + if self._normalize(existing) != self._normalize(doc): + to_replace.append(doc) + + if to_insert: + self._client.insert_document(to_insert, commit_msg=commit_msg) + if to_replace: + self._client.replace_document(to_replace, commit_msg=commit_msg) + + return { + "inserted": len(to_insert), + "replaced": len(to_replace), + } + + def _edge_id(self, t): + profile_managed = bool((t.properties or {}).get("_profile_managed")) + origin = "profile" if profile_managed else (t.source or "runtime") + edge_key = f"{t.relation}\x1f{origin}" + subject = ( + t.subject[len(t.subject_type):].lstrip("/") + if t.subject.startswith(t.subject_type) + else t.subject + ) + object = ( + t.object[len(t.object_type):].lstrip("/") + if t.object.startswith(t.object_type) + else t.object + ) + return f"GraphEdge/{subject}_{object}_{edge_key}", edge_key + + async def upsert_triples(self, triples: Sequence[GraphTriple]) -> None: + from terminusdb_client.woqlquery import WOQLQuery + added_triples, updated_triples = [], [] + with self._state_lock: + for t in triples: + # Nodes + nodes = [] + node_replace = False + edge_replace = False + for side, node_id, node_type in ( + ("subject", t.subject, t.subject_type), + ("object", t.object, t.object_type), + ): + entity_id = f"entity/{node_id}" + node = {"@type": node_type, "@id": entity_id, "label": node_id} + result = self._upsert_documents([node], commit_msg=f"Add Node {entity_id}") + node_replace = bool(result["replaced"]) + nodes.append(node) + + # Profile snapshots and runtime writes have separate identities; + # exact duplicates within either origin overwrite deterministically. + edge_id, edge_key = self._edge_id(t) + subject_id = f"entity/{t.subject}" + object_id = f"entity/{t.object}" + edge_doc = { + "@type": "GraphEdge", + "@id": edge_id, + # TerminusDB requires ``@type: @id`` to distinguish a + # document reference from an embedded document. + "subject": {"@id": subject_id, "@type": "@id"}, + "object": {"@id": object_id, "@type": "@id"}, + "relation": t.relation, + "key": edge_key, + "weight": float(t.confidence), + "source": t.source or "", + "properties": {f"p_{k}": v for k, v in (t.properties or {}).items()}, + } + + result = self._upsert_documents([edge_doc], commit_msg=f"Add Edge {t.subject} --{t.relation}--> {t.object}") + node_replace = bool(result["replaced"]) + WOQLQuery().add_triple( + nodes[0]["@id"], + "outgoing_edge", + edge_doc["@id"], + ).execute(self._client) + + WOQLQuery().add_triple( + edge_doc["@id"], + "incoming_edge", + nodes[1]["@id"], + ).execute(self._client) + if node_replace or edge_replace: + updated_triples.append(t) + else: + added_triples.append(t) + self._mutation_revision += 1 + self._dirty = True + return self.get_triples_info(added=added_triples, updated=updated_triples) + + def _validate_key(self, doc, key, value=None): + props = doc.get("properties") or {} + if key == "profile": + return bool(props.get("p__profile_managed")) + elif key == "source": + if not value: + return False + source_path = str(props.get("p__source_path") or "") + return ( + source_path == value + or re.fullmatch( + re.escape(value) + r"#part=\d+", + source_path, + ) is not None + ) + elif key == "lineage": + return props.get("p__lineage_root") == value + return True + + def _get_triples(self, key="profile", value=None) -> list[GraphTriple]: + with self._state_lock: + docs = self._client.get_documents_by_type( + "GraphEdge", + as_list=True, + ) + + result = [] + + for doc in docs: + if not self._validate_key(doc, key, value): continue + props = doc.get("properties") or {} + subject = doc.get("subject") + obj = doc.get("object") + if isinstance(subject, dict): + subject = subject.get("@id") + if isinstance(obj, dict): + obj = obj.get("@id") + if not subject or not obj: + continue + result.append( + GraphTriple( + subject=subject, + relation=doc["relation"], + object=obj, + subject_type="entity", + object_type="entity", + confidence=float(doc.get("weight", 1.0)), + source=doc.get("source", ""), + properties=props, + ) + ) + return result + + def _diff_triples(self, previous_triples, triples): + def edge_identity(t: GraphTriple) -> tuple: + source_path = getattr(t, "source_path", "") or "" + + return ( + t.subject, + t.relation, + t.object, + source_path, + ) + + def edge_payload(t: GraphTriple) -> str: + payload = { + "confidence": float(t.confidence), + "source": getattr(t, "source", "") or "", + "properties": dict(t.properties or {}), + } + + return json.dumps( + payload, + sort_keys=True, + ensure_ascii=False, + separators=(",", ":"), + default=str, + ) + old = {edge_identity(t): t for t in previous_triples} + new = {edge_identity(t): t for t in triples} + + added = [new[k] for k in new.keys() - old.keys()] + removed = [old[k] for k in old.keys() - new.keys()] + + updated = [ + new[k] + for k in old.keys() & new.keys() + if edge_payload(old[k]) != edge_payload(new[k]) + ] + return added, updated, removed + + def _delete_edge_document_and_links(self, edge_doc: dict) -> None: + from terminusdb_client.woqlquery import WOQLQuery + edge_id = edge_doc["@id"] + subject = edge_doc.get("subject") + obj = edge_doc.get("object") + + if isinstance(subject, dict): + subject = subject.get("@id") + if isinstance(obj, dict): + obj = obj.get("@id") + + if not subject or not obj: + raise ValueError( + f"GraphEdge {edge_id} Lack of effective subject/object: " + f"subject={subject}, object={obj}" + ) + + WOQLQuery().delete_triple( + subject, + "outgoing_edge", + edge_id, + ).execute(self._client) + WOQLQuery().delete_triple( + edge_id, + "incoming_edge", + obj, + ).execute(self._client) + if self._has_doc(edge_id): + self._client.delete_document( + edge_id, + commit_msg=f"Delete GraphEdge {edge_id}", + ) + + def _delete_orphaned_nodes( + self, + candidate_node_ids: set[str], + *, + key: str = "profile", + value: str | None = None, + ) -> list[str]: + if not candidate_node_ids: + return [] + + connected_node_ids = set() + docs = self._client.get_all_documents(as_list=True) + for doc in docs: + if not isinstance(doc, dict): + continue + doc_id = doc.get("@id", "") + if not doc_id.startswith("GraphEdge/"): + continue + + subject = doc.get("subject") + obj = doc.get("object") + + if isinstance(subject, dict) and subject.get("@id"): + connected_node_ids.add(subject["@id"]) + if isinstance(obj, dict) and obj.get("@id"): + connected_node_ids.add(obj["@id"]) + + orphaned_node_ids = sorted( + node_id + for node_id in candidate_node_ids + if node_id not in connected_node_ids + ) + + deleted_node_ids = [] + for node_id in orphaned_node_ids: + if not self._has_doc(node_id): + continue + self._client.delete_document( + node_id, + commit_msg=f"Delete orphaned profile node {node_id}", + ) + deleted_node_ids.append(node_id) + return deleted_node_ids + + async def _delete_triples(self, triples, key, value=None): + candidate_nodes = set() + for t in triples: + candidate_nodes.add(t.subject) + candidate_nodes.add(t.object) + for t in triples: + edge_id, _ = self._edge_id(t) + edge_doc = self._client.get_document(edge_id) + if edge_doc: + self._delete_edge_document_and_links(edge_doc) + + if candidate_nodes: + self._delete_orphaned_nodes(candidate_nodes, key=key, value=value) + + def _docs_to_triples(self, edge_docs: list[dict]) -> list[GraphTriple]: + triples = [] + for doc in edge_docs: + subject = doc.get("subject") + obj = doc.get("object") + if isinstance(subject, dict): + subject = subject.get("@id") + if isinstance(obj, dict): + obj = obj.get("@id") + if not subject or not obj: + continue + triples.append( + GraphTriple( + subject=subject, + relation=doc["relation"], + object=obj, + subject_type="entity", + object_type="entity", + confidence=float(doc.get("weight", 1.0)), + source=doc.get("source", ""), + properties=doc.get("properties") or {}, + ) + ) + return triples + + def get_triples_info(self, added=[], updated=[], removed=[]): + def to_dict(triple: GraphTriple) -> dict: + props = triple.properties or {} + return { + "subject": triple.subject, + "relation": triple.relation, + "object": triple.object, + "source": props.get("_source_path") or triple.source or "", + } + + def serialize(triples) -> list[dict]: + return sorted( + (to_dict(triple) for triple in triples), + key=lambda t: (t["source"], t["subject"], t["relation"], t["object"]), + ) + + return { + "triples_added": len(added), + "triples_updated": len(updated), + "triples_deleted": len(removed), + "added_triples": serialize(added), + "updated_triples": serialize(updated), + "deleted_triples": serialize(removed), + } + + async def reconcile_profile_triples(self, triples): + previous_triples = self._get_triples(key="profile") + added, updated, removed = self._diff_triples( + previous_triples, + triples, + ) + changed = [*added, *updated] + if changed: + await self.upsert_triples(changed) + # Add replacement edges before orphan cleanup so nodes shared by an + # old and a new edge are never deleted between the two operations. + await self._delete_triples(removed, key="profile") + + return self.get_triples_info(added, updated, removed) + + async def reconcile_source_triples( + self, source: str, triples: Sequence[GraphTriple] + ) -> None: + """Replace edges produced from one source file while preserving others.""" + previous_triples = self._get_triples(key="source", value=source) + added, updated, removed = self._diff_triples( + previous_triples, + triples, + ) + changed = [*added, *updated] + if changed: + await self.upsert_triples(changed) + # A replacement edge may retain one endpoint from the removed edge. + # Insert it first so source-scoped orphan cleanup keeps that node. + await self._delete_triples(removed, key="source", value=source) + + return self.get_triples_info(added, updated, removed) + + def source_paths(self) -> set[str]: + """List document-managed origins, excluding lineage and explicit writes.""" + with self._state_lock: + docs = self._client.get_documents_by_type( + "GraphEdge", + as_list=True, + ) + + paths = set() + for doc in docs: + props = doc.get("properties") or {} + source_path = props.get("p__source_path") + + if not props.get("p__source_managed") or not source_path: + continue + + path = re.sub(r"#part=\d+$", "", str(source_path)) + paths.add(path) + + return paths + + async def reconcile_lineage_triples( + self, root: str, triples: Sequence[GraphTriple] + ) -> None: + """Replace all deterministic lineage edges for one workspace root.""" + previous_triples = self._get_triples(key="lineage", value=root) + added, updated, removed = self._diff_triples( + previous_triples, + triples, + ) + changed = [*added, *updated] + if changed: + await self.upsert_triples(changed) + # Lineage revisions commonly retain workspace/file endpoints. Insert + # replacements before orphan cleanup so those shared nodes survive. + await self._delete_triples(removed, key="lineage", value=root) + + return self.get_triples_info(added, updated, removed) + + async def delete_from_source( + self, + source: str, + key: str, + *, + value: str | None = None, + ) -> dict: + if not source: + raise ValueError("Source cannot be empty") + + with self._state_lock: + edge_docs = self._client.get_documents_by_type( + "GraphEdge", + as_list=True, + ) + def matches_source(doc): + if doc.get("source") == source: + return True + props = doc.get("properties") or {} + if props.get("p__source_path") == source: + return True + return False + + target_edges = [ + doc for doc in edge_docs + if matches_source(doc) + ] + target_triples = self._docs_to_triples(target_edges) + await self._delete_triples(target_triples, key, value) + + return { + "triples_deleted": len(target_triples) + } + + async def reset(self) -> None: + await asyncio.to_thread(self._reset_sync) + + def _reset_sync(self) -> None: + from terminusdb_client import Client + + admin_client = Client(self._dsn) + admin_client.connect( + team=self._team, + user=self._team, + key=self._password, + server_url=self._dsn, + ) + + if admin_client.has_database(self._db, team=self._team): + admin_client.delete_database(self._db, team=self._team) + + admin_client.create_database( + self._db, + team=self._team, + label=self._db, + description="Graph of DataMind stores in TerminusDB", + include_schema=False, + ) + + self._client = Client(self._dsn) + self._client.connect( + team=self._team, + db=self._db, + user=self._team, + key=self._password, + server_url=self._dsn, + ) + + self._schema() + with sqlite3.connect(self._history_path) as conn: + conn.execute("DELETE FROM history WHERE database = ?", (self._db,)) + + # ------------------------------------------------------------- lookup + + def _documents_at_version(self, version: str | None) -> list[dict]: + """Read instance documents from HEAD or an immutable commit.""" + with self._state_lock: + if version is not None: + # get_all_documents silently ignores commit=...; select the + # immutable commit through the client's ref URL instead. + client = self._client + # Workspace-pinned stores are already connected to their + # durable dm_build_ branch. In that path the + # version is only metadata; reading branch HEAD avoids + # resolving a commit against the wrong ref. + if client.branch and client.branch.startswith("dm_build_"): + return client.get_all_documents(as_list=True) + previous_ref = client.ref + try: + client.ref = str(version) + return client.get_all_documents(as_list=True) + finally: + client.ref = previous_ref + return self._client.get_all_documents(as_list=True) + + + @staticmethod + def _entity_id(document_id: str) -> str: + return document_id.removeprefix("entity/") + + async def search_entities(self, query: str, *, top_k: int = 5, version: str | None = None,) -> list[Entity]: + q = query.lower().strip() + if not q: + return [] + scored: list[tuple[float, str, dict]] = [] + docs = self._documents_at_version(version) + for data in docs: + if data.get("@type") == "GraphEdge": + continue + nid = data["@id"] + public_id = self._entity_id(nid) + label = str(data.get("label", public_id)) + # Exact and substring hits beat fuzzy. + if public_id.lower() == q or label.lower() == q: + score = 1.0 + elif q in public_id.lower() or q in label.lower(): + score = 0.7 + else: + ratio = difflib.SequenceMatcher(None, q, label.lower()).ratio() + if ratio < 0.4: + continue + score = 0.4 + 0.4 * ratio # cap below substring score + scored.append((score, public_id, data)) + scored.sort(key=lambda x: -x[0]) + return [ + Entity( + id=nid, + label=str(data.get("label", nid)), + type=str(data.get("@type", "entity")), + score=score, + properties={ + k: v for k, v in data.items() if k not in {"label", "@type"} + }, + ) + for score, nid, data in scored[:top_k] + ] + + async def neighbors( + self, + entity: str, + *, + direction: str = "both", + relation_filter: list[str] | None = None, + limit: int = 100, + version: str | None = None, + ) -> list[Edge]: + if direction not in {"in", "out", "both"}: + raise ValueError(f"Invalid direction: {direction}") + if limit <= 0: + return [] + + docs = self._documents_at_version(version) + + allowed = set(relation_filter) if relation_filter else None + edges: list[Edge] = [] + + for doc in docs: + if doc.get("@type") != "GraphEdge": + continue + subject = doc.get("subject") + obj = doc.get("object") + if isinstance(subject, dict): + subject = subject.get("@id") + if isinstance(obj, dict): + obj = obj.get("@id") + if not isinstance(subject, str) or not isinstance(obj, str): + continue + subject_id = self._entity_id(subject) + object_id = self._entity_id(obj) + relation = doc.get("relation", "related") + + if allowed is not None and relation not in allowed: + continue + matches_out = ( + direction in {"out", "both"} and subject_id == entity + ) + matches_in = ( + direction in {"in", "both"} and object_id == entity + ) + + if matches_out or matches_in: + edges.append(self._edge(subject_id, object_id, doc)) + if len(edges) >= limit: + break + return edges + + async def traverse( + self, + start: str, + *, + max_hops: int = 3, + relation_filter: list[str] | None = None, + max_results: int = 100, + version: str | None = None + ) -> list[GraphPath]: + from terminusdb_client import WOQLQuery + + if max_hops < 1 or max_results <= 0: + return [] + + internal_start = ( + start if start.startswith("entity/") else f"entity/{start}" + ) + allowed = set(relation_filter) if relation_filter else None + # Each business edge of the jump operation = Entity -> GraphEdge -> Entity + pattern = ( + f"(outgoing_edge,incoming_edge){{1,{max_hops}}}" + ) + if version is None: + query = WOQLQuery().path( + internal_start, + pattern, + "v:End", + "v:Path", + ) + else: + query = WOQLQuery().using( + f"{self._team}/{self._db}/local/commit/{version}" + ).path( + internal_start, + pattern, + "v:End", + "v:Path", + ) + + docs = self._documents_at_version(version) + + result = self._client.query(query) + bindings = result.get("bindings", []) + + edge_doc_cache: dict[str, dict] = {} + def get_edge_doc(edge_id: str) -> dict: + if edge_id not in edge_doc_cache: + doc = next( + (d for d in docs if d.get("@id") == edge_id), + None, + ) + if not doc: + raise ValueError(f"GraphEdge document not found: {edge_id}") + edge_doc_cache[edge_id] = doc + return edge_doc_cache[edge_id] + + candidates = [] + for binding in bindings: + raw_path = binding.get("Path", []) + if len(raw_path) % 2 != 0: + continue + internal_nodes = [internal_start] + nodes = [self._entity_id(internal_start)] + edges = [] + seen = {internal_start} + bfs_keys = [] + valid = True + for i in range(0, len(raw_path), 2): + out_edge = raw_path[i] + in_edge = raw_path[i + 1] + + if not ( + out_edge.get("woql:predicate", "").endswith( + "outgoing_edge" + ) + and in_edge.get("woql:predicate", "").endswith( + "incoming_edge" + ) + ): + valid = False + break + + source_id = out_edge.get("woql:subject") + edge_id = out_edge.get("woql:object") + intermediate_id = in_edge.get("woql:subject") + target_id = in_edge.get("woql:object") + + if not edge_id or intermediate_id != edge_id: + valid = False + break + if source_id != internal_nodes[-1] or not target_id: + valid = False + break + if target_id in seen: + valid = False + break + + edge_doc = get_edge_doc(edge_id) + relation = edge_doc.get("relation", "related") + if allowed is not None and relation not in allowed: + valid = False + break + edge_obj = self._edge( + self._entity_id(source_id), + self._entity_id(target_id), + edge_doc, + ) + edges.append(edge_obj) + internal_nodes.append(target_id) + nodes.append(self._entity_id(target_id)) + seen.add(target_id) + + edge_key = str( + edge_doc.get("key") + or edge_doc.get("@id") + or edge_id + ) + bfs_keys.append( + (target_id, str(relation), edge_key) + ) + + if not valid or not edges: + continue + if len(edges) > max_hops: + continue + score = sum(edge.weight for edge in edges) / len(edges) + path_obj = GraphPath( + nodes=nodes, + edges=edges, + score=score, + ) + candidates.append( + (len(edges), tuple(bfs_keys), path_obj) + ) + + candidates.sort(key=lambda item: (item[0], item[1])) + selected = candidates[:max_results] + selected.sort(key=lambda item: -item[2].score) + return [item[2] for item in selected] + + async def rollback(self, version: str): + before_version = self._get_latest_version() + diff = self._client.diff_version(before_version, version) + if not diff: + return { + "success": True, + "before_version": before_version, + "target_version": version, + "message": "Current status matches the target version; no rollback is required.", + "info": { + "triples_added": 0, + "triples_updated": 0, + "triples_deleted": 0, + }, + "diff": [], + } + + message = f"Rollback to version {version}" + self._client.apply( + before_version=before_version, + after_version=version, + message=message, + ) + current = self._get_latest_version() + info = { + "success": True, + "message": message, + "version": current, + "diff": diff, + } + await self.add_version(info) + return info + + async def add_version(self, info: dict): + version = self._get_latest_version() + with sqlite3.connect(self._history_path) as conn: + conn.execute("INSERT OR REPLACE INTO history VALUES (?, ?, ?)", + (self._db, version, json.dumps(info, default=str))) + + async def list_versions(self): + with sqlite3.connect(self._history_path) as conn: + return [ + {"version": row[0], "payload": json.loads(row[1])} + for row in conn.execute( + "SELECT version, payload FROM history " + "WHERE database = ? ORDER BY rowid", + (self._db,), + ) + ] + + async def schema(self, version: str | None = None) -> dict[str, Any]: + """Return observed graph vocabulary from TerminusDB instance documents.""" + from collections import Counter + + def _schema_sync(): + with self._state_lock: + documents = self._documents_at_version(version) + + types = Counter() + relations = Counter() + + for doc in documents: + if not isinstance(doc, dict): + continue + + doc_type = doc.get("@type") + + if doc_type != "GraphEdge": + types[doc_type or "entity"] += 1 + else: + relations[doc.get("relation", "related")] += 1 + + return { + "node_types": dict(sorted(types.items())), + "relations": dict(sorted(relations.items())), + "extensible": True, + "nodes": sum(types.values()), + "edges": sum(relations.values()), + } + + return await asyncio.to_thread(_schema_sync) + + # -------------------------------------------------------------- helpers + + @staticmethod + def _edge(u: str, v: str, data: dict) -> Edge: + return Edge( + source=u, + target=v, + relation=str(data.get("relation", "related")), + weight=float(data.get("weight", 1.0)), + properties={ + k[2:] if k.startswith("p_") else k: val + for k, val in data.items() + if k not in {"relation", "weight"} + }, + ) + + def stats(self, version: str | None = None) -> dict[str, int]: + """Return the number of graph nodes and edges at a specific version.""" + with self._state_lock: + documents = self._documents_at_version(version) + + node_ids: set[str] = set() + edge_count = 0 + + for doc in documents: + if not isinstance(doc, dict): + continue + + if doc.get("@type") != "GraphEdge": + continue + + subject = doc.get("subject") + obj = doc.get("object") + + if isinstance(subject, dict): + subject = subject.get("@id") + if isinstance(obj, dict): + obj = obj.get("@id") + + if not isinstance(subject, str) or not subject: + continue + if not isinstance(obj, str) or not obj: + continue + + node_ids.add(subject) + node_ids.add(obj) + edge_count += 1 + + return { + "nodes": len(node_ids), + "edges": edge_count, + } + + def _get_latest_version(self): + history = self._client.get_commit_history(max_history=1) + + if not history: + raise RuntimeError("No commit history found") + + if isinstance(history, dict): + first_key = next(iter(history)) + first_value = history[first_key] + + if isinstance(first_value, dict): + return ( + first_value.get("identifier") + or first_value.get("commit_id") + or first_key + ) + + return first_key + + if isinstance(history, list): + item = history[0] + return ( + item.get("identifier") + or item.get("commit_id") + or item.get("commit") + ) + + raise TypeError(f"Unexpected commit history: {type(history)}") + + + diff --git a/datamind/capabilities/graph/service.py b/datamind/capabilities/graph/service.py index b7d6ab2..8d728d8 100644 --- a/datamind/capabilities/graph/service.py +++ b/datamind/capabilities/graph/service.py @@ -11,6 +11,7 @@ from datamind.core.protocols import GraphEvidence, GraphStore, GraphTriple from .evidence import persist_input from datamind.core.registry import graph_registry +from datamind.core.errors import CapabilityError # Importing providers populates graph_registry. from . import providers # noqa: F401 @@ -82,6 +83,13 @@ async def load_from_profile(self) -> dict[str, int]: async def search_entities(self, query: str, *, top_k: int = 5) -> list[dict[str, Any]]: return [e.model_dump() for e in await self.store.search_entities(query, top_k=top_k)] + + async def search_entities_at_version(self, query: str, *, top_k: int = 5, version: str = None) -> list[dict[str, Any]]: + if self.store.name != "terminusdb": + raise CapabilityError("kb", "rollback is only supported by the LanceDB vector store") + if not isinstance(version, str): + raise CapabilityError("kb", f"version must be an string, got {type(version).__name__}") + return [e.model_dump() for e in await self.store.search_entities(query, top_k=top_k, version=version)] async def traverse( self, @@ -96,6 +104,25 @@ async def traverse( max_results=max_results, ) return [p.model_dump() for p in paths] + + async def traverse_at_version( + self, + start: str, + *, + max_hops: int = 2, + relation_filter: list[str] | None = None, + max_results: int = 100, + version: str = None, + ) -> list[dict[str, Any]]: + if self.store.name != "terminusdb": + raise CapabilityError("kb", "rollback is only supported by the LanceDB vector store") + if not isinstance(version, str): + raise CapabilityError("kb", f"version must be an string, got {type(version).__name__}") + paths = await self.store.traverse( + start, max_hops=max_hops, relation_filter=relation_filter, + max_results=max_results, version=version, + ) + return [p.model_dump() for p in paths] async def neighbors( self, @@ -109,6 +136,20 @@ async def neighbors( entity, direction=direction, relation_filter=relation_filter, limit=limit, ) return [e.model_dump() for e in edges] + + async def neighbors_at_version( + self, + entity: str, + *, + direction: str = "both", + relation_filter: list[str] | None = None, + limit: int = 100, + version: str = None, + ) -> list[dict[str, Any]]: + edges = await self.store.neighbors( + entity, direction=direction, relation_filter=relation_filter, limit=limit, version=version + ) + return [e.model_dump() for e in edges] async def upsert(self, triples: list[dict[str, Any]]) -> dict[str, int]: parsed = [GraphTriple(**t) for t in triples] @@ -127,8 +168,38 @@ async def upsert(self, triples: list[dict[str, Any]]) -> dict[str, int]: persist = getattr(self.store, "persist", None) if callable(persist): await persist() + + add_version = getattr(self.store, "add_version", None) + if add_version and callable(add_version): + info = self.store.get_triples_info(added=parsed) + await add_version(info) + return {"upserted": len(parsed)} + async def rollback(self, version: str = None) -> dict[str, Any]: + if self.store.name != "terminusdb": + raise CapabilityError("kb", "rollback is only supported by the LanceDB vector store") + if not isinstance(version, str): + raise CapabilityError("kb", f"version must be an string, got {type(version).__name__}") + + versions = await self.list_versions() + if not any(v["version"] == version for v in versions): + raise CapabilityError("kb", f"version {version} does not exist in this LanceDB table") + + result = await self.store.rollback(version=version) + return { + "status": "ok", + "message": f"Successfully rolled back to version {version}", + "rollback_version": version, # Target + "new_version": result["version"], # New Commit + } + + async def list_versions(self) -> dict[str, Any]: + if self.store.name != "terminusdb": + raise CapabilityError("kb", "list_versions is only supported by the LanceDB vector store") + + return await self.store.list_versions() + def stats(self) -> dict[str, int]: stats = getattr(self.store, "stats", None) return stats() if callable(stats) else {} @@ -148,6 +219,13 @@ def build_graph_service(settings: Settings) -> GraphService: "networkx", persist_path=str(storage_dir / "graph.json"), ) + elif backend == "terminusdb": + store = graph_registry.create( + "terminusdb", + dsn=settings.graph.dsn, + password=settings.graph.password, + persist_path=str(storage_dir / "graph"), + ) else: # DSN-based remote backends (neo4j, ...) — future work. store = graph_registry.create(backend, dsn=settings.graph.dsn) diff --git a/datamind/capabilities/graph/tools.py b/datamind/capabilities/graph/tools.py index ddaf41e..dc43a23 100644 --- a/datamind/capabilities/graph/tools.py +++ b/datamind/capabilities/graph/tools.py @@ -12,6 +12,10 @@ def build_graph_tools(graph: GraphService) -> list[ToolSpec]: async def _search(query: str, top_k: int = 5) -> dict: entities = await graph.search_entities(query, top_k=top_k) return {"query": query, "count": len(entities), "entities": entities} + + async def _search_at_version(query: str, top_k: int = 5, version: str = None) -> dict: + entities = await graph.search_entities(query, top_k=top_k, version=version) + return {"query": query, "count": len(entities), "entities": entities} async def _traverse( start: str, @@ -30,6 +34,25 @@ async def _traverse( "total_count": len(paths), "truncated": truncated, "next_cursor": max_results if truncated else None, "paths": visible, } + + async def _traverse_at_version( + start: str, + max_hops: int = 2, + relation_filter: list[str] | None = None, + max_results: int = 100, + version: str = None, + ) -> dict: + paths = await graph.traverse( + start, max_hops=max_hops, relation_filter=relation_filter, + max_results=max_results + 1, version=version, + ) + truncated = len(paths) > max_results + visible = paths[:max_results] + return { + "start": start, "max_hops": max_hops, "count": len(visible), + "total_count": len(paths), "truncated": truncated, + "next_cursor": max_results if truncated else None, "paths": visible, + } async def _neighbors( entity: str, @@ -47,14 +70,38 @@ async def _neighbors( "total_count": len(edges), "truncated": truncated, "next_cursor": limit if truncated else None, "edges": visible, } + + async def _neighbors_at_version( + entity: str, + direction: str = "both", + relation_filter: list[str] | None = None, + limit: int = 100, + version: str = None + ) -> dict: + edges = await graph.neighbors( + entity, direction=direction, relation_filter=relation_filter, limit=limit + 1, version=version, + ) + truncated = len(edges) > limit + visible = edges[:limit] + return { + "entity": entity, "direction": direction, "count": len(visible), + "total_count": len(edges), "truncated": truncated, + "next_cursor": limit if truncated else None, "edges": visible, + } async def _upsert(triples: list[dict]) -> dict: return await graph.upsert(triples) async def _schema(): - return graph.store.schema() + return await graph.store.schema() - return [ + async def _list_versions(): + return await graph.list_versions() + + async def _rollback(version: str = None): + return await graph.rollback(version=version) + + tools = [ ToolSpec(name='graph_schema', description='Discover observed node types and relation names before graph search.', input_schema={'type': 'object', 'properties': {}}, handler=_schema, metadata={'surface': 'graph', 'access': 'read'}), @@ -178,7 +225,148 @@ async def _schema(): }, ), ] - + + if ( + graph is not None + and getattr(getattr(graph, "store", None), "name", None) == "terminusdb" + ): + tools.extend([ + ToolSpec( + name="graph_search_entities_at_version", + description=( + "Search for knowledge graph entities by name or alias in a specific " + "graph version. Use this tool first when the query mentions an entity " + "by name and its entity ID is unknown. Returns matching entities from " + "the specified historical state without changing the current graph state." + ), + input_schema={ + "type": "object", + "properties": { + "query": {"type": "string", "description": "Entity name or alias."}, + "top_k": {"type": "integer", "minimum": 1, "maximum": 20, "default": 5}, + "version": { + "type": "string", + "description": ( + "The exact graph version or commit identifier to query. " + "Use an identifier returned by list_versions. " + ), + }, + }, + "required": ["query", "version"], + }, + handler=_search_at_version, + metadata={"group": "graph", "surface": "graph", "access": "read"}, + ), + ToolSpec( + name="graph_traverse_at_version", + description=( + "Traverse the knowledge graph from a starting entity in a specific " + "historical version. Explores paths up to max_hops and returns a bounded " + "set of paths ranked by average edge weight. Use relation_filter to " + "restrict traversal to exact relation types. The query does not change " + "the current graph state." + ), + input_schema={ + "type": "object", + "properties": { + "start": {"type": "string", "description": "Entity id to start from."}, + "max_hops": {"type": "integer", "minimum": 1, "maximum": 5, "default": 2}, + "relation_filter": { + "type": "array", + "items": {"type": "string"}, + "description": "Optional allow-list of relation names.", + }, + "max_results": { + "type": "integer", "minimum": 1, "maximum": 500, "default": 100, + "description": "Hard cap on returned paths.", + }, + "version": { + "type": "string", + "description": ( + "The exact graph version or commit identifier to query. " + "Use an identifier returned by list_versions. " + ), + }, + }, + "required": ["start", "version"], + }, + handler=_traverse_at_version, + metadata={"group": "graph", "surface": "graph", "access": "read"}, + ), + ToolSpec( + name="graph_neighbors_at_version", + description=( + "Retrieve edges connected to an entity in a specific historical graph " + "version. Use direction='out' for outgoing edges, 'in' for incoming " + "edges, or 'both' for both directions. Optionally restrict results using " + "relation_filter and cap the number of returned edges with limit. " + "Use this tool to inspect an entity's historical relationships without " + "changing the current graph state." + ), + input_schema={ + "type": "object", + "properties": { + "entity": {"type": "string"}, + "direction": { + "type": "string", + "enum": ["out", "in", "both"], + "default": "both", + }, + "relation_filter": { + "type": "array", "items": {"type": "string"}, + "description": "Only return these exact relation types.", + }, + "limit": { + "type": "integer", "minimum": 1, "maximum": 500, "default": 100, + }, + "version": { + "type": "string", + "description": ( + "The exact graph version or commit identifier to query. " + "Use an identifier returned by list_versions. " + ), + }, + }, + "required": ["entity", "version"], + }, + handler=_neighbors_at_version, + metadata={"group": "graph", "surface": "graph", "access": "read"}, + ), + ToolSpec( + name="list_versions", + description=( + "List available knowledge graph versions or commits. Use this tool to " + "discover valid version identifiers before calling historical entity " + "search, traversal, or neighbor tools. Use the returned identifiers " + "as the version argument for historical queries." + ), + input_schema={}, + handler=_list_versions, + metadata={"group": "graph", "surface": "graph", "access": "read"}, + ), + ToolSpec( + name="rollback", + description=( + "Restore the knowledge graph to a specified historical version. " + "This is a state-changing operation, unlike historical read tools. " + "Only use it when explicitly requested to roll back the graph; do not " + "use it merely to query historical data." + ), + input_schema={ + "version": { + "type": "string", + "description": ( + "The exact graph version or commit identifier to query. " + "Use an identifier returned by list_versions. " + ), + }, + }, + handler=_rollback, + metadata={"group": "graph", "surface": "graph", "access": "write", "destructive": True,}, + ), + ]) + + return tools @tool_provider_registry.register("graph") class _GraphToolProvider: diff --git a/datamind/capabilities/ingest/lineage.py b/datamind/capabilities/ingest/lineage.py new file mode 100644 index 0000000..df2edaf --- /dev/null +++ b/datamind/capabilities/ingest/lineage.py @@ -0,0 +1,116 @@ +"""Pure deterministic file-lineage construction shared by all graph backends.""" +from __future__ import annotations + +import csv +import hashlib +import re +from dataclasses import dataclass +from pathlib import Path + +from datamind.core.errors import CapabilityError +from datamind.core.protocols import GraphEvidence, GraphTriple + +TEXT_EXTS = {".txt", ".md", ".markdown", ".html", ".json", ".xml", ".py", ".java"} +TABLE_EXTS = {".csv", ".tsv"} +VERSION_MARKER = re.compile( + r"[ _\-]*(v\d+|final|draft|copy|revised|updated|old|new|\(\d+\)|\d{4}[-_]\d{2}[-_]\d{2})$", + re.IGNORECASE, +) + + +@dataclass(frozen=True) +class LineageFile: + uri: str + relative_path: str + version: str + content_path: Path + + +def build_lineage_triples(*, root: str | Path, files: list[LineageFile], dependencies=None): + root = str(Path(root).resolve()) + root_id = hashlib.sha256((root + "lineage").encode()).hexdigest()[:12] + workspace = f"workspace::{root_id}" + nodes = {item.uri: f"file::{root_id}::{item.relative_path}" for item in files} + props = {"_lineage_root": root, "_source_managed": True} + by_uri = {item.uri: item for item in files} + + def evidence(uris, *, method="inferred", pipeline_version="lineage-v1"): + return [ + GraphEvidence( + source_id="file:" + hashlib.sha256(uri.encode()).hexdigest(), + source_version=by_uri[uri].version, + locator={"uri": uri, "relative_path": by_uri[uri].relative_path}, + method=method, + pipeline_version=pipeline_version, + ) + for uri in dict.fromkeys(uris) + if uri in by_uri + ] + + triples = [GraphTriple(subject=workspace, relation="contains", object=nodes[item.uri], + source=root, evidence=evidence([item.uri]), + properties={**props, "source_file": item.uri, "sha256": item.version}) + for item in files] + + texts, columns = {}, {} + for item in files: + suffix = Path(item.uri).suffix.lower() + try: + if suffix in TEXT_EXTS: + texts[item.uri] = item.content_path.read_text(encoding="utf-8", errors="replace")[:200_000] + elif suffix in TABLE_EXTS: + delimiter = "\t" if suffix == ".tsv" else "," + with item.content_path.open("r", encoding="utf-8", errors="replace", newline="") as handle: + header = next(csv.reader(handle, delimiter=delimiter), []) + columns[item.uri] = {re.sub(r"[^a-z0-9]+", "_", c.strip().lower()).strip("_") + for c in header if c.strip()} + except OSError: + continue + + for left, content in texts.items(): + for right in nodes: + if left != right and Path(right).name in content: + triples.append(GraphTriple(subject=nodes[left], relation="mentions", object=nodes[right], + source=root, evidence=evidence([left, right]), + properties={**props, "source_file": left})) + table_items = list(columns.items()) + for index, (left, left_cols) in enumerate(table_items): + for right, right_cols in table_items[index + 1:]: + union = left_cols | right_cols + score = len(left_cols & right_cols) / len(union) if union else 0 + if score >= .5: + triples.append(GraphTriple(subject=nodes[left], relation="schema_overlap", object=nodes[right], + source=root, evidence=evidence([left, right]), + properties={**props, "source_file": left, "jaccard": round(score, 3)})) + + for index, left in enumerate(files): + left_stem = VERSION_MARKER.sub("", Path(left.uri).stem).strip().lower() + for right in files[index + 1:]: + right_stem = VERSION_MARKER.sub("", Path(right.uri).stem).strip().lower() + if left_stem and left_stem == right_stem and Path(left.uri).name != Path(right.uri).name: + triples.append(GraphTriple(subject=nodes[left.uri], relation="version_of", object=nodes[right.uri], + source=root, evidence=evidence([left.uri, right.uri]), + properties={**props, "source_file": left.uri})) + + by_hash = {} + for item in files: + by_hash.setdefault(item.version, []).append(item.uri) + for version, members in by_hash.items(): + for index, left in enumerate(members): + for right in members[index + 1:]: + triples.append(GraphTriple(subject=nodes[left], relation="shared_artifact", object=nodes[right], + source=root, evidence=evidence([left, right]), + properties={**props, "source_file": left, "sha256": version})) + + by_relative = {item.relative_path: item.uri for item in files} + for edge in dependencies or []: + if edge.get("relation", "depends_on") != "depends_on": + raise CapabilityError("ingest", "explicit file dependencies only support depends_on") + try: + left, right = (by_relative[edge[field]] for field in ("source", "target")) + except (KeyError, TypeError): + raise CapabilityError("ingest", "dependency endpoints must be workspace files") from None + triples.append(GraphTriple(subject=nodes[left], relation="depends_on", object=nodes[right], source=root, + evidence=evidence([left, right], method="explicit", pipeline_version="lineage-explicit-v1"), + properties={**props, "source_file": left})) + return triples diff --git a/datamind/capabilities/ingest/service.py b/datamind/capabilities/ingest/service.py index 4f9d709..1bd0648 100644 --- a/datamind/capabilities/ingest/service.py +++ b/datamind/capabilities/ingest/service.py @@ -52,6 +52,7 @@ from datamind.core.protocols import GraphEvidence, GraphTriple, TextModelClient from .formats import DOCUMENT_EXTS, TABLE_EXTS, extract_document, extract_tabular +from .lineage import LineageFile, build_lineage_triples _log = get_logger("ingest") @@ -75,13 +76,7 @@ ".pptx": ("kb", "graph_text"), } -_LINEAGE_TEXT_EXTS = {".txt", ".md", ".markdown", ".html", ".json", ".xml", ".py", ".java"} -_LINEAGE_TABLE_EXTS = {".csv", ".tsv"} _GRAPH_DOCUMENT_EXTS = DOCUMENT_EXTS - TABLE_EXTS -_LINEAGE_VERSION_MARKER_RE = re.compile( - r"[ _\-]*(v\d+|final|draft|copy|revised|updated|old|new|\(\d+\)|\d{4}[-_]\d{2}[-_]\d{2})$", - re.IGNORECASE, -) def _infer_table_name(stem: str) -> str: @@ -310,126 +305,16 @@ async def graph_build_lineage( } root = str(resolved if resolved.is_dir() else resolved.parent) - root_id = _hash(root, "lineage")[:12] - workspace_node = f"workspace::{root_id}" - file_nodes: dict[str, str] = {} - file_text: dict[str, str] = {} - file_columns: dict[str, set[str]] = {} - for item in files: - candidate = Path(item["path"]) - rel = item.get("relative_path") or candidate.name - node = f"file::{root_id}::{rel}" - file_nodes[str(candidate)] = node - suffix = item.get("extension", "") - if suffix in _LINEAGE_TEXT_EXTS: - try: - file_text[str(candidate)] = candidate.read_text( - encoding="utf-8", errors="replace" - )[:200_000] - except OSError: - pass - if suffix in _LINEAGE_TABLE_EXTS: - try: - delimiter = "\t" if suffix == ".tsv" else "," - with candidate.open("r", encoding="utf-8", errors="replace", newline="") as handle: - header = next(csv.reader(handle, delimiter=delimiter), []) - file_columns[str(candidate)] = { - re.sub(r"[^a-z0-9]+", "_", col.strip().lower()).strip("_") - for col in header if col.strip() - } - except (OSError, StopIteration): - pass - - triples: list[GraphTriple] = [] - properties_base = {"_lineage_root": root, "_source_managed": True} - for item in files: - candidate = Path(item["path"]) - node = file_nodes[str(candidate)] - triples.append(GraphTriple( - subject=workspace_node, - relation="contains", - object=node, - source=root, - properties={**properties_base, "source_file": str(candidate), "sha256": item.get("sha256")}, - )) - - # Mentions: a text file explicitly names another file in the same workspace. - for source_path, text_content in file_text.items(): - source_node = file_nodes[source_path] - for target_path, target_node in file_nodes.items(): - if source_path == target_path: - continue - target_name = Path(target_path).name - if target_name and target_name in text_content: - triples.append(GraphTriple( - subject=source_node, relation="mentions", object=target_node, - source=root, - properties={**properties_base, "source_file": source_path}, - )) - - table_items = list(file_columns.items()) - for index, (left_path, left_cols) in enumerate(table_items): - for right_path, right_cols in table_items[index + 1:]: - union = left_cols | right_cols - if not union: - continue - jaccard = len(left_cols & right_cols) / len(union) - if jaccard >= 0.5: - triples.append(GraphTriple( - subject=file_nodes[left_path], relation="schema_overlap", - object=file_nodes[right_path], source=root, - properties={**properties_base, "source_file": left_path, "jaccard": round(jaccard, 3)}, - )) - - paths = list(file_nodes.items()) - for index, (left_path, left_node) in enumerate(paths): - left_stem = _LINEAGE_VERSION_MARKER_RE.sub( - "", Path(left_path).stem - ).strip().lower() - for right_path, right_node in paths[index + 1:]: - if left_stem and left_stem == _LINEAGE_VERSION_MARKER_RE.sub( - "", Path(right_path).stem - ).strip().lower() and Path(left_path).name != Path(right_path).name: - triples.append(GraphTriple( - subject=left_node, relation="version_of", object=right_node, - source=root, - properties={**properties_base, "source_file": left_path}, - )) - - by_hash: dict[str, list[str]] = {} - for item in files: - digest = item.get("sha256") - if digest: - by_hash.setdefault(str(digest), []).append(str(item["path"])) - for digest, members in by_hash.items(): - if len(members) < 2: - continue - for index, left_path in enumerate(members): - for right_path in members[index + 1:]: - triples.append(GraphTriple( - subject=file_nodes[left_path], relation="shared_artifact", - object=file_nodes[right_path], source=root, - properties={**properties_base, "source_file": left_path, "sha256": digest}, - )) - - # Explicit file dependencies use workspace-relative paths and are never inferred. - for edge in dependencies or []: - if edge.get("relation", "depends_on") != "depends_on": - raise CapabilityError("ingest", "explicit file dependencies only support depends_on") - endpoints = [] - for field in ("source", "target"): - value = edge.get(field) - if not value: - raise CapabilityError("ingest", f"dependency {field} is required") - endpoint = str((Path(root) / value).resolve()) - if endpoint not in file_nodes: - raise CapabilityError("ingest", f"dependency file not in inventory: {value}") - endpoints.append(endpoint) - triples.append(GraphTriple( - subject=file_nodes[endpoints[0]], relation="depends_on", - object=file_nodes[endpoints[1]], source=root, - properties={**properties_base, "source_file": endpoints[0], "evidence": "explicit"}, - )) + lineage_files = [LineageFile( + uri=str(Path(item["path"])), + relative_path=item.get("relative_path") or Path(item["path"]).name, + version=str(item.get("sha256") or ""), + content_path=Path(item["path"]), + ) for item in files] + triples = build_lineage_triples( + root=root, files=lineage_files, dependencies=dependencies, + ) + workspace_node = next((t.subject for t in triples if t.relation == "contains"), None) reconcile = getattr(self._graph.store, "reconcile_lineage_triples", None) if callable(reconcile): @@ -441,6 +326,21 @@ async def graph_build_lineage( result = persist() if hasattr(result, "__await__"): await result + + add_version = getattr(self._graph.store, "add_version", None) + if callable(add_version): + lineage_version = hashlib.sha256( + json.dumps( + [(t.subject, t.relation, t.object) for t in triples], + ensure_ascii=False, sort_keys=True, separators=(",", ":"), + ).encode("utf-8") + ).hexdigest() + await add_version({ + "kind": "lineage", "root": root, + "version": f"sha256:{lineage_version}", + "files": len(files), "edges": len(triples), + }) + return { "path": str(resolved), "workspace_node": workspace_node, @@ -1023,7 +923,7 @@ async def kb_delete_file(self, *, path: str): "chunks_deleted": len(delete_ids), "version": version_info.get("version") if version_info else None, "msg": version_info.get("msg") if version_info else None, - "delete_parsed_chunks": parsed_chunks + "delete_parsed_chunks": parsed_chunks, } def _sync_prechunked_files( @@ -1856,17 +1756,29 @@ async def _extract_graph_triples( triples.append(triple) return triples - async def _commit_graph_source(self, source: str | None, triples: list[GraphTriple]) -> None: + async def _commit_graph_source(self, source: str | None, triples: list[GraphTriple], version_info: bool = False) -> None: + """ + Args: + version_info: Whether to record this update as a new graph version + when the store supports versioning. + """ + reconcile = getattr(self._graph.store, "reconcile_source_triples", None) if source and callable(reconcile): - await reconcile(source, triples) + info = await reconcile(source, triples) else: - await self._graph.store.upsert_triples(triples) + info = await self._graph.store.upsert_triples(triples) persist = getattr(self._graph.store, "persist", None) if callable(persist): result = persist() if hasattr(result, "__await__"): await result + + if version_info: + add_version = getattr(self._graph.store, "add_version", None) + if add_version and callable(add_version): + await add_version(info) + return info or {} async def graph_add_triples_from_text( self, *, text: str, max_triples: int = 30, source: str | None = None, @@ -1884,13 +1796,14 @@ async def graph_add_triples_from_text( source_version=source_version, locator={"uri": str(evidence_path), "char_start": 0, "char_end": len(text)}, ) - await self._commit_graph_source(source, triples) + await self._commit_graph_source(source, triples, version_info=True) + return {"triples_added": len(triples), "source": source, "samples": [t.model_dump(mode="json") for t in triples[:8]]} async def graph_add_path( self, *, path: str, recursive: bool = True, max_triples_per_file: int = 30, - relation_types: list[str] | None = None, + relation_types: list[str] | None = None, commit: bool = True, ) -> dict[str, Any]: """Replace each complete document revision, preserving other source supports. @@ -1916,6 +1829,7 @@ async def graph_add_path( raise CapabilityError("ingest", f"path does not exist: {resolved}") processed, skipped, removed = [], [], [] + infos = {} for candidate in candidates: try: candidate = _resolve_safe_path(str(candidate), self._allowed_roots) @@ -1945,25 +1859,129 @@ async def graph_add_path( results.append({"source": part_source, "triples_added": len(extracted)}) if self._file_hash(candidate) != raw_version: raise CapabilityError("ingest", "source changed during extraction; retry") - await self._commit_graph_source(str(candidate), triples) + info = await self._commit_graph_source(str(candidate), triples) if commit else {} + processed.append({"path": str(candidate), "source": str(candidate), "source_version": raw_version, "triples_added": len(triples), "parts_processed": len(results), "parts": results, - "truncated": len(results) < len(parts) or len(triples) >= max_triples_per_file}) + "truncated": len(results) < len(parts) or len(triples) >= max_triples_per_file, + "triples": [triple.model_dump(mode="json") for triple in triples], **info}) + + for key, value in info.items(): + if isinstance(value, (int, float)): + infos[key] = infos.get(key, 0) + value + elif isinstance(value, list): + infos.setdefault(key, []).extend(value) + except Exception as exc: skipped.append(f"{candidate}: {type(exc).__name__}: {exc}") sources = getattr(self._graph.store, "source_paths", None) - if directory and callable(sources): + if commit and directory and callable(sources): for old_source in sorted(sources()): old = Path(old_source) if old.is_relative_to(resolved) and (recursive or old.parent == resolved) and not old.exists(): - await self._commit_graph_source(old_source, []) + info = await self._commit_graph_source(old_source, []) + for key, value in info.items(): + if isinstance(value, (int, float)): + infos[key] = infos.get(key, 0) + value + elif isinstance(value, list): + infos.setdefault(key, []).extend(value) removed.append(old_source) + + add_version = getattr(self._graph.store, "add_version", None) + if commit and add_version and callable(add_version): + await add_version(infos) return {"path": str(resolved), "files_processed": len(processed), - "triples_added": sum(p["triples_added"] for p in processed), "files": processed, - "removed_sources": removed, "skipped": skipped[:20], "skipped_count": len(skipped)} + "triples_added": sum(p["triples_added"] for p in processed), + "triples_deleted": sum(p.get("triples_deleted") or 0 for p in processed), + "files": processed, "removed_sources": removed, + "skipped": skipped[:20], "skipped_count": len(skipped)} + + async def graph_delete_path( + self, + *, + path: str, + recursive: bool = True, + ) -> dict[str, Any]: + """ + Delete graph data associated with a file or directory and remove source files from disk. + The source files must not be deleted by the caller before invoking this method. + This method is responsible for coordinating graph cleanup and filesystem deletion. + """ + + if self._graph is None: + raise CapabilityError("ingest", "Graph surface is disabled") + if not path or not path.strip(): + raise CapabilityError("ingest", "path is required") + if getattr(self._graph.store, "name", None) != "terminusdb": + raise CapabilityError("ingest", "graph_delete_path is only supported by the TerminusDB graph backend") + + resolved = _resolve_safe_path(path, self._allowed_roots) + if resolved.is_file(): + if resolved.suffix.lower() not in _GRAPH_DOCUMENT_EXTS: + raise CapabilityError( + "ingest", + f"unsupported extension '{resolved.suffix}'. Supported: {sorted(_GRAPH_DOCUMENT_EXTS)}", + ) + candidates = [resolved] + elif resolved.is_dir(): + iterator = resolved.rglob("*") if recursive else resolved.glob("*") + candidates = sorted( + item for item in iterator + if item.is_file() and item.suffix.lower() in _GRAPH_DOCUMENT_EXTS + ) + else: + raise CapabilityError("ingest", f"path does not exist: {resolved}") + + processed: list[dict[str, Any]] = [] + skipped: list[str] = [] + total = 0 + for candidate in candidates: + try: + if candidate.suffix.lower() in _TEXT_EXTS: + content = candidate.read_text(encoding="utf-8", errors="replace") + else: + content = extract_document(candidate).text + if not content.strip(): + skipped.append(f"{candidate}: empty") + candidate.unlink() + continue + result = await self._graph.store.delete_from_source( + source=str(candidate), key="source", value=str(candidate) + ) + deleted = int(result.get("triples_deleted", 0)) + total += deleted + + # Actually delete the source files from disk. + candidate.unlink() + processed.append({ + "path": str(candidate), + "triples_deleted": deleted, + }) + except Exception as exc: + _log.warning( + "graph_delete_file_skipped", + extra={"path": str(candidate), "error": str(exc)}, + ) + skipped.append(f"{candidate}: {type(exc).__name__}: {exc}") + + + result = { + "path": str(resolved), + "files_deleted": len(processed), + "triples_deleted": total, + "files": processed, + "skipped": skipped[:20], + "skipped_count": len(skipped), + } + add_version = getattr(self._graph.store, "add_version", None) + if add_version and callable(add_version): + await add_version(result) + + return result + @staticmethod def _parse_triples_json(raw: str) -> list[dict[str, Any]]: """Valid empty output retracts edges; malformed output must never do so.""" diff --git a/datamind/capabilities/ingest/tools.py b/datamind/capabilities/ingest/tools.py index 23b683e..d229c7e 100644 --- a/datamind/capabilities/ingest/tools.py +++ b/datamind/capabilities/ingest/tools.py @@ -666,7 +666,7 @@ async def _graph_add_path( ), ] - if svc.ingest_mode == "cocoindex": + if svc and svc.ingest_mode == "cocoindex": tools.extend([ ToolSpec( name="kb_update_workspace", @@ -681,6 +681,35 @@ async def _graph_add_path( ), ]) + if ( + svc is not None + and getattr(svc, "_graph", None) is not None + and getattr(svc._graph, "store", None) is not None + and svc._graph.store.name == "terminusdb" + ): + tools.extend([ + ToolSpec( + name="graph_delete_path", + description=( + "Delete a file or directory from the dist workspace and remove " + "the graph triples ingested from that path. " + "When recursive is true, recursively delete directory contents. " + "Only delete paths inside the configured dist workspace. " + "Preserve graph data unrelated to the deleted path." + ), + input_schema={ + "type": "object", + "properties": { + "path": {"type": "string", "description": "Path to the source file or directory to delete."}, + "recursive": {"type": "boolean", "default": True}, + }, + "required": ["path"], + }, + handler=_kb_update_worksapce, + metadata={"group": "ingest", "surface": "graph", "access": "write"}, + ), + ]) + return tools __all__ = ["build_ingest_tools"] diff --git a/datamind/capabilities/kb/providers/chroma_store.py b/datamind/capabilities/kb/providers/chroma_store.py index a21c97a..5204e16 100644 --- a/datamind/capabilities/kb/providers/chroma_store.py +++ b/datamind/capabilities/kb/providers/chroma_store.py @@ -25,7 +25,6 @@ def __init__( dimension: int, ) -> None: import chromadb # type: ignore - self.name = "chroma" self.dimension = dimension self._persist_dir = Path(persist_dir) self._persist_dir.mkdir(parents=True, exist_ok=True) diff --git a/datamind/capabilities/kb/providers/lancedb_store.py b/datamind/capabilities/kb/providers/lancedb_store.py index 48d2284..0247117 100644 --- a/datamind/capabilities/kb/providers/lancedb_store.py +++ b/datamind/capabilities/kb/providers/lancedb_store.py @@ -33,7 +33,6 @@ def __init__( except ImportError as exc: from datamind.core.errors import ConfigError raise ConfigError("LanceDB requires pip install datamind[versioning]") from exc - self.name = "lancedb" self._write_lock = threading.RLock() self.dimension = dimension self._persist_dir = Path(persist_dir) diff --git a/datamind/config.py b/datamind/config.py index f119ca2..5e21c12 100644 --- a/datamind/config.py +++ b/datamind/config.py @@ -123,6 +123,7 @@ class GraphConfig(BaseModel): backend: str = "networkx" # networkx | neo4j | ... dsn: str | None = None # e.g. bolt://user:pw@host:7687 + password: str | None = None # for terminusdb | ... embed_entities: bool = False diff --git a/datamind/core/registry.py b/datamind/core/registry.py index a28ec23..60008eb 100644 --- a/datamind/core/registry.py +++ b/datamind/core/registry.py @@ -46,6 +46,7 @@ def deco(cls: type[T]) -> type[T]: f"new={cls.__module__}.{cls.__name__})" ) self._entries[name] = cls + cls.name = name # Set the backend identifier for runtime backend detection. return cls return deco diff --git a/datamind/direct_tools.py b/datamind/direct_tools.py index 989186e..0107297 100644 --- a/datamind/direct_tools.py +++ b/datamind/direct_tools.py @@ -204,12 +204,12 @@ async def replace_source(path, expected_version, triples): triple.evidence = [GraphEvidence(source_id='file:' + hashlib.sha256(str(source).encode()).hexdigest(), source_version=version, locator={'uri': str(evidence_path), 'original_uri': str(source)}, method='extracted', pipeline_version='codex-explicit-v1')] - await ingest._commit_graph_source(str(source), parsed) + await ingest._commit_graph_source(str(source), parsed, version_info=True) return {'source': str(source), 'source_version': version, 'triples_added': len(parsed)} async def forget_source(path): source = _resolve_safe_path(path, ingest._allowed_roots) - await ingest._commit_graph_source(str(source), []) + await ingest._commit_graph_source(str(source), [], version_info=True) return {'source': str(source), 'retracted': True, 'source_file_deleted': False} return [ diff --git a/datamind/mcp.py b/datamind/mcp.py index 19b1d56..fbb8c1d 100644 --- a/datamind/mcp.py +++ b/datamind/mcp.py @@ -15,6 +15,7 @@ from datamind.config import DataSettings from datamind.core.contracts import ToolAccess from datamind.core.logging import bind_context +from datamind.core.errors import CapabilityError from datamind.data_plane import build_data_plane from datamind.direct_tools import catalogue, workspace_tools from datamind.operations import OperationStore @@ -112,6 +113,11 @@ async def execute(name, arguments): if revision and args.get('pin_workspace') is False: raise ValueError('revision cannot be combined with pin_workspace=false') if mode == 'read' and (revision or args.get('pin_workspace', True)): + # Reconcile the source workspace before acquiring the read lease. + # The lease therefore always pins a complete, immutable revision; + # no graph mutation can occur inside a read operation. + if revision is None: + await workspace.assert_current_revision() lease,revision = workspace.acquire_lease(revision) if revision is None: raise ValueError('No published workspace. Start a write operation to publish one, or set pin_workspace=false for live stores.') @@ -175,10 +181,24 @@ async def execute(name, arguments): if target.startswith('build_'): enable = {'graph'} system = await build_data_plane(settings, enable=enable, workspace_enable=workspace_enable) + track_inventory = ( + spec.access == ToolAccess.WRITE + and target not in {'workspace_publish', 'workspace_migrate'} + ) + before_inventory = ( + await workspace.source_inventory_version() if track_inventory else None + ) try: result = await system.call(target, args, context=ctx, read_only=mode == 'read') finally: await system.aclose() + if before_inventory is not None: + try: + changed = await workspace.source_inventory_version() != before_inventory + except (OSError, CapabilityError): + changed = True + if changed: + workspace.mark_dirty() return {'operation_id': token, 'workspace_revision': None if live else revision, 'result': result} except BaseException as exc: failed=type(exc).__name__ diff --git a/datamind/runtime.py b/datamind/runtime.py index 0e003f7..47d9f99 100644 --- a/datamind/runtime.py +++ b/datamind/runtime.py @@ -80,11 +80,11 @@ async def assemble_data_plane(settings: Settings, *, enable: set[str] | None = N if "memory" in active else None ) ingest = build_ingest_service( - settings=settings, - kb=kb, - db=db, - graph=graph, - llm_client=fallback_client, + settings=settings, + kb=kb, + db=db, + graph=graph, + llm_client=fallback_client, ) services = DataServices( client=client, diff --git a/datamind/server.py b/datamind/server.py index f7bcc35..fa75ea9 100644 --- a/datamind/server.py +++ b/datamind/server.py @@ -397,8 +397,12 @@ async def upload_file( # A same-name upload represents a new revision of the same user document. # Keep its stable path so re-ingestion can replace KB chunks and graph # edges that came from the previous revision. - replaced = target.exists() and target.read_bytes() != body - target.write_bytes(body) + existed = target.exists() + changed = not existed or target.read_bytes() != body + replaced = existed and changed + if changed: + target.write_bytes(body) + Workspace(st.settings).mark_dirty(target) return { "saved_to": str(target), diff --git a/datamind/tests/test_graph_versioning.py b/datamind/tests/test_graph_versioning.py new file mode 100644 index 0000000..276aba8 --- /dev/null +++ b/datamind/tests/test_graph_versioning.py @@ -0,0 +1,288 @@ +"""Integration tests for graph versioning backed by TerminusDB. + +Data is deliberately written through ``IngestService``. That is the public +path which both reconciles a document and registers the resulting TerminusDB +commit in DataMind's graph-version history. + +Run these tests against a disposable TerminusDB instance with:: + + DATAMIND_TEST_TERMINUSDB_DSN=http://127.0.0.1:6363 \ + DATAMIND_TEST_TERMINUSDB_PASSWORD=xxxxxxx pytest -q \ + datamind/tests/test_graph_versioning.py + +The configured database is reset between tests, so the suite is opt-in. +""" +from __future__ import annotations + +import inspect +import importlib.util +import os +import sqlite3 +from pathlib import Path + +import httpx +import pytest +import pytest_asyncio + +from datamind.capabilities.graph.providers.terminusdb_store import ( + TerminusDBGraphStore, +) +from datamind.capabilities.graph.service import GraphService +from datamind.capabilities.ingest.service import IngestService + + +class _Model: + """Deterministic extractor; the entity is selected from the input text.""" + + async def generate_text(self, prompt: str, **kwargs) -> str: + if "Bob" in prompt: + return '[{"subject":"Bob","relation":"owns","object":"Project"}]' + return '[{"subject":"Alice","relation":"owns","object":"Project"}]' + + +async def _maybe_await(value): + """TerminusDB store methods are synchronous while services are async.""" + return await value if inspect.isawaitable(value) else value + + +@pytest.fixture(scope="session") +def terminusdb_config() -> tuple[str, str]: + if importlib.util.find_spec("terminusdb_client") is None: + pytest.skip("terminusdb_client is not installed") + dsn = os.getenv("DATAMIND_TEST_TERMINUSDB_DSN") + if not dsn: + pytest.skip( + "set DATAMIND_TEST_TERMINUSDB_DSN to run destructive TerminusDB tests" + ) + password = os.getenv("DATAMIND_TEST_TERMINUSDB_PASSWORD") + if not password: + pytest.skip("set DATAMIND_TEST_TERMINUSDB_PASSWORD to run TerminusDB tests") + return dsn, password + + +@pytest_asyncio.fixture +async def graph(terminusdb_config, tmp_path: Path): + dsn, password = terminusdb_config + try: + store = TerminusDBGraphStore( + dsn=dsn, + password=password, + persist_path=tmp_path / "terminusdb", + ) + except Exception as exc: # pragma: no cover - depends on local service + pytest.skip(f"TerminusDB is unavailable: {type(exc).__name__}: {exc}") + + # The provider currently owns one fixed database (kg_default). Requiring + # an explicit test DSN above prevents accidental resets of normal setups. + await _maybe_await(store.reset()) + service = GraphService( + store=store, + data_dir=tmp_path / "profile", + storage_dir=tmp_path / "storage", + ) + yield service + await _maybe_await(service.aclose()) + + +@pytest.fixture +def profile(tmp_path: Path) -> Path: + path = tmp_path / "profile" + (path / "uploads").mkdir(parents=True) + return path + + +@pytest.fixture +def ingest(graph: GraphService, profile: Path) -> IngestService: + return IngestService( + kb=None, + db=None, + graph=graph, + llm_client=_Model(), + llm_model="test", + profile_data_dir=profile, + chunk_size=512, + chunk_overlap=64, + ingest_mode="direct", + ) + + +def _write(profile: Path, text: str, name: str = "ownership.md") -> Path: + path = profile / "uploads" / name + path.write_text(text, encoding="utf-8") + return path + + +async def _versions(graph: GraphService) -> list[dict]: + return await _maybe_await(graph.store.list_versions()) + + +async def _latest_version(graph: GraphService) -> str: + versions = await _versions(graph) + assert versions, "ingest did not register a graph version" + version = versions[-1].get("version") + assert isinstance(version, str) and version + return version + + +async def test_list_versions_uses_ingest_order_not_commit_hash_order(tmp_path: Path): + store = TerminusDBGraphStore.__new__(TerminusDBGraphStore) + store._db = "kg_default" + store._history_path = tmp_path / "version_history.sqlite3" + with sqlite3.connect(store._history_path) as conn: + conn.execute( + "CREATE TABLE history (" + "database TEXT, version TEXT, payload TEXT, " + "PRIMARY KEY(database, version))" + ) + conn.execute( + "INSERT INTO history VALUES (?, ?, ?)", + (store._db, "ffff", '{"step": 1}'), + ) + conn.execute( + "INSERT INTO history VALUES (?, ?, ?)", + (store._db, "0000", '{"step": 2}'), + ) + + versions = await store.list_versions() + + assert [item["version"] for item in versions] == ["ffff", "0000"] + assert versions[-1]["payload"] == {"step": 2} + + +@pytest.mark.asyncio +async def test_text_ingest_registers_a_graph_version(ingest, graph): + before = len(await _versions(graph)) + + result = await ingest.graph_add_triples_from_text(text="Alice owns Project") + + assert result["triples_added"] == 1 + versions = await _versions(graph) + assert len(versions) == before + 1 + assert versions[-1]["payload"]["triples_added"] == 1 + assert versions[-1]["version"] + + +@pytest.mark.asyncio +async def test_path_ingest_records_added_and_deleted_triples( + ingest, graph, profile, +): + source = _write(profile, "Alice owns Project") + first = await ingest.graph_add_path(path=str(source)) + assert first["triples_added"] == 1 + + source.write_text("Bob owns Project", encoding="utf-8") + second = await ingest.graph_add_path(path=str(source)) + + assert second["triples_added"] == 1 + version = (await _versions(graph))[-1] + assert version["payload"]["triples_added"] == 1 + assert version["payload"]["triples_deleted"] == 1 + + +@pytest.mark.asyncio +async def test_graph_delete_path_removes_file_edges_and_registers_version( + ingest, graph, profile, +): + source = _write(profile, "Alice owns Project") + await ingest.graph_add_path(path=str(source)) + count_before = len(await _versions(graph)) + + result = await ingest.graph_delete_path(path=str(source)) + + assert result["files_deleted"] == 1 + assert result["triples_deleted"] == 1 + assert result["skipped_count"] == 0 + assert not source.exists() + assert await graph.neighbors("Alice") == [] + versions = await _versions(graph) + assert len(versions) == count_before + 1 + assert versions[-1]["payload"]["triples_deleted"] == 1 + + +@pytest.mark.asyncio +async def test_versioned_search_reads_the_ingest_snapshot(ingest, graph, profile): + source = _write(profile, "Alice owns Project") + await ingest.graph_add_path(path=str(source)) + alice_version = await _latest_version(graph) + + source.write_text("Bob owns Project", encoding="utf-8") + await ingest.graph_add_path(path=str(source)) + + current = await graph.search_entities("Bob", top_k=10) + historical = await graph.search_entities_at_version( + "Alice", top_k=10, version=alice_version + ) + missing_historically = await graph.search_entities_at_version( + "Bob", top_k=10, version=alice_version + ) + + assert {item["id"] for item in current} == {"Bob"} + assert {item["id"] for item in historical} == {"Alice"} + assert missing_historically == [] + + +@pytest.mark.asyncio +async def test_neighbors_at_version_reads_the_ingest_snapshot( + ingest, graph, profile, +): + source = _write(profile, "Alice owns Project") + await ingest.graph_add_path(path=str(source)) + alice_version = await _latest_version(graph) + + source.write_text("Bob owns Project", encoding="utf-8") + await ingest.graph_add_path(path=str(source)) + + historical = await graph.neighbors_at_version( + "Alice", direction="out", version=alice_version + ) + assert [(edge["source"], edge["relation"], edge["target"]) + for edge in historical] == [("Alice", "owns", "Project")] + assert await graph.neighbors_at_version( + "Bob", direction="out", version=alice_version + ) == [] + + +@pytest.mark.asyncio +async def test_traverse_at_version_reads_the_ingest_snapshot( + ingest, graph, profile, +): + source = _write(profile, "Alice owns Project") + await ingest.graph_add_path(path=str(source)) + alice_version = await _latest_version(graph) + + source.write_text("Bob owns Project", encoding="utf-8") + await ingest.graph_add_path(path=str(source)) + + paths = await graph.traverse_at_version( + "Alice", max_hops=1, version=alice_version + ) + assert [path["nodes"] for path in paths] == [["Alice", "Project"]] + assert await graph.traverse_at_version( + "Bob", max_hops=1, version=alice_version + ) == [] + + +@pytest.mark.asyncio +async def test_rollback_restores_snapshot_and_appends_history( + ingest, graph, profile, +): + source = _write(profile, "Alice owns Project") + await ingest.graph_add_path(path=str(source)) + alice_version = await _latest_version(graph) + + source.write_text("Bob owns Project", encoding="utf-8") + await ingest.graph_add_path(path=str(source)) + count_before = len(await _versions(graph)) + + result = await graph.rollback(alice_version) + + assert result["status"] == "ok" + assert result["rollback_version"] == alice_version + assert result["new_version"] != alice_version + + assert len(await _versions(graph)) == count_before + 1 + assert {item["id"] for item in await graph.search_entities("Alice")} == { + "Alice" + } + assert await graph.search_entities("Bob") == [] + diff --git a/datamind/tests/test_workspace_dirty.py b/datamind/tests/test_workspace_dirty.py new file mode 100644 index 0000000..93b7cb2 --- /dev/null +++ b/datamind/tests/test_workspace_dirty.py @@ -0,0 +1,472 @@ +"""Workspace dirty tracking and strict latest-read consistency checks.""" + +import hashlib +import importlib.util +import os +import sqlite3 + +import httpx +import pytest + +from datamind.config import Settings +from datamind.core.errors import CapabilityError, ConfigError +from datamind.capabilities.ingest.lineage import LineageFile, build_lineage_triples +from datamind.workspace import Workspace + + +class Model: + """Minimal deterministic graph extractor used by workspace tests.""" + + async def generate_text(self, prompt, **kwargs): + # The production extractor only needs a JSON array from the client; + # avoid network/model setup in tests. + return '[{"subject":"Alice","relation":"owns","object":"Project"}]' + + async def aclose(self): + pass + + +def graph_source(path): + return [{ + "uri": str(path.resolve()), + "version": "sha256:" + hashlib.sha256(path.read_bytes()).hexdigest(), + "triples": [{"subject": "Alice", "relation": "owns", "object": "Project"}], + }] + + +def dirty(workspace): + with workspace._connect() as db: + return bool(db.execute("SELECT dirty FROM head WHERE singleton=1").fetchone()[0]) + + +def detached(workspace): + with workspace._connect() as db: + return bool(db.execute("SELECT detached FROM head WHERE singleton=1").fetchone()[0]) + + +async def reset_terminusdb(settings, path): + from datamind.capabilities.graph.providers.terminusdb_store import TerminusDBGraphStore + + store = TerminusDBGraphStore( + dsn=settings.graph.dsn, + password=settings.graph.password, + persist_path=path, + ) + try: + await store.reset() + finally: + await store.aclose() + + +def require_terminusdb(): + """Return test credentials without triggering provider auto-installation.""" + if importlib.util.find_spec("terminusdb_client") is None: + pytest.skip("terminusdb_client is not installed") + dsn = os.getenv("DATAMIND_TEST_TERMINUSDB_DSN") + password = os.getenv("DATAMIND_TEST_TERMINUSDB_PASSWORD") + if not dsn or not password: + pytest.skip("TerminusDB integration test credentials are not configured") + try: + response = httpx.get(f"{dsn.rstrip('/')}/api/ok", timeout=2.0) + except httpx.HTTPError as exc: + pytest.skip(f"TerminusDB is not already running: {type(exc).__name__}") + if not response.is_success: + pytest.skip(f"TerminusDB health check returned HTTP {response.status_code}") + return dsn, password + + +@pytest.fixture +def snapshot_workspace(tmp_path): + settings = Settings( + _env_file=None, + llm={"api_key": "test", "model": "test"}, + data={"base_dir": tmp_path / "runtime"}, + workspace={"storage_mode": "snapshot", "auto_gc": False}, + ) + root = (tmp_path / "source").resolve() + root.mkdir() + document = root / "owner.md" + document.write_text("Alice owns Project", encoding="utf-8") + + async def builder(cfg, enable): + from datamind.data_plane import build_data_plane + return await build_data_plane(cfg, enable=enable, text_client=Model()) + + return Workspace(settings, builder=builder), root, document + + +@pytest.mark.asyncio +async def test_mark_dirty_only_for_paths_inside_published_root(snapshot_workspace, tmp_path): + workspace, root, document = snapshot_workspace + await workspace.update(str(root), surfaces=("graph",)) + assert dirty(workspace) is False + + assert workspace.mark_dirty(tmp_path / "outside.md") is False + assert dirty(workspace) is False + assert workspace.mark_dirty(document) is True + assert dirty(workspace) is True + + +@pytest.mark.asyncio +async def test_latest_check_rejects_dirty_without_publishing(snapshot_workspace): + workspace, root, document = snapshot_workspace + published = await workspace.update(str(root), surfaces=("graph",)) + workspace.mark_dirty(document) + + with pytest.raises(CapabilityError, match="workspace_publish"): + await workspace.assert_current_revision() + + assert workspace.current_revision == published["workspace_revision"] + assert dirty(workspace) is True + + +@pytest.mark.asyncio +async def test_latest_check_detects_external_change_without_dirty_flag(snapshot_workspace): + workspace, root, document = snapshot_workspace + published = await workspace.update(str(root), surfaces=("graph",)) + document.write_text("Bob owns Project", encoding="utf-8") + + assert dirty(workspace) is False + with pytest.raises(CapabilityError, match="source files changed"): + await workspace.assert_current_revision() + assert workspace.current_revision == published["workspace_revision"] + + +@pytest.mark.asyncio +async def test_latest_check_allows_clean_rollback_to_immutable_revision(snapshot_workspace): + workspace, root, document = snapshot_workspace + first = await workspace.update(str(root), surfaces=("graph",)) + document.write_text("Bob owns Project", encoding="utf-8") + second = await workspace.update(str(root), surfaces=("graph",)) + + workspace.rollback( + first["workspace_revision"], + expected_revision=second["workspace_revision"], + ) + + assert await workspace.assert_current_revision() == first["workspace_revision"] + async with workspace.reader() as reader: + result = await reader.call("graph_neighbors", entity="Project") + assert "Alice" in str(result) + assert "Bob" not in str(result) + + +@pytest.mark.asyncio +async def test_write_after_rollback_marks_detached_head_dirty(snapshot_workspace): + workspace, root, document = snapshot_workspace + first = await workspace.update(str(root), surfaces=("graph",)) + document.write_text("Bob owns Project", encoding="utf-8") + second = await workspace.update(str(root), surfaces=("graph",)) + workspace.rollback(first["workspace_revision"], expected_revision=second["workspace_revision"]) + + assert detached(workspace) is True + assert dirty(workspace) is False + assert workspace.mark_dirty(document) is True + assert detached(workspace) is True + assert dirty(workspace) is True + with pytest.raises(CapabilityError, match="workspace_publish"): + await workspace.assert_current_revision() + + +@pytest.mark.asyncio +async def test_publish_after_rollback_branches_from_target_and_clears_state(snapshot_workspace): + workspace, root, document = snapshot_workspace + first = await workspace.update(str(root), surfaces=("graph",)) + document.write_text("Bob owns Project", encoding="utf-8") + second = await workspace.update(str(root), surfaces=("graph",)) + workspace.rollback(first["workspace_revision"], expected_revision=second["workspace_revision"]) + document.write_text("Carol owns Project", encoding="utf-8") + workspace.mark_dirty(document) + + third = await workspace.update(str(root), surfaces=("graph",)) + + assert third["parent_revision"] == first["workspace_revision"] + assert workspace.current_revision == third["workspace_revision"] + assert dirty(workspace) is False + assert detached(workspace) is False + async with workspace.reader(second["workspace_revision"]) as reader: + assert "Bob" in (await reader.call("raw_file_read", path=str(document)))["text"] + + +@pytest.mark.asyncio +async def test_failed_publish_preserves_head_and_dirty(snapshot_workspace): + workspace, root, document = snapshot_workspace + first = await workspace.update(str(root), surfaces=("graph",)) + document.write_text("Bob owns Project", encoding="utf-8") + workspace.mark_dirty(document) + + with pytest.raises(ConfigError, match="vocabulary"): + await workspace.update( + str(root), + surfaces=("graph",), + relation_types=["depends_on"], + graph_sources=graph_source(document), + ) + + assert workspace.current_revision == first["workspace_revision"] + assert dirty(workspace) is True + + +@pytest.mark.asyncio +async def test_snapshot_publish_clears_dirty(snapshot_workspace): + workspace, root, document = snapshot_workspace + first = await workspace.update(str(root), surfaces=("graph",)) + document.write_text("Bob owns Project", encoding="utf-8") + workspace.mark_dirty(document) + + second = await workspace.update(str(root), surfaces=("graph",)) + + assert second["workspace_revision"] != first["workspace_revision"] + assert dirty(workspace) is False + assert await workspace.assert_current_revision() == second["workspace_revision"] + + +@pytest.mark.asyncio +async def test_incremental_publish_clears_dirty(tmp_path): + root = (tmp_path / "source").resolve() + root.mkdir() + document = root / "owner.md" + document.write_text("Alice owns Project", encoding="utf-8") + settings = Settings( + _env_file=None, + data={"base_dir": tmp_path / "runtime"}, + workspace={"storage_mode": "incremental", "auto_gc": False}, + kb={"vector_store": "lancedb"}, + db={"dialect": "dolt"}, + ) + workspace = Workspace(settings) + await workspace.update( + str(root), surfaces=("graph",), graph_sources=graph_source(document), + ) + document.write_text("Alice owns Project v2", encoding="utf-8") + workspace.mark_dirty(document) + + result = await workspace.update( + str(root), surfaces=("graph",), graph_sources=graph_source(document), + ) + + assert dirty(workspace) is False + assert await workspace.assert_current_revision() == result["workspace_revision"] + + +def test_existing_catalog_migrates_dirty_column(tmp_path): + settings = Settings(_env_file=None, data={"base_dir": tmp_path / "runtime"}) + root = (settings.data.storage_dir / "workspace").resolve() + root.mkdir(parents=True) + catalog = root / "catalog.sqlite3" + with sqlite3.connect(catalog) as db: + db.executescript(""" + CREATE TABLE revisions (id TEXT PRIMARY KEY, parent TEXT, state TEXT, error TEXT); + CREATE TABLE head (singleton INTEGER PRIMARY KEY, revision TEXT, epoch INTEGER NOT NULL DEFAULT 0); + INSERT INTO head (singleton, revision) VALUES (1, NULL); + """) + + Workspace(settings) + + with sqlite3.connect(catalog) as db: + columns = {row[1] for row in db.execute("PRAGMA table_info(head)")} + values = db.execute( + "SELECT dirty, detached FROM head WHERE singleton=1" + ).fetchone() + assert "dirty" in columns + assert "detached" in columns + assert values == (0, 0) + + +@pytest.mark.asyncio +async def test_mcp_latest_read_rejects_dirty_workspace(snapshot_workspace, monkeypatch): + pytest.importorskip("jsonschema") + from datamind import mcp + + workspace, root, document = snapshot_workspace + await workspace.update(str(root), surfaces=("graph",)) + workspace.mark_dirty(document) + monkeypatch.setattr(mcp, "load_settings", lambda profile=None: workspace.settings) + + with pytest.raises(CapabilityError, match="workspace_publish"): + await mcp.execute("datamind_operation_begin", {"mode": "read"}) + + await workspace.update(str(root), surfaces=("graph",)) + document.write_text("Bob owns Project", encoding="utf-8") + with pytest.raises(CapabilityError, match="workspace_publish"): + await mcp.execute("datamind_operation_begin", {"mode": "read"}) + + +@pytest.mark.asyncio +async def test_mcp_explicit_revision_reads_while_workspace_is_dirty(snapshot_workspace, monkeypatch): + pytest.importorskip("jsonschema") + from datamind import mcp + + workspace, root, document = snapshot_workspace + published = await workspace.update(str(root), surfaces=("graph",)) + document.write_text("Bob owns Project", encoding="utf-8") + workspace.mark_dirty(document) + monkeypatch.setattr(mcp, "load_settings", lambda profile=None: workspace.settings) + + operation = await mcp.execute( + "datamind_operation_begin", + {"mode": "read", "revision": published["workspace_revision"]}, + ) + + assert operation["revision"] == published["workspace_revision"] + await mcp.execute( + "datamind_operation_close", + {"operation_id": operation["operation_id"]}, + ) + +@pytest.mark.asyncio +async def test_pinned_read_ignores_current_source_changes(snapshot_workspace): + workspace, root, document = snapshot_workspace + first = await workspace.update(str(root), surfaces=("graph",)) + document.write_text("Bob owns Project", encoding="utf-8") + + async with workspace.reader(first["workspace_revision"]) as reader: + result = await reader.call("graph_neighbors", entity="Project") + + assert "Alice" in str(result) + assert "Bob" not in str(result) + + +def test_lineage_evidence_covers_all_input_files(tmp_path): + left = tmp_path / "left.csv" + right = tmp_path / "right.csv" + left.write_text("id,name\n1,Alice\n", encoding="utf-8") + right.write_text("id,name\n2,Bob\n", encoding="utf-8") + files = [ + LineageFile( + uri=str(path), + relative_path=path.name, + version="sha256:" + hashlib.sha256(path.read_bytes()).hexdigest(), + content_path=path, + ) + for path in (left, right) + ] + + triples = build_lineage_triples(root=tmp_path, files=files) + overlap = next(triple for triple in triples if triple.relation == "schema_overlap") + + assert {item.locator["uri"] for item in overlap.evidence} == { + str(left), str(right) + } + assert all(item.method == "inferred" for item in overlap.evidence) + assert all(item.pipeline_version == "lineage-v1" for item in overlap.evidence) + + +@pytest.mark.asyncio +async def test_v1_reader_includes_lineage(snapshot_workspace): + workspace, root, document = snapshot_workspace + related = root / "project.md" + related.write_text("See owner.md", encoding="utf-8") + revision = await workspace.update(str(root), surfaces=("graph",)) + + async with workspace.reader(revision["workspace_revision"]) as reader: + schema = await reader.call("graph_schema") + + assert schema["relations"]["contains"] == 2 + assert schema["relations"]["mentions"] == 1 + + +@pytest.mark.asyncio +async def test_v2_reader_includes_lineage(tmp_path): + root = (tmp_path / "source").resolve() + root.mkdir() + document = root / "owner.md" + related = root / "project.md" + document.write_text("Alice owns Project", encoding="utf-8") + related.write_text("See owner.md", encoding="utf-8") + settings = Settings( + _env_file=None, + data={"base_dir": tmp_path / "runtime"}, + workspace={"storage_mode": "incremental", "auto_gc": False}, + kb={"vector_store": "lancedb"}, + db={"dialect": "dolt"}, + ) + workspace = Workspace(settings) + assertions = [ + { + "uri": str(path), + "version": "sha256:" + hashlib.sha256(path.read_bytes()).hexdigest(), + "triples": [], + } + for path in (document, related) + ] + revision = await workspace.update(str(root), surfaces=("graph",), graph_sources=assertions) + + async with workspace.reader(revision["workspace_revision"]) as reader: + schema = await reader.call("graph_schema") + + assert schema["relations"]["contains"] == 2 + assert schema["relations"]["mentions"] == 1 + + +@pytest.mark.asyncio +async def test_terminusdb_migration_requires_target_backend(tmp_path): + """Provider conversion is explicit and only runs with TerminusDB configured.""" + dsn, password = require_terminusdb() + root = (tmp_path / "source").resolve() + root.mkdir() + document = root / "owner.md" + document.write_text("Alice owns Project", encoding="utf-8") + settings = Settings( + _env_file=None, + data={"base_dir": tmp_path / "runtime"}, + workspace={"storage_mode": "snapshot", "auto_gc": False}, + graph={"backend": "networkx"}, + ) + async def builder(cfg, enable): + from datamind.data_plane import build_data_plane + return await build_data_plane(cfg, enable=enable, text_client=Model()) + workspace = Workspace(settings, builder=builder) + first = await workspace.update(str(root), surfaces=("graph",)) + settings.graph.backend = "terminusdb" + settings.graph.dsn = dsn + settings.graph.password = password + await reset_terminusdb(settings, tmp_path / "terminusdb-reset") + workspace.settings = settings + migrated = await workspace.migrate() + + assert migrated["migrated_from"] == first["workspace_revision"] + assert migrated["snapshots"]["graph"]["provider"] == "terminusdb" + async with workspace.reader(first["workspace_revision"]) as before: + before_schema = await before.call("graph_schema") + before_neighbors = await before.call("graph_neighbors", entity="Project") + async with workspace.reader(migrated["workspace_revision"]) as after: + after_schema = await after.call("graph_schema") + after_neighbors = await after.call("graph_neighbors", entity="Project") + assert after_schema["relations"] == before_schema["relations"] + assert "Alice" in str(before_neighbors) + assert "Alice" in str(after_neighbors) + + +@pytest.mark.asyncio +async def test_terminusdb_incremental_can_extract_without_graph_sources(tmp_path): + """TerminusDB mode may use the ingest extractor when assertions are omitted.""" + dsn, password = require_terminusdb() + root = (tmp_path / "source").resolve() + root.mkdir() + document = root / "owner.md" + document.write_text("Alice owns Project", encoding="utf-8") + settings = Settings( + _env_file=None, + llm={"api_key": "test", "model": "test"}, + data={"base_dir": tmp_path / "runtime"}, + workspace={"storage_mode": "incremental", "auto_gc": False}, + graph={ + "backend": "terminusdb", + "dsn": dsn, + "password": password, + }, + ) + + async def builder(cfg, enable): + from datamind.data_plane import build_data_plane + return await build_data_plane(cfg, enable=enable, text_client=Model()) + + await reset_terminusdb(settings, tmp_path / "terminusdb-reset") + workspace = Workspace(settings, builder=builder) + result = await workspace.update(str(root), surfaces=("graph",), graph_sources=None) + + assert result["snapshots"]["graph"]["provider"] == "terminusdb" + async with workspace.reader(result["workspace_revision"]) as reader: + assert "Alice" in str(await reader.call("graph_neighbors", entity="Project")) + diff --git a/datamind/tests/test_workspace_incremental.py b/datamind/tests/test_workspace_incremental.py index b46854b..6159e37 100644 --- a/datamind/tests/test_workspace_incremental.py +++ b/datamind/tests/test_workspace_incremental.py @@ -2,12 +2,15 @@ import asyncio import hashlib +from contextlib import asynccontextmanager +from types import SimpleNamespace import pytest from datamind.config import Settings from datamind.core.errors import CapabilityError from datamind.workspace import Workspace +from datamind.workspace_native import prune_native, TerminusDBRepository class Embed: @@ -23,6 +26,88 @@ async def embed_query(self, text): return [1.0, 0.0, 0.0, 0.0] +@pytest.mark.asyncio +async def test_terminusdb_gc_removes_unretained_revision_branches(tmp_path, monkeypatch): + root = tmp_path / "runtime" + (root / "shared" / "terminusdb").mkdir(parents=True) + workspace = SimpleNamespace(root=root) + workspace.settings = SimpleNamespace( + workspace=SimpleNamespace(lock_timeout_s=5), + ) + + class Client: + def __init__(self): + self.deleted = [] + + def get_all_branches(self): + return [ + {"name": "main"}, + {"name": "dm_build_keep"}, + {"name": "dm_build_remove"}, + {"name": "feature"}, + ] + + def delete_branch(self, name): + self.deleted.append(name) + + client = Client() + store = SimpleNamespace(_client=client) + + @asynccontextmanager + async def session(self, revision=None): + yield store + + monkeypatch.setattr(TerminusDBRepository, "session", session) + manifests = [ + { + "format_version": 2, + "snapshots": { + "graph": {"provider": "terminusdb", "revision": "revision-keep"} + }, + } + ] + + report = await prune_native(workspace, manifests) + + assert client.deleted == ["dm_build_remove"] + assert report["graph"] == { + "retained_branches": ["dm_build_keep"], + "removed_branches": ["dm_build_remove"], + "physical_gc_pending": True, + } + + +@pytest.mark.asyncio +async def test_terminusdb_gc_keeps_branches_for_legacy_manifests(tmp_path, monkeypatch): + root = tmp_path / "runtime" + (root / "shared" / "terminusdb").mkdir(parents=True) + workspace = SimpleNamespace(root=root) + workspace.settings = SimpleNamespace( + workspace=SimpleNamespace(lock_timeout_s=5), + ) + + class Client: + def get_all_branches(self): + return [{"name": "dm_build_old"}] + + def delete_branch(self, name): + raise AssertionError("legacy revision branch must be retained") + + @asynccontextmanager + async def session(self, revision=None): + yield SimpleNamespace(_client=Client()) + + monkeypatch.setattr(TerminusDBRepository, "session", session) + manifests = [{ + "format_version": 2, + "snapshots": {"graph": {"provider": "terminusdb", "version": "commit"}}, + }] + + report = await prune_native(workspace, manifests) + + assert report["graph"]["removed_branches"] == [] + + @pytest.fixture def incremental(tmp_path, monkeypatch): from datamind import runtime diff --git a/datamind/workspace.py b/datamind/workspace.py index 5a1a081..7ddc381 100644 --- a/datamind/workspace.py +++ b/datamind/workspace.py @@ -26,6 +26,7 @@ from datamind.core.errors import CapabilityError, ConfigError, ExternalServiceError from datamind.core.logging import bind_context, current_context from datamind.core.protocols import GraphTriple, GraphEvidence +from datamind.capabilities.ingest.lineage import LineageFile, build_lineage_triples from datamind.core.recovery import recover_read, recovery_summary from datamind.workspace_catalog import WorkspaceCatalog @@ -66,11 +67,15 @@ def __init__(self, settings, *, builder=None): conn.executescript(''' CREATE TABLE IF NOT EXISTS revisions ( id TEXT PRIMARY KEY, parent TEXT, state TEXT, error TEXT); - CREATE TABLE IF NOT EXISTS head (singleton INTEGER PRIMARY KEY, revision TEXT, epoch INTEGER NOT NULL DEFAULT 0); + CREATE TABLE IF NOT EXISTS head (singleton INTEGER PRIMARY KEY, revision TEXT, epoch INTEGER NOT NULL DEFAULT 0, dirty INTEGER NOT NULL DEFAULT 0, detached INTEGER NOT NULL DEFAULT 0); INSERT OR IGNORE INTO head (singleton, revision) VALUES (1, NULL); ''') if 'epoch' not in {row[1] for row in conn.execute('PRAGMA table_info(head)')}: conn.execute('ALTER TABLE head ADD COLUMN epoch INTEGER NOT NULL DEFAULT 0') + if 'dirty' not in {row[1] for row in conn.execute('PRAGMA table_info(head)')}: + conn.execute('ALTER TABLE head ADD COLUMN dirty INTEGER NOT NULL DEFAULT 0') + if 'detached' not in {row[1] for row in conn.execute('PRAGMA table_info(head)')}: + conn.execute('ALTER TABLE head ADD COLUMN detached INTEGER NOT NULL DEFAULT 0') self._init_catalog() def _connect(self): @@ -81,6 +86,65 @@ def current_revision(self) -> str | None: with self._connect() as conn: return conn.execute('SELECT revision FROM head WHERE singleton=1').fetchone()[0] + def mark_dirty(self, path: str | Path | None = None) -> bool: + """Mark the published workspace dirty when a source path was mutated.""" + revision = self.current_revision + if revision is None: + return False + manifest = self.manifest(revision) + if path is not None: + source = Path(manifest['source_root']).resolve() + candidate = Path(path).expanduser().resolve() + if candidate != source and not candidate.is_relative_to(source): + return False + with self._connect() as conn: + conn.execute('UPDATE head SET dirty=1 WHERE singleton=1') + return True + + async def source_inventory_version(self) -> str | None: + """Hash the current source inventory for mutation-boundary checks.""" + revision = self.current_revision + if revision is None: + return None + source = Path(self.manifest(revision)['source_root']) + inventory = await asyncio.to_thread(self._inventory, source) + return _digest(json.dumps([ + (item['relative_path'], item['uri'], item['version']) + for item in inventory + ], ensure_ascii=False, separators=(',', ':')).encode()) + + async def assert_current_revision(self) -> str | None: + """Reject latest reads when source files differ from the published revision.""" + revision = self.current_revision + if revision is None: + return None + with self._connect() as conn: + dirty, detached = conn.execute( + 'SELECT dirty, detached FROM head WHERE singleton=1' + ).fetchone() + if detached and not dirty: + return revision + manifest = self.manifest(revision) + source = Path(manifest['source_root']) + current = await asyncio.to_thread(self._inventory, source) + + published = _digest(json.dumps([ + (item['relative_path'], item['uri'], item['version']) + for item in manifest.get('sources', []) + ], ensure_ascii=False, separators=(',', ':')).encode()) + existing = _digest(json.dumps([ + (item['relative_path'], item['uri'], item['version']) + for item in current + ], ensure_ascii=False, separators=(',', ':')).encode()) + + if not dirty and published == existing: + return revision + raise CapabilityError( + 'workspace', + 'workspace source files changed after the latest published revision; ' + 'run workspace_publish in a write operation before reading latest data', + ) + def status(self, revision: str) -> dict[str, Any]: with self._connect() as conn: row = conn.execute('SELECT id, parent, state, error FROM revisions WHERE id=?', (revision,)).fetchone() @@ -106,7 +170,11 @@ def _rollback(self, revision, *, expected_revision): head = conn.execute('SELECT revision FROM head WHERE singleton=1').fetchone()[0] if expected_revision is not None and head != expected_revision: raise CapabilityError('workspace', 'publication conflict; current revision changed') - conn.execute('UPDATE head SET revision=?, epoch=epoch+1 WHERE singleton=1', (revision,)) + conn.execute( + 'UPDATE head SET revision=?, epoch=epoch+1, dirty=0, detached=1 ' + 'WHERE singleton=1', + (revision,), + ) return {'workspace_revision': revision, 'previous_revision': head} def _settings_for(self, revision: str): @@ -333,6 +401,10 @@ async def _update(self, path: str, *, surfaces, relation_types, max_triples_per_ source_version=item['version'], locator={'part': i}, relation_types=relation_types, ) + for triple in triples: + for evidence in (triple.evidence or []): + evidence.pipeline_version = "legacy-model" + evidence.method = "extracted" prepared['triples'].extend(t.model_dump(mode='json') for t in triples) if len(triples) >= min(30, remaining): prepared['truncated'] = True @@ -392,6 +464,15 @@ async def _update(self, path: str, *, surfaces, relation_types, max_triples_per_ metrics['materialization_ms'] += (time.monotonic() - stage_started) * 1000 stage_started = time.monotonic() if 'graph' in selected: + lineage = build_lineage_triples( + root=source, + files=[LineageFile( + uri=item['uri'], relative_path=item['relative_path'], + version=item['version'], + content_path=Path(settings.data.base_dir) / item['snapshot'], + ) for item in manifest_sources], + ) + graph_triples.extend(lineage) await system.services.graph.store.upsert_triples(graph_triples) await system.services.graph.store.persist() if 'kb' in selected: @@ -404,6 +485,10 @@ async def _update(self, path: str, *, surfaces, relation_types, max_triples_per_ graph_path = settings.data.storage_dir / 'graph.json' bindings['graph'] = {'provider': 'networkx', 'path': str(graph_path.relative_to(settings.data.base_dir)), + 'lineage_version': 'sha256:' + _digest(json.dumps( + [(t.subject, t.relation, t.object) for t in lineage], + ensure_ascii=False, separators=(',', ':'), + ).encode()), 'sha256': _digest(graph_path.read_bytes())} if 'db' in selected: db_path = Path(system.services.db.engine.url.database) @@ -427,7 +512,11 @@ async def _update(self, path: str, *, surfaces, relation_types, max_triples_per_ if (current, epoch) != (parent, parent_epoch): raise CapabilityError('workspace', 'publication conflict; another build published first') conn.execute('UPDATE revisions SET state=? WHERE id=?', ('PUBLISHED', revision)) - conn.execute('UPDATE head SET revision=?, epoch=epoch+1 WHERE singleton=1', (revision,)) + conn.execute( + 'UPDATE head SET revision=?, epoch=epoch+1, dirty=0, detached=0 ' + 'WHERE singleton=1', + (revision,), + ) metrics['publication_ms'] = (time.monotonic() - stage_started) * 1000 return manifest except BaseException as exc: diff --git a/datamind/workspace_graph.py b/datamind/workspace_graph.py index 49f5a8e..722f235 100644 --- a/datamind/workspace_graph.py +++ b/datamind/workspace_graph.py @@ -10,7 +10,6 @@ from pathlib import Path from datamind.core.protocols import Edge, Entity, GraphPath, GraphTriple - def connect(root): db = sqlite3.connect(Path(root) / "graph.sqlite3", timeout=30) db.execute("PRAGMA journal_mode=WAL") @@ -71,6 +70,7 @@ def prune_graph(root, manifests, *, compact=False): for s in m["sources"] if s.get("graph_artifact") } + keep.update(m.get("lineage_artifact") for m in manifests if m.get("lineage_artifact")) with closing(connect(root)) as db, db: for (key,) in db.execute("SELECT id FROM artifacts").fetchall(): if key not in keep: @@ -111,7 +111,7 @@ def stats(self): ).fetchone()[0], } - def schema(self): + async def schema(self): stats = self.stats() types = Counter(typ for _, typ in self._nodes()) relations = dict( diff --git a/datamind/workspace_incremental.py b/datamind/workspace_incremental.py index aa13920..006744a 100644 --- a/datamind/workspace_incremental.py +++ b/datamind/workspace_incremental.py @@ -200,15 +200,18 @@ async def update_incremental( relation_types, max_triples_per_file, metrics, - graph_sources, + graph_sources=None, table_mappings=None, ): from datamind.workspace import _atomic_json + from datamind.capabilities.ingest.lineage import LineageFile, build_lineage_triples from datamind.workspace_graph import put_graph - from datamind.workspace_native import DoltRepository + from datamind.workspace_native import DoltRepository, TerminusDBRepository selected = set(surfaces) cfg = workspace.settings + terminus_graph = "graph" in selected and cfg.graph.backend == "terminusdb" + auto_graph_extraction = terminus_graph and graph_sources is None if not selected or not selected <= {"kb", "graph", "db"}: raise ConfigError("Invalid workspace surfaces") if not 1 <= max_triples_per_file <= 10000: @@ -221,8 +224,11 @@ async def update_incremental( raise ConfigError( "Incremental Workspace DB requires local Dolt without an external DSN" ) - if "graph" in selected and graph_sources is None: - raise ConfigError("Incremental Workspace requires Codex-prepared graph_sources") + if "graph" in selected and graph_sources is None and not auto_graph_extraction: + raise ConfigError( + "Incremental Workspace requires Codex-prepared graph_sources " + "or automatic graph extraction" + ) source = Path(path).expanduser().resolve() started = time.monotonic() inventory = await blocking(workspace._inventory, source) @@ -301,7 +307,7 @@ async def update_incremental( "triples": [t.model_dump(mode="json") for t in triples], "truncated": bool(entry.get("truncated")), } - if "graph" in selected: + if "graph" in selected and not auto_graph_extraction: for uri, item in docs.items(): if uri not in assertions: before = prior.get(uri) @@ -314,6 +320,35 @@ async def update_incremental( "Supply graph assertions for every new or changed document, including empty lists" ) # Unchanged assertions are immutable and reused, not re-extracted. + if auto_graph_extraction: + try: + system = await workspace._system(cfg, {"graph"}) + ingest = system.services.ingest + for uri, item in docs.items(): + before = prior.get(uri) + if ( + before is None + or before["version"] != item["version"] + or not before.get("graph_triples") + ): + results = await ingest.graph_add_path( + path=uri, recursive=False, max_triples_per_file=max_triples_per_file, + relation_types=relation_types, commit=False + ) + file_result = results["files"][0] + triples = file_result.get("triples", []) + for triple in triples: + for evidence in triple.get("evidence", []): + evidence["pipeline_version"] = "legacy-model" + evidence["method"] = "extracted" + + assertions[uri] = { + "triples": triples, + "truncated": bool(file_result.get("truncated")), + } + finally: + await system.aclose() + system = None context = current_context() skills = list(context.extra.get("applied_skills", {}).values()) if context else [] recipe = { @@ -348,6 +383,7 @@ async def update_incremental( embedding = None sources = [] bindings = {} + source_graphs = {} all_chunks = {} tables = {} try: @@ -433,10 +469,16 @@ async def update_incremental( elif item["uri"] in getattr(workspace, "_migration_prepared", {}): prepared = dict(workspace._migration_prepared[item["uri"]]) if "graph" in selected: - prepared["graph_artifact"] = put_graph( - workspace.root, - assertion["triples"] if assertion else [], - ) + if not terminus_graph: + prepared["graph_artifact"] = put_graph( + workspace.root, + assertion["triples"] if assertion else [], + ) + else: + prepared["graph_triples"] = assertion["triples"] if assertion else [] + prepared["graph_artifact"] = None + + prepared["truncated"] = assertion["truncated"] if assertion else False artifact = workspace.root / save_artifact( workspace.root, key, prepared ) @@ -445,6 +487,7 @@ async def update_incremental( prepared = { "chunks": [], "tables": [], + "graph_triples": [], "graph_artifact": None, "truncated": False, } @@ -524,17 +567,21 @@ async def update_incremental( for i, part in enumerate(parts) ] if "graph" in selected: - if assertion is not None: - prepared["graph_artifact"] = put_graph( - workspace.root, assertion["triples"] - ) - prepared["truncated"] = assertion["truncated"] - elif graph_ref: - prepared["graph_artifact"] = graph_ref + if not terminus_graph: + if assertion is not None: + prepared["graph_artifact"] = put_graph( + workspace.root, assertion["triples"] + ) + prepared["truncated"] = assertion["truncated"] + elif graph_ref: + prepared["graph_artifact"] = graph_ref + else: + prepared["graph_artifact"] = put_graph( + workspace.root, [] + ) else: - prepared["graph_artifact"] = put_graph( - workspace.root, [] - ) + prepared["graph_triples"] = assertion["triples"] + prepared["truncated"] = assertion["truncated"] if "db" in selected and suffix in TABLE_EXTS: previous = prior.get(item["uri"], {}) old_tables = [] @@ -565,6 +612,10 @@ async def update_incremental( "truncated": prepared["truncated"], } sources.append(source_entry) + if terminus_graph: + source_graphs[item["uri"]] = [ + GraphTriple(**value) for value in prepared.get("graph_triples", []) + ] all_chunks.update((c["id"], c) for c in prepared["chunks"]) tables.update((t["name"], t) for t in prepared["tables"]) metrics["sources_completed"] += 1 @@ -573,7 +624,6 @@ async def update_incremental( from datamind.capabilities.kb.providers.lancedb_store import ( LanceDBVectorStore, ) - kb_key = digest( encoded( [ @@ -672,10 +722,39 @@ async def update_incremental( + len(before.keys() - hashes.keys()) ) if "graph" in selected: - bindings["graph"] = { - "provider": "sqlite-source-graph", - "path": "graph.sqlite3", - } + lineage = build_lineage_triples(root=source, files=[ + LineageFile( + uri=item["uri"], relative_path=item["relative_path"], + version=item["version"], + content_path=workspace.root / item["snapshot"], + ) + for item in sources + ]) + + lineage_artifact = None if terminus_graph else put_graph( + workspace.root, [t.model_dump(mode="json") for t in lineage] + ) + lineage_version = "sha256:" + digest(encoded([ + (t.subject, t.relation, t.object) for t in lineage + ])) + if terminus_graph: + version = await TerminusDBRepository(workspace).build( + root=source, prior=prior, sources=sources, + source_graphs=source_graphs, lineage_graphs=lineage, + revision=revision + ) + bindings["graph"] = { + "provider": "terminusdb", "version": version, + "revision": revision, + "lineage_version": lineage_version + } + else: + bindings["graph"] = { + "provider": "sqlite-source-graph", + "path": "graph.sqlite3", + "lineage_artifact": lineage_artifact, + "lineage_version": lineage_version + } metrics["materialization_ms"] += (time.monotonic() - start) * 1000 if await blocking(workspace._inventory, source) != inventory: raise CapabilityError( @@ -691,6 +770,10 @@ async def update_incremental( "pipeline_hash": pipeline, "sources": sources, "snapshots": bindings, + "lineage_artifact": lineage_artifact if "graph" in selected else None, + "lineage_version": ( + lineage_version if "graph" in selected else None + ), "reused_sources": metrics["sources_reused"], "applied_skills": skills, "truncated_sources": [s["uri"] for s in sources if s["truncated"]], @@ -710,7 +793,7 @@ async def update_incremental( "UPDATE revisions SET state='PUBLISHED' WHERE id=?", (revision,) ) db.execute( - "UPDATE head SET revision=?,epoch=epoch+1 WHERE singleton=1", + "UPDATE head SET revision=?,epoch=epoch+1,dirty=0,detached=0 WHERE singleton=1", (revision,), ) metrics["publication_ms"] = (time.monotonic() - start) * 1000 @@ -735,7 +818,7 @@ async def reader_system(workspace, manifest, selected): from datamind.capabilities.kb.providers.lancedb_store import LanceDBVectorStore from datamind.core.registry import retriever_registry from datamind.core.contracts import ToolAccess - from datamind.workspace_native import PinnedVectors, PinnedDB + from datamind.workspace_native import PinnedVectors, PinnedDB, PinnedTerminusGraph from datamind.workspace_graph import GraphView from datamind import runtime @@ -802,18 +885,30 @@ async def documents(): [s for s in build_kb_tools(kb) if s.access != ToolAccess.WRITE] ) if "graph" in selected: - graph = GraphService( - store=GraphView( - workspace.root, - { - s["graph_artifact"] - for s in manifest["sources"] - if s.get("graph_artifact") - }, - ), - data_dir=settings.data.data_dir, - storage_dir=workspace.root, - ) + binding = manifest["snapshots"]["graph"] + if binding["provider"] == "terminusdb": + graph = GraphService( + store=PinnedTerminusGraph( + workspace, binding["version"], binding.get("revision") + ), + data_dir=settings.data.data_dir, + storage_dir=workspace.root, + ) + else: + artifacts = { + s["graph_artifact"] + for s in manifest["sources"] + if s.get("graph_artifact") + } + lineage_artifact = binding.get("lineage_artifact") or manifest.get("lineage_artifact") + if lineage_artifact: + artifacts.add(lineage_artifact) + store = GraphView(workspace.root, artifacts) + graph = GraphService( + store=store, + data_dir=settings.data.data_dir, + storage_dir=workspace.root, + ) system.services.graph = graph system.tools.extend( [s for s in build_graph_tools(graph) if s.access != ToolAccess.WRITE] diff --git a/datamind/workspace_migration.py b/datamind/workspace_migration.py index 68e3750..9aa23ba 100644 --- a/datamind/workspace_migration.py +++ b/datamind/workspace_migration.py @@ -149,6 +149,12 @@ async def _migrate(workspace): store = services.graph.store for source, target, data in store._g.edges(data=True): edge = store._edge(source, target, data) + # V1 stores deterministic workspace lineage in the same + # graph as document assertions. V2 rebuilds lineage from + # the frozen inventory, so it must not be assigned to a + # document source or migrated as a user assertion. + if edge.properties.get("_lineage_root"): + continue uri = edge.properties.get("source") if uri not in graph: raise ConfigError("Legacy graph source is absent from manifest") diff --git a/datamind/workspace_native.py b/datamind/workspace_native.py index 2ad3722..a62471b 100644 --- a/datamind/workspace_native.py +++ b/datamind/workspace_native.py @@ -15,6 +15,7 @@ import time import socket import subprocess +from collections import Counter from contextlib import asynccontextmanager from datetime import timedelta from filelock import AsyncFileLock @@ -22,6 +23,7 @@ from sqlalchemy.engine import make_url from datamind.core.errors import CapabilityError from datamind.capabilities.db.providers.dolt import DoltDialect +from datamind.capabilities.graph.providers.terminusdb_store import TerminusDBGraphStore async def completed(awaitable): @@ -485,4 +487,185 @@ def gc(): "retained_versions": sorted(versions), "physical_gc_pending": not compact, } + graph_bindings = [ + m.get("snapshots", {}).get("graph", {}) + for m in manifests + if m.get("format_version") == 2 + ] + graph_bindings = [b for b in graph_bindings if b.get("provider") == "terminusdb"] + graph_path = workspace.root / "shared" / "terminusdb" + if graph_path.is_dir() and graph_bindings: + # Older manifests do not contain the branch name. Keep all branches + # until those revisions are migrated, rather than deleting blindly. + unknown = any("revision" not in binding for binding in graph_bindings) + retained = { + "dm_build_" + str(binding["revision"]).removeprefix("revision-") + for binding in graph_bindings + if binding.get("revision") is not None + } + repo = TerminusDBRepository(workspace) + removed = [] + async with repo.session() as store: + branches = await blocking(store._client.get_all_branches) + for item in branches: + name = item.get("name") + if ( + not isinstance(name, str) + or not name.startswith("dm_build_") + or unknown + or name in retained + ): + continue + await blocking(store._client.delete_branch, name) + removed.append(name) + report["graph"] = { + "retained_branches": sorted(retained), + "removed_branches": sorted(removed), + "physical_gc_pending": True, + } return report + + +class TerminusDBRepository: + def __init__(self, workspace): + self.workspace = workspace + self.path = workspace.root / "shared" / "terminusdb" + self.path.mkdir(parents=True, exist_ok=True) + + def _revision_branch(self, revision: str) -> str: + return f"dm_build_{revision.removeprefix('revision-')}" + + def _new_store(self, branch=None): + kwargs = { + "dsn": self.workspace.settings.graph.dsn, + "password": self.workspace.settings.graph.password, + "persist_path": self.path, + } + if branch is not None: + kwargs["branch"] = branch + return TerminusDBGraphStore(**kwargs) + + @asynccontextmanager + async def session(self, revision=None): + async with AsyncFileLock( + str(self.workspace.root / "terminusdb.lock"), + timeout=self.workspace.settings.workspace.lock_timeout_s, + ): + store = None + try: + branch = self._revision_branch(revision) if revision is not None else None + # Client construction performs HTTP requests, schema checks and + # possibly process startup; never block the event loop on it. + store = await blocking(self._new_store, branch) + yield store + finally: + # TerminusDB is shared by repository sessions. The provider's + # ``aclose`` stops a managed server, so calling it here would + # tear down the server after every query and force a restart on + # the next one. Server lifetime is managed by the application. + pass + + async def create_revision_branch( + self, + branch: str, + ): + async with self.session() as store: + client = store._client + + branches = await blocking(client.get_all_branches) + existing = { + item["name"] for item in branches + } + + if branch not in existing: + await blocking(client.create_branch, branch) + + async def delete_revision_branch(self, branch: str): + async with self.session() as store: + client = store._client + + branches = await blocking(client.get_all_branches) + existing = { + item["name"] for item in branches + } + + if branch in existing: + await blocking(client.delete_branch, branch) + + async def build(self, root, prior, sources, source_graphs, lineage_graphs, revision): + branch = self._revision_branch(revision) + await self.create_revision_branch(branch) + try: + async with self.session(revision) as store: + current_uris = {item["uri"] for item in sources} + previous_uris = set(prior) + changes = [] + for item in sources: + changes.append(await store.reconcile_source_triples( + item["uri"], source_graphs.get(item["uri"], []) + )) + for uri in previous_uris - current_uris: + changes.append(await store.reconcile_source_triples(uri, [])) + changes.append(await store.reconcile_lineage_triples(str(root), lineage_graphs)) + await store.add_version({"kind": "workspace", "workspace_revision": revision, + "lineage_artifact": None, "changes": changes}) + + version = store._get_latest_version() + + if not version: + raise RuntimeError( + f"TerminusDB commit not found for revision {revision}" + ) + + return version + except BaseException: + # A successful commit must remain reachable through its revision + # branch so pinned reads can resolve it later. Remove only failed + # staging branches; retention/GC owns successful branch cleanup. + await self.delete_revision_branch(branch) + raise + +class PinnedTerminusGraph: + """Read-only GraphStore view pinned to one immutable TerminusDB commit.""" + + def __init__(self, workspace, version, revision=None): + self.repository = TerminusDBRepository(workspace) + self._version = version + self._revision = revision + + def _session(self): + return self.repository.session(self._revision) + + async def _documents(self): + async with self._session() as store: + return store._documents_at_version(self._version) + + async def search_entities(self, query, *, top_k=5): + async with self._session() as store: + return await store.search_entities(query, top_k=top_k, version=self._version) + + async def neighbors(self, entity, *, direction="both", relation_filter=None, limit=100): + async with self._session() as store: + return await store.neighbors( + entity, direction=direction, relation_filter=relation_filter, + limit=limit, version=self._version, + ) + + async def traverse(self, start, *, max_hops=2, relation_filter=None, max_results=100): + async with self._session() as store: + return await store.traverse( + start, max_hops=max_hops, relation_filter=relation_filter, + max_results=max_results, version=self._version, + ) + + async def stats(self): + async with self._session() as store: + return await store.stats(version=self._version) + + async def schema(self): + async with self._session() as store: + return await store.schema(version=self._version) + + async def aclose(self): + pass + diff --git a/pyproject.toml b/pyproject.toml index ebd4e46..e96f078 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -91,6 +91,7 @@ cocoindex = ["lancedb>=0.40,<0.41", "pylance>=9,<10", "pyarrow>=19", "cocoindex> dolt = ["pymysql>=1.1", "mysql-connector-python>=9"] mysql = ["pymysql>=1.1", "cryptography>=42"] postgres = ["psycopg[binary]>=3.1"] +terminusdb = ["terminusdb-client>=10.2"] voyage = ["voyageai>=0.2"] huggingface = ["sentence-transformers>=2.5"] dev = [