MPEdb becomes the store, and the graph becomes tables SQL can ask

.gitignore

@@ -25,3 +25,6 @@ build/
 spell.mpedb*
 *.mpedb
 *.mpedb.wlock
+
+# The graph tables the app records finished runs into.
+knowledge.graph.db*

dev/tests/test_graphdb.py

@@ -0,0 +1,151 @@
+"""The graph as tables: recorded once per run, queryable in SQL, in memory too.
+
+The RDF is the graph as a document; this is the graph as a database. Both draw
+from the same job events, so the test that matters most here is the one that
+feeds a settle-shaped event stream in and asks the questions a consumer would:
+which findings did the run make, how was agreement reached, what did the vote
+say. If SQL can answer those, the tables carry the structure.
+"""
+
+from __future__ import annotations
+
+import json
+import time
+
+import pytest
+from conftest import normal_script, step
+
+from mpe_lkg.backends import DeterministicEmbedding, ScriptedChat
+from mpe_lkg.graphdb import GraphDB
+from mpe_lkg.jobs import Job
+
+
+def settled_job() -> Job:
+    """A job holding the events a settle run produces, without running a model."""
+    job = Job("How many minutes are there in a fortnight?")
+    job.mode = "settle"
+    job.session = "clocks"
+    for event in [
+        {"type": "round", "round": 1, "of": 3},
+        {"type": "finding", "question": "How many days are in a fortnight?",
+         "answer": "A fortnight has 14 days."},
+        {"type": "finding", "question": "How many minutes are in a day?",
+         "answer": "There are 1440 minutes in a day."},
+        {"type": "finding", "level": 1, "question": "a sub-run's own finding",
+         "answer": "must not be recorded at the top level"},
+        {"type": "step", "step": 1, "title": "Convert", "content": "Ask for it."},
+        {"type": "calc", "step": 1, "expression": "14*1440", "value": "20,160"},
+        {"type": "convert", "step": 1, "request": "1 fortnight to minutes",
+         "result": "1 fortnight = 20,160 minute"},
+        {"type": "vote", "agree": 2, "disagree": 1, "about": "the total"},
+        {"type": "agreed", "by": "vote", "round": 2, "answer": "20160"},
+        {"type": "final", "content": "There are 20160 minutes in a fortnight.",
+         "graph": {"nodes": [], "edges": [
+             {"from": "Step1", "to": "Step2", "value": 0.83}]}},
+    ]:
+        job.append(event)
+    job.answer = "There are 20160 minutes in a fortnight."
+    job.done.set()
+    return job
+
+
+class TestRecording:
+    def test_a_settle_run_is_answerable_in_sql(self):
+        graph = GraphDB(":memory:")
+        graph.record(settled_job())
+
+        # Qualified columns: both tables carry `question` and `answer`, and
+        # mpedb refuses the ambiguity where sqlite silently picked one. The
+        # refusal is the better behaviour.
+        findings = graph.query(
+            "SELECT findings.question, findings.answer FROM findings"
+            " JOIN runs ON findings.run = runs.id"
+            " WHERE runs.session = 'clocks' ORDER BY findings.question")
+        assert len(findings) == 2
+        assert findings[0][1] == "A fortnight has 14 days."
+
+        how = graph.query(
+            "SELECT agreement.reached_by, votes.agree, votes.disagree FROM agreement"
+            " JOIN votes ON votes.run = agreement.run")
+        assert how == [("vote", 2, 1)]
+
+        facts = graph.query("SELECT kind, statement FROM facts ORDER BY kind")
+        assert ("calculation", "14*1440 = 20,160") in facts
+        assert ("conversion", "1 fortnight = 20,160 minute") in facts
+
+    def test_sub_run_findings_stay_out_of_the_top_level(self):
+        graph = GraphDB(":memory:")
+        graph.record(settled_job())
+        texts = [row[0] for row in graph.query("SELECT question FROM findings")]
+        assert "a sub-run's own finding" not in texts
+
+    def test_recording_twice_records_once(self):
+        graph = GraphDB(":memory:")
+        job = settled_job()
+        graph.record(job)
+        graph.record(job)
+        assert graph.query("SELECT COUNT(*) FROM runs")[0][0] == 1
+        assert graph.query("SELECT COUNT(*) FROM findings")[0][0] == 2
+
+    def test_session_stats_count_what_a_session_did(self):
+        graph = GraphDB(":memory:")
+        graph.record(settled_job())
+        stats = graph.session_stats("clocks")
+        assert stats == {"runs": 1, "agreed": 1, "exact_facts": 2}
+        assert graph.session_stats("elsewhere") == {
+            "runs": 0, "agreed": 0, "exact_facts": 0}
+
+    def test_edges_carry_their_similarity(self):
+        graph = GraphDB(":memory:")
+        graph.record(settled_job())
+        assert graph.query("SELECT a, b, sim FROM edges") == [("Step1", "Step2", 0.83)]
+
+
+class TestThroughTheApi:
+    @pytest.fixture
+    def client(self, tmp_path):
+        import mpe_lkg.app as app_module
+
+        app_module.app.config["DB_PATH"] = str(tmp_path / "app.db")
+        app_module.app.config["GRAPH_DB_PATH"] = str(tmp_path / "graph.db")
+        app_module.app.config["BACKENDS_FACTORY"] = lambda: (
+            ScriptedChat(normal_script(4), repeat_last=True),
+            DeterministicEmbedding(24))
+        app_module.JOBS = app_module.Registry()
+        with app_module.app.test_client() as c:
+            yield c, str(tmp_path / "graph.db")
+
+    def test_a_finished_job_lands_in_the_tables(self, client):
+        c, graph_path = client
+        job_id = c.post("/jobs", json={"query": "What is the capital of France?",
+                                       "session": "geo"}).get_json()["id"]
+        deadline = time.time() + 20
+        while time.time() < deadline:
+            if c.get(f"/jobs/{job_id}").get_json()["state"] == "done":
+                break
+            time.sleep(0.05)
+
+        graph = GraphDB(graph_path)
+        runs = graph.query("SELECT id, session, question FROM runs")
+        assert (job_id, "geo", "What is the capital of France?") in runs
+        assert graph.query("SELECT COUNT(*) FROM steps WHERE run = ?",
+                           (job_id,))[0][0] > 0
+
+    def test_the_sessions_listing_carries_the_graphs_numbers(self, client):
+        c, _ = client
+        job_id = c.post("/jobs", json={"query": "q", "session": "geo"}).get_json()["id"]
+        deadline = time.time() + 20
+        while time.time() < deadline:
+            if c.get(f"/jobs/{job_id}").get_json()["state"] == "done":
+                break
+            time.sleep(0.05)
+        listed = {s["name"]: s for s in c.get("/sessions").get_json()["sessions"]}
+        assert listed["geo"]["runs"] == 1
+        assert "exact_facts" in listed["geo"]
+
+
+def test_the_scripted_step_helper_still_matches_the_schema():
+    # The settle-shaped stream above is hand-built; this pins that the helper
+    # the rest of the suite uses produces steps record() can read.
+    parsed = json.loads(step("T", "C"))
+    assert {"title", "content", "next_action"} <= set(parsed)

dev/tests/test_jobs.py

@@ -27,6 +27,7 @@ def client(tmp_path):
     import mpe_lkg.app as app_module
 
     app_module.app.config["DB_PATH"] = str(tmp_path / "jobs.db")
+    app_module.app.config["GRAPH_DB_PATH"] = str(tmp_path / "graph.db")
     app_module.app.config["BACKENDS_FACTORY"] = lambda: (
         ScriptedChat(normal_script(5)), DeterministicEmbedding(32))
     app_module.JOBS = app_module.Registry()

dev/tests/test_render.py

@@ -34,6 +34,7 @@ def __init__(self, script, tmp_path, *, repeat_last=False, dim=48, delay=0.0):
         chat = ScriptedChat(script, repeat_last=repeat_last, delay=delay)
         app_module.app.config["BACKENDS_FACTORY"] = lambda: (chat, DeterministicEmbedding(dim))
         app_module.app.config["DB_PATH"] = str(tmp_path / "render.db")
+        app_module.app.config["GRAPH_DB_PATH"] = str(tmp_path / "graph.db")
         self.port = free_port()
         self._server = make_server("127.0.0.1", self.port, app_module.app, threaded=True)
         self._thread = threading.Thread(target=self._server.serve_forever, daemon=True)

dev/tests/test_sessions.py

@@ -107,6 +107,9 @@ def client(tmp_path):
     import mpe_lkg.app as app_module
 
     app_module.app.config["DB_PATH"] = str(tmp_path / "app.db")
+    # Without this the sessions endpoint records into the repo root -- the
+    # stray-file mistake this project has already committed once.
+    app_module.app.config["GRAPH_DB_PATH"] = str(tmp_path / "graph.db")
     app_module.app.config["BACKENDS_FACTORY"] = lambda: (
         ScriptedChat(normal_script(4), repeat_last=True), DeterministicEmbedding(24))
     app_module.JOBS = app_module.Registry()

pyproject.toml

@@ -43,6 +43,11 @@ dependencies = [
     # would leave the conversions silently missing instead of failing loudly,
     # which is the worst way for a dependency floor to be wrong.
     "mpeqs>=0.3.0",
+    # The store engine. A drop-in for sqlite3 that also opens existing
+    # embeddings.db files in place -- probed before adoption, tested after --
+    # plus the in-memory mode the run-graph tables use. Wheels cover the same
+    # matrix this package supports: 3.10+, Intel and ARM Mac, Linux, Windows.
+    "mpedb>=0.3.0",
 ]
 
 [project.optional-dependencies]

src/mpe_lkg/app.py

@@ -17,6 +17,7 @@
 from flask import Flask, Response, jsonify, render_template, request
 
 from . import backends, rdf
+from .graphdb import GraphDB
 from .jobs import Job, Registry
 from .reasoning import explore, reason, settle
 from .store import EmbeddingStore
@@ -163,8 +164,24 @@ def favicon():
 JOBS = Registry()
 
 
+GRAPH_DB = "knowledge.graph.db"
+
+
+def _record(job) -> None:
+    """One finished job into the graph tables. Failures are logged, not fatal:
+    the answer has already been delivered, and recording is bookkeeping."""
+    try:
+        graph = GraphDB(app.config.get("GRAPH_DB_PATH", GRAPH_DB))
+        try:
+            graph.record(job)
+        finally:
+            graph.close()
+    except Exception:  # noqa: BLE001
+        app.logger.exception("could not record run %s", job.id)
+
+
 def _job_events(user_query: str, store_path: str, mode: str = "reason",
-                session: str = ""):
+                session: str = "", job=None):
     """The reasoning events for a headless run, with its own backends and store."""
     chat, embedder = app.config.get("BACKENDS_FACTORY", make_backends)()
     store = EmbeddingStore(store_path)
@@ -184,6 +201,8 @@ def _job_events(user_query: str, store_path: str, mode: str = "reason",
                               decompose=decompose)
     finally:
         store.close()
+        if job is not None:
+            _record(job)
 
 
 @app.route("/jobs", methods=["GET", "POST"])
@@ -209,7 +228,8 @@ def jobs():
     job.mode = mode
     job.session = session
     JOBS.add(job)
-    job.start(_job_events(user_query, app.config.get("DB_PATH", DB_PATH), mode, session))
+    job.start(_job_events(user_query, app.config.get("DB_PATH", DB_PATH), mode,
+                          session, job=job))
     # 202: accepted and still running. The Location header is where to look.
     return jsonify(job.status()), 202, {"Location": f"/jobs/{job.id}"}
 
@@ -218,9 +238,19 @@ def jobs():
 def sessions():
     store = EmbeddingStore(app.config.get("DB_PATH", DB_PATH))
     try:
-        return jsonify({"sessions": store.sessions()})
+        listed = store.sessions()
     finally:
         store.close()
+    try:
+        graph = GraphDB(app.config.get("GRAPH_DB_PATH", GRAPH_DB))
+        try:
+            for entry in listed:
+                entry.update(graph.session_stats(entry["name"]))
+        finally:
+            graph.close()
+    except Exception:  # noqa: BLE001 -- stats are additive, never blocking
+        app.logger.exception("could not read graph stats")
+    return jsonify({"sessions": listed})
 
 
 @app.route("/sessions/<name>", methods=["DELETE"])