"""Graphify output: nodes and edges, each joined back to the note it came from. `docs/phase-4-corpus.md` §4: *"Graph is not chunked. Nodes and edges go into their own collection, joined to note chunks by `path`, so a graph answer can name the note it came from."* The graph is not rebuilt here. It already exists and is rebuilt on the build host by its own skill; this module consumes `graph.json`. Two rules from §3 drive the whole file: * **A node with no resolvable source note is refused, not defaulted.** Graphify emits nodes for files outside the vault (`settings.json`, `.pi/…`). Those are not traceable to a note, so they are rejected and counted. * **Media never becomes text.** A media node becomes an image reference: path, origin URL and note, and nothing else. """ from __future__ import annotations import hashlib import json import re import uuid from .. import areas, exclusions, globs from ..sources.gitea import file_url class GraphError(RuntimeError): """graph.json is not a graph this pipeline can admit.""" class GraphResult: def __init__(self, platform: str = "graphify"): self.platform = platform self.nodes: list = [] self.edges: list = [] self.images: list = [] self.rejections: list = [] self.warnings: list = [] self.digest = "" self.source_path = "" self.built_at_commit = "" def counts(self) -> dict: return { "nodes": len(self.nodes), "edges": len(self.edges), "images": len(self.images), "rejected": len(self.rejections), } def parse( raw: bytes, *, platform: str, base: str, repo: str, ref: str, last_modified: str, area: str, vault_paths: set, max_nodes: int = 20000, max_edges: int = 40000, source_path: str = "", ) -> GraphResult: """Turn graph.json bytes into rows that carry their note. Never guesses.""" result = GraphResult(platform) result.digest = hashlib.sha256(raw).hexdigest() result.source_path = source_path try: data = json.loads(raw.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise GraphError(f"graph.json is not valid UTF-8 JSON: {exc}") from exc if not isinstance(data, dict): raise GraphError("graph.json must be a JSON object") result.built_at_commit = str(data.get("built_at_commit") or "") raw_nodes = data.get("nodes") or [] raw_links = data.get("links") or data.get("edges") or [] if not isinstance(raw_nodes, list) or not isinstance(raw_links, list): raise GraphError("graph.json must contain `nodes` and `links` arrays") labels = {} notes = {} for item in raw_nodes: if not isinstance(item, dict): continue node_id = str(item.get("id") or "").strip() if not node_id: continue labels[node_id] = str(item.get("label") or node_id) notes[node_id] = _note_for(item, vault_paths) for item in raw_nodes: if len(result.nodes) >= max_nodes: result.warnings.append(f"graph node limit {max_nodes} reached; remaining nodes skipped") break if not isinstance(item, dict): continue node_id = str(item.get("id") or "").strip() if not node_id: result.rejections.append( _rejection("graph.json", platform, "graph node has no id") ) continue source_file = globs.normalise(item.get("source_file") or "") note = notes.get(node_id) if not note: result.rejections.append( _rejection( source_file or f"node:{node_id}", platform, "no resolvable source note", "graph node does not point at a file in the vault", ) ) continue if globs.normalise(source_file) != note: # A node can be derived from one note while naming another file; the # note is the traceable origin either way. pass if exclusions.classify(note, None)[0] == "image" or exclusions.extension(note) in exclusions.MEDIA_EXTENSIONS: result.images.append( { "path": note, "label": str(item.get("label") or node_id), "origin_url": file_url(base, repo, ref, note), "platform": platform, "note": note, } ) continue node_area = areas.area_for(note, (area,), fallback=area) label = str(item.get("label") or node_id) content_hash = _hash( {"kind": "node", "id": node_id, "label": label, "note": note, "community": item.get("community"), "type": item.get("node_type")} ) text = ( f"{label}. Knowledge graph node" + (f" in community {item.get('community')}" if item.get("community") is not None else "") + f", from the note {note}." ) result.nodes.append( { "id": node_id, "point_id": _point_id("node", node_id, content_hash), "graph_id": node_id, "label": label, "norm_label": str(item.get("norm_label") or _norm(label)), "node_type": str(item.get("node_type") or item.get("file_type") or "entity"), "community": item.get("community"), "source_note": note, "origin_url": file_url(base, repo, ref, note), "area": node_area, "tier": 2, "platform": platform, "last_modified": last_modified, "content_hash": content_hash, "text": text, } ) admitted_nodes = {row["graph_id"]: row for row in result.nodes} for item in raw_links: if len(result.edges) >= max_edges: result.warnings.append(f"graph edge limit {max_edges} reached; remaining edges skipped") break if not isinstance(item, dict): continue source = str(item.get("source") or "").strip() target = str(item.get("target") or "").strip() if not source or not target: result.rejections.append(_rejection("graph.json", platform, "graph edge has no endpoints")) continue note = notes.get(source) or _note_for(item, vault_paths) if not note: result.rejections.append( _rejection( f"edge:{source}->{target}", platform, "no resolvable source note", "graph edge does not point at a file in the vault", ) ) continue source_label = labels.get(source, source) target_label = labels.get(target, target) relation = str(item.get("relation") or item.get("rel") or "related_to") area_name = areas.area_for(note, (area,), fallback=area) content_hash = _hash( {"kind": "edge", "source": source, "target": target, "relation": relation, "note": note} ) text = f"{source_label} {relation} {target_label}, from the note {note}." result.edges.append( { "id": _edge_id(source, relation, target, content_hash), "point_id": _point_id("edge", f"{source}>{relation}>{target}", content_hash), "source_id": source, "target_id": target, "relation": relation, "confidence": str(item.get("confidence") or "INFERRED"), "confidence_score": item.get("confidence_score"), "source_note": note, "origin_url": file_url(base, repo, ref, note), "area": area_name, "tier": 2, "platform": platform, "last_modified": last_modified, "content_hash": content_hash, "text": text, } ) # Edges whose endpoints were not admitted are still traceable and are kept; # an answer can name the relation and its note even when the node itself had # no origin. `admitted_node_ids` is exposed for verify, which reports the gap. result.admitted_node_ids = set(admitted_nodes) result.orphan_edges = [ edge for edge in result.edges if edge["source_id"] not in admitted_nodes or edge["target_id"] not in admitted_nodes ] return result def point_payload(row: dict, kind: str) -> dict: """One Qdrant point payload for a node or edge. Same provenance as the row.""" payload = { "platform": row["platform"], "tier": row["tier"], "source_type": "graph_node" if kind == "node" else "graph_edge", "area": row["area"], "origin_url": row["origin_url"], "path": row["source_note"], "source_note": row["source_note"], "title": row.get("label") or row.get("relation") or row["source_note"], "confidence": row.get("confidence") or "INFERRED", "last_modified": row["last_modified"], "content_hash": row["content_hash"], "text": row["text"], "kind": kind, "graph_id": row.get("graph_id") or row.get("id"), } if kind == "node": payload["community"] = row.get("community") payload["node_type"] = row.get("node_type") else: payload["relation"] = row.get("relation") payload["source_id"] = row.get("source_id") payload["target_id"] = row.get("target_id") payload["confidence_score"] = row.get("confidence_score") return payload def query(cfg, store, text: str, area: str = None, limit: int = 10) -> dict: """Graph query: keyword hits from SQLite, vector hits from Qdrant when up. Returning both is deliberate: the keyword side always works, and the vector side proves the graph collection is live. Each row carries `source_note`. """ from ..qdrant import Qdrant, QdrantError report = {"query": text, "area": area, "keyword": [], "vector": [], "ok": True} report["keyword"] = store.graph_search(text, area=area, limit=limit) if not report["keyword"]: report["ok"] = False report["reason"] = "no graph rows matched the query" qdrant = Qdrant(cfg.qdrant_url, timeout=cfg.timeout) if not qdrant.health() or not qdrant.collection_exists(cfg.graph_collection): report["vector"] = [] report["vector_note"] = f"graph collection {cfg.graph_collection} is not available" return report from .. import embeddings provider = embeddings.build_provider(cfg) try: vector = provider.embed([text])[0] hits = qdrant.search(cfg.graph_collection, vector, limit=limit, area=area) except (QdrantError, Exception) as exc: # noqa: BLE001 - a graph query must not crash a shell report["vector_note"] = f"graph vector search unavailable: {exc}" return report report["vector"] = [ { "score": round(hit.get("score", 0.0), 4), "kind": (hit.get("payload") or {}).get("kind"), "label": (hit.get("payload") or {}).get("title"), "source_note": (hit.get("payload") or {}).get("source_note"), "origin_url": (hit.get("payload") or {}).get("origin_url"), } for hit in hits ] return report # -- helpers ------------------------------------------------------------- def _note_for(item: dict, vault_paths: set) -> str: source = globs.normalise(item.get("source_file") or "") if not source: return "" if source in vault_paths: return source # Graphify sometimes prefixes `./` or uses an absolute scan root. A suffix # match is only accepted when it is unique, so a note is never guessed. candidates = [path for path in vault_paths if path.endswith("/" + source)] if len(candidates) == 1: return candidates[0] return "" def _rejection(path: str, platform: str, reason: str, detail: str = ""): rejection = exclusions.Rejection(path, platform, reason, detail) return rejection.to_dict() def _hash(value) -> str: body = json.dumps(value, sort_keys=True, separators=(",", ":")).encode("utf-8") return hashlib.sha256(body).hexdigest() def _point_id(kind: str, key: str, content_hash: str) -> str: digest = hashlib.sha256(f"graph\x00{kind}\x00{key}\x00{content_hash}".encode("utf-8")).hexdigest() return str(uuid.UUID(hex=digest[:32])) def _edge_id(source: str, relation: str, target: str, content_hash: str) -> str: return hashlib.sha256( f"edge\x00{source}\x00{relation}\x00{target}\x00{content_hash}".encode("utf-8") ).hexdigest()[:32] def _norm(label: str) -> str: return re.sub(r"[^0-9a-z]+", "_", (label or "").lower()).strip("_")