|
@@ -0,0 +1,52 @@
|
|
|
|
|
+from __future__ import annotations
|
|
|
|
|
+
|
|
|
|
|
+import json
|
|
|
|
|
+import sqlite3
|
|
|
|
|
+from pathlib import Path
|
|
|
|
|
+from typing import Any
|
|
|
|
|
+
|
|
|
|
|
+ROOT = Path(__file__).resolve().parents[2]
|
|
|
|
|
+DATA_DIR = ROOT / "data"
|
|
|
|
|
+DB_PATH = DATA_DIR / "hermes_mcp.sqlite3"
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def _connect() -> sqlite3.Connection:
|
|
|
|
|
+ DATA_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
|
+ conn = sqlite3.connect(DB_PATH)
|
|
|
|
|
+ conn.row_factory = sqlite3.Row
|
|
|
|
|
+ return conn
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def init_db() -> None:
|
|
|
|
|
+ with _connect() as conn:
|
|
|
|
|
+ conn.execute(
|
|
|
|
|
+ """
|
|
|
|
|
+ create table if not exists state (
|
|
|
|
|
+ key text primary key,
|
|
|
|
|
+ value text not null,
|
|
|
|
|
+ updated_at text not null default current_timestamp
|
|
|
|
|
+ )
|
|
|
|
|
+ """
|
|
|
|
|
+ )
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def get_state() -> dict[str, Any]:
|
|
|
|
|
+ init_db()
|
|
|
|
|
+ with _connect() as conn:
|
|
|
|
|
+ row = conn.execute("select value from state where key = ?", ("snapshot",)).fetchone()
|
|
|
|
|
+ if not row:
|
|
|
|
|
+ return {
|
|
|
|
|
+ "status": "stub",
|
|
|
|
|
+ "thinking": "Hermes is scaffolded and waiting for integrations.",
|
|
|
|
|
+ "layers": ["overview", "signals", "features", "narrative", "decision", "explanation"],
|
|
|
|
|
+ }
|
|
|
|
|
+ return json.loads(row["value"])
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def put_state(payload: dict[str, Any]) -> None:
|
|
|
|
|
+ init_db()
|
|
|
|
|
+ with _connect() as conn:
|
|
|
|
|
+ conn.execute(
|
|
|
|
|
+ "insert into state(key, value, updated_at) values(?, ?, current_timestamp) on conflict(key) do update set value=excluded.value, updated_at=current_timestamp",
|
|
|
|
|
+ ("snapshot", json.dumps(payload)),
|
|
|
|
|
+ )
|