| """Typed knowledge graph with governed, versioned facts (decisions 18, 38, 47; section 11). |
| |
| Entities: project, mission, hypothesis, candidate, composition, atom/site, structure, property, method, |
| solver_version, run, dataset, publication, evidence_claim, agent, software_version. |
| |
| Facts are append-only versioned events with status proposed | provisional | verified | contradicted | deprecated. |
| Automatic ingestion never promotes to 'verified': see knowledge/governance.py. |
| """ |
| from __future__ import annotations |
|
|
| from typing import Any, Optional |
|
|
| from core.errors import ErrorCode, MatterOSError |
| from core.provenance import utcnow |
| from core.schemas import KnowledgeStatus, new_id |
| from knowledge import governance |
| from storage.db import Database |
|
|
| ENTITY_TYPES = ["project", "mission", "hypothesis", "candidate", "composition", "atom_site", "structure", "property", |
| "method", "solver_version", "run", "dataset", "publication", "evidence_claim", "agent", "software_version"] |
|
|
|
|
| class KnowledgeGraph: |
| def __init__(self, db: Database): |
| self.db = db |
|
|
| |
| def upsert_entity(self, entity_id: str, etype: str, label: str = "", props: dict | None = None) -> str: |
| if etype not in ENTITY_TYPES: |
| raise MatterOSError(ErrorCode.INVALID_INPUT, f"unknown entity type {etype}") |
| if not self.db.one("SELECT 1 FROM kg_entities WHERE entity_id=?", (entity_id,)): |
| self.db.execute("INSERT INTO kg_entities VALUES (?,?,?,?,?)", (entity_id, etype, label, self.db.dumps(props or {}), utcnow())) |
| return entity_id |
|
|
| def link(self, src: str, dst: str, etype: str, props: dict | None = None, status: str = "provisional") -> str: |
| eid = new_id("edge") |
| self.db.execute("INSERT INTO kg_edges VALUES (?,?,?,?,?,?,?)", (eid, src, dst, etype, self.db.dumps(props or {}), status, utcnow())) |
| return eid |
|
|
| |
| def propose_event(self, subject: str, predicate: str, obj: Any, *, actor: str, evidence_refs: list[str] | None = None, |
| provenance: dict | None = None, note: str = "") -> dict: |
| status = governance.initial_status(evidence_refs or [], actor) |
| fact_id = new_id("fact") |
| return self._append(fact_id, 1, subject, predicate, obj, status, evidence_refs or [], provenance or {}, actor, note) |
|
|
| def _append(self, fact_id, version, subject, predicate, obj, status, evidence_refs, provenance, actor, note) -> dict: |
| ev = {"event_id": new_id("kge"), "fact_id": fact_id, "version": version, "subject": subject, "predicate": predicate, |
| "object": obj, "status": status, "evidence_refs": evidence_refs, "provenance": provenance, "actor": actor, |
| "note": note, "created_at": utcnow()} |
| self.db.execute("INSERT INTO kg_events VALUES (?,?,?,?,?,?,?,?,?,?,?,?)", |
| (ev["event_id"], fact_id, version, subject, predicate, self.db.dumps(obj), status, |
| self.db.dumps(evidence_refs), self.db.dumps(provenance), actor, note, ev["created_at"])) |
| return ev |
|
|
| def current(self, fact_id: str) -> dict: |
| r = self.db.one("SELECT * FROM kg_events WHERE fact_id=? ORDER BY version DESC LIMIT 1", (fact_id,)) |
| if not r: |
| raise MatterOSError(ErrorCode.INVALID_INPUT, f"fact not found {fact_id}") |
| return self._row(r) |
|
|
| def history(self, fact_id: str) -> list[dict]: |
| return [self._row(r) for r in self.db.query("SELECT * FROM kg_events WHERE fact_id=? ORDER BY version", (fact_id,))] |
|
|
| def _row(self, r): |
| r = dict(r) |
| for k in ("object", "evidence_refs", "provenance"): |
| r[k] = self.db.loads(r[k]) |
| return r |
|
|
| def transition(self, fact_id: str, new_status: str, *, actor: str, evidence_refs: list[str] | None = None, |
| evidence_check=None, note: str = "") -> dict: |
| cur = self.current(fact_id) |
| refs = list(dict.fromkeys((cur["evidence_refs"] or []) + (evidence_refs or []))) |
| governance.check_transition(cur["status"], new_status, refs, actor, evidence_check) |
| return self._append(fact_id, cur["version"] + 1, cur["subject"], cur["predicate"], cur["object"], new_status, refs, |
| cur["provenance"], actor, note) |
|
|
| def verify_event(self, fact_id: str, *, actor: str, evidence_refs: list[str], evidence_check=None) -> dict: |
| return self.transition(fact_id, KnowledgeStatus.VERIFIED.value, actor=actor, evidence_refs=evidence_refs, |
| evidence_check=evidence_check, note="verification requested") |
|
|
| def search_graph(self, text: str = "", status: Optional[str] = None, etype: Optional[str] = None, limit: int = 200) -> dict: |
| ents = self.db.query("SELECT * FROM kg_entities WHERE (label LIKE ? OR entity_id LIKE ?)" + (" AND type=?" if etype else "") + |
| " LIMIT ?", (f"%{text}%", f"%{text}%", *( [etype] if etype else []), limit)) |
| |
| facts = self.db.query("SELECT e.* FROM kg_events e JOIN (SELECT fact_id, MAX(version) v FROM kg_events GROUP BY fact_id) m " |
| "ON e.fact_id=m.fact_id AND e.version=m.v WHERE (e.subject LIKE ? OR e.predicate LIKE ? OR e.object LIKE ?)" |
| + (" AND e.status=?" if status else "") + " ORDER BY e.created_at DESC LIMIT ?", |
| (f"%{text}%", f"%{text}%", f"%{text}%", *([status] if status else []), limit)) |
| return {"entities": ents, "facts": [self._row(f) for f in facts]} |
|
|
| def status_counts(self) -> dict[str, int]: |
| rows = self.db.query("SELECT e.status, COUNT(*) n FROM kg_events e JOIN (SELECT fact_id, MAX(version) v FROM kg_events GROUP BY fact_id) m " |
| "ON e.fact_id=m.fact_id AND e.version=m.v GROUP BY e.status") |
| return {r["status"]: r["n"] for r in rows} |
|
|
| def edges(self, node: Optional[str] = None) -> list[dict]: |
| if node: |
| return self.db.query("SELECT * FROM kg_edges WHERE src=? OR dst=?", (node, node)) |
| return self.db.query("SELECT * FROM kg_edges") |
|
|
