feat: context store — fact and document CRUD

app/context/store.py

@@ -0,0 +1,133 @@
+"""Context fact and document CRUD — MIT licensed."""
+from __future__ import annotations
+
+import sqlite3
+import uuid
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+
+
+@dataclass(frozen=True)
+class ContextFact:
+    id: str
+    category: str
+    key: str
+    value: str
+    source: str | None
+    created_at: str
+
+
+@dataclass(frozen=True)
+class ContextDocument:
+    id: str
+    filename: str
+    doc_type: str
+    full_text: str
+    file_size: int | None
+    uploaded_at: str
+
+
+def _connect(db_path: Path) -> sqlite3.Connection:
+    conn = sqlite3.connect(str(db_path))
+    conn.execute("PRAGMA journal_mode=WAL")
+    conn.execute("PRAGMA foreign_keys=ON")
+    conn.row_factory = sqlite3.Row
+    return conn
+
+
+def add_fact(db_path: Path, category: str, key: str, value: str, source: str | None = None) -> ContextFact:
+    fact = ContextFact(
+        id=str(uuid.uuid4()),
+        category=category,
+        key=key,
+        value=value,
+        source=source,
+        created_at=datetime.now(timezone.utc).isoformat(),
+    )
+    conn = _connect(db_path)
+    conn.execute(
+        "INSERT INTO context_facts(id, category, key, value, source, created_at) VALUES (?,?,?,?,?,?)",
+        (fact.id, fact.category, fact.key, fact.value, fact.source, fact.created_at),
+    )
+    conn.commit()
+    conn.close()
+    return fact
+
+
+def list_facts(db_path: Path, category: str | None = None) -> list[ContextFact]:
+    conn = _connect(db_path)
+    if category:
+        rows = conn.execute(
+            "SELECT * FROM context_facts WHERE category=? ORDER BY created_at", (category,)
+        ).fetchall()
+    else:
+        rows = conn.execute(
+            "SELECT * FROM context_facts ORDER BY category, created_at"
+        ).fetchall()
+    conn.close()
+    return [
+        ContextFact(
+            id=r["id"], category=r["category"], key=r["key"],
+            value=r["value"], source=r["source"], created_at=r["created_at"],
+        )
+        for r in rows
+    ]
+
+
+def delete_fact(db_path: Path, fact_id: str) -> bool:
+    conn = _connect(db_path)
+    cursor = conn.execute("DELETE FROM context_facts WHERE id=?", (fact_id,))
+    conn.commit()
+    conn.close()
+    return cursor.rowcount > 0
+
+
+def add_document(
+    db_path: Path,
+    filename: str,
+    doc_type: str,
+    full_text: str,
+    file_size: int | None = None,
+) -> ContextDocument:
+    doc = ContextDocument(
+        id=str(uuid.uuid4()),
+        filename=filename,
+        doc_type=doc_type,
+        full_text=full_text,
+        file_size=file_size,
+        uploaded_at=datetime.now(timezone.utc).isoformat(),
+    )
+    conn = _connect(db_path)
+    conn.execute(
+        "INSERT INTO context_documents(id, filename, doc_type, full_text, file_size, uploaded_at)"
+        " VALUES (?,?,?,?,?,?)",
+        (doc.id, doc.filename, doc.doc_type, doc.full_text, doc.file_size, doc.uploaded_at),
+    )
+    conn.commit()
+    conn.close()
+    return doc
+
+
+def list_documents(db_path: Path) -> list[ContextDocument]:
+    conn = _connect(db_path)
+    rows = conn.execute(
+        "SELECT id, filename, doc_type, full_text, file_size, uploaded_at"
+        " FROM context_documents ORDER BY uploaded_at DESC"
+    ).fetchall()
+    conn.close()
+    return [
+        ContextDocument(
+            id=r["id"], filename=r["filename"], doc_type=r["doc_type"],
+            full_text=r["full_text"], file_size=r["file_size"], uploaded_at=r["uploaded_at"],
+        )
+        for r in rows
+    ]
+
+
+def delete_document(db_path: Path, doc_id: str) -> bool:
+    conn = _connect(db_path)
+    cursor = conn.execute("DELETE FROM context_documents WHERE id=?", (doc_id,))
+    conn.commit()
+    conn.close()
+    return cursor.rowcount > 0

tests/context/test_store.py

@@ -0,0 +1,105 @@
+"""Tests for app/context/store.py — fact and document CRUD."""
+import sqlite3
+import pytest
+from pathlib import Path
+from app.context.store import (
+    add_fact, list_facts, delete_fact,
+    add_document, list_documents, delete_document,
+    ContextFact, ContextDocument,
+)
+
+
+@pytest.fixture
+def db(tmp_path):
+    db_path = tmp_path / "t.db"
+    conn = sqlite3.connect(str(db_path))
+    conn.executescript("""
+        CREATE TABLE context_facts (
+            id TEXT PRIMARY KEY, category TEXT NOT NULL, key TEXT NOT NULL,
+            value TEXT NOT NULL, source TEXT, created_at TEXT NOT NULL
+        );
+        CREATE TABLE context_documents (
+            id TEXT PRIMARY KEY, filename TEXT NOT NULL, doc_type TEXT NOT NULL,
+            full_text TEXT NOT NULL, file_size INTEGER, uploaded_at TEXT NOT NULL
+        );
+        CREATE TABLE context_chunks (
+            id TEXT PRIMARY KEY, document_id TEXT NOT NULL
+                REFERENCES context_documents(id) ON DELETE CASCADE,
+            chunk_index INTEGER NOT NULL, text TEXT NOT NULL, embedding BLOB
+        );
+    """)
+    conn.commit()
+    conn.close()
+    return db_path
+
+
+def test_add_fact_returns_fact(db):
+    fact = add_fact(db, "host", "hostname", "heimdall.local", source="wizard")
+    assert isinstance(fact, ContextFact)
+    assert fact.id
+    assert fact.category == "host"
+    assert fact.key == "hostname"
+    assert fact.value == "heimdall.local"
+    assert fact.source == "wizard"
+    assert fact.created_at
+
+
+def test_list_facts_empty(db):
+    assert list_facts(db) == []
+
+
+def test_list_facts_all(db):
+    add_fact(db, "host", "hostname", "heimdall.local")
+    add_fact(db, "service", "plex", "port:32400")
+    facts = list_facts(db)
+    assert len(facts) == 2
+
+
+def test_list_facts_by_category(db):
+    add_fact(db, "host", "hostname", "heimdall.local")
+    add_fact(db, "service", "plex", "port:32400")
+    add_fact(db, "service", "sonarr", "port:8989")
+    assert len(list_facts(db, category="host")) == 1
+    assert len(list_facts(db, category="service")) == 2
+
+
+def test_delete_fact_returns_true(db):
+    fact = add_fact(db, "note", "k", "v")
+    assert delete_fact(db, fact.id) is True
+    assert list_facts(db) == []
+
+
+def test_delete_fact_missing_returns_false(db):
+    assert delete_fact(db, "nonexistent") is False
+
+
+def test_add_document_returns_document(db):
+    doc = add_document(db, "runbook.md", "markdown", "# Plex\nRestart with systemctl", file_size=100)
+    assert isinstance(doc, ContextDocument)
+    assert doc.id
+    assert doc.filename == "runbook.md"
+    assert doc.doc_type == "markdown"
+
+
+def test_list_documents_empty(db):
+    assert list_documents(db) == []
+
+
+def test_delete_document_cascades_chunks(db):
+    doc = add_document(db, "test.md", "markdown", "content")
+    conn = sqlite3.connect(str(db))
+    conn.execute(
+        "INSERT INTO context_chunks(id, document_id, chunk_index, text) VALUES (?,?,0,?)",
+        ("c1", doc.id, "chunk text"),
+    )
+    conn.commit()
+    conn.close()
+    assert delete_document(db, doc.id) is True
+    conn = sqlite3.connect(str(db))
+    chunks = conn.execute("SELECT * FROM context_chunks WHERE document_id=?", (doc.id,)).fetchall()
+    conn.close()
+    assert chunks == []
+
+
+def test_delete_document_missing_returns_false(db):
+    assert delete_document(db, "nonexistent") is False