logos_kernel.py

"""
logos_kernel.py — LOGOS Temporal Knowledge Graph
=================================================
iNFAMØUS OS cognitive memory layer.

Ingests events from all NEXUS ports, builds a temporal graph in SQLite,
and detects causal/correlative patterns across modules.

Graph semantics:
  NODE  = any event published on the NEXUS bus
  EDGE  = relationship between events:
    precedes        — node A happened within 60s before node B (same source)
    triggered_by    — SUBSTRATE trigger caused by HELIOS SOC overflow
    correlated_with — THEIA response_start during specific energy budget
    caused_by       — explicit causal chain (future: learned)

Usage:
  python logos_kernel.py            # run continuously
  python logos_kernel.py --query    # drop into query REPL
"""

from __future__ import annotations

import argparse
import json
import logging
import os
import sqlite3
import sys
import time
import uuid
from pathlib import Path

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s  [LOGOS] %(message)s",
)
log = logging.getLogger(__name__)

# ── paths ──────────────────────────────────────────────────────────────────────
_HERE    = Path(__file__).parent
_DB_PATH = Path(os.environ.get("LOGOS_DB", str(_HERE / "data" / "logos.sqlite")))
_SCHEMA  = _HERE / "schema" / "init.sql"

_HELIOS_DB = Path(os.environ.get(
    "HELIOS_DB",
    r"C:\Users\Drizzy\Desktop\HELIOS\data\energy_bus.sqlite"
))

sys.path.insert(0, str(_HERE))
from nexus.logos_subscriber import NexusSubscriber  # noqa: E402


# ── DB init ────────────────────────────────────────────────────────────────────
def _init_db(path: Path) -> sqlite3.Connection:
    path.parent.mkdir(parents=True, exist_ok=True)
    conn = sqlite3.connect(str(path), check_same_thread=False)
    conn.row_factory = sqlite3.Row
    schema = _SCHEMA.read_text(encoding="utf-8")
    conn.executescript(schema)
    conn.commit()
    log.info(f"DB ready: {path}")
    return conn


# ── HELIOS snapshot helper ─────────────────────────────────────────────────────
def _helios_snapshot() -> dict | None:
    """Read latest energy state from HELIOS DB."""
    if not _HELIOS_DB.exists():
        return None
    try:
        hc = sqlite3.connect(str(_HELIOS_DB))
        row = hc.execute(
            "SELECT power, soc, irradiance_wm2 FROM power_telemetry ORDER BY id DESC LIMIT 1"
        ).fetchone()
        fc  = hc.execute(
            "SELECT forecast_value FROM ai_forecasts ORDER BY id DESC LIMIT 1"
        ).fetchone()
        hc.close()
        if not row:
            return None
        return {
            "power_w":        row[0],
            "soc":            row[1],
            "irradiance_wm2": row[2],
            "forecast":       fc[0] if fc else None,
        }
    except Exception:
        return None


# ── graph builder ──────────────────────────────────────────────────────────────
class LogosGraph:
    """Temporal knowledge graph backed by SQLite."""

    # sliding window for `precedes` edges: events within N seconds
    PRECEDES_WINDOW_S = 60.0

    # HELIOS tick throttle — only store if state changed meaningfully
    HELIOS_SOC_DELTA    = 0.005   # 0.5% SOC change
    HELIOS_BUDGET_CHANGE = True   # always store on budget change
    HELIOS_MIN_INTERVAL  = 60.0  # store at least 1 node/min regardless

    def __init__(self, conn: sqlite3.Connection):
        self._conn = conn
        # in-memory index: source → last node_id + ts
        self._last: dict[str, tuple[str, float]] = {}
        # track response_start node for response_end correlation
        self._pending_response: str | None = None
        # last energy budget from nexus.star
        self._last_budget: str | None = None
        # HELIOS throttle state
        self._last_soc: float | None = None
        self._last_helios_stored_ts: float = 0.0

    # ── node creation ──────────────────────────────────────────────────────────
    def add_node(self, frame: dict, event_type: str = "unknown") -> str:
        node_id = str(uuid.uuid4())
        ts      = frame.get("ts", time.time())
        source  = frame.get("_source", "unknown")
        payload = json.dumps(frame.get("payload", frame))

        self._conn.execute(
            "INSERT INTO nodes (id, ts, source, event_type, payload) VALUES (?,?,?,?,?)",
            (node_id, ts, source, event_type, payload),
        )
        self._conn.commit()
        return node_id

    # ── edge creation ──────────────────────────────────────────────────────────
    def add_edge(self, src: str, dst: str, relation: str, weight: float = 1.0) -> None:
        self._conn.execute(
            "INSERT INTO edges (src, dst, relation, weight, ts) VALUES (?,?,?,?,?)",
            (src, dst, relation, weight, time.time()),
        )
        self._conn.commit()

    # ── snapshot ───────────────────────────────────────────────────────────────
    def add_snapshot(self, node_id: str, snap: dict) -> None:
        self._conn.execute(
            """INSERT INTO snapshots
               (ts, node_id, soc, irradiance_wm2, power_w, forecast, energy_budget)
               VALUES (?,?,?,?,?,?,?)""",
            (
                time.time(), node_id,
                snap.get("soc"), snap.get("irradiance_wm2"),
                snap.get("power_w"), snap.get("forecast"),
                self._last_budget,
            ),
        )
        self._conn.commit()

    # ── event dispatcher ───────────────────────────────────────────────────────
    def ingest(self, frame: dict) -> None:
        source = frame.get("_source", "unknown")

        # HELIOS bridge publishes flat payload (no "event" wrapper)
        # TheiaPublisher wraps: {"ts":..., "event":..., "payload":{...}}
        if "event" in frame:
            event_type = frame["event"]
            payload    = frame.get("payload", {})
        else:
            # flat format → infer event_type from source + keys
            if source == "nexus.star":
                event_type = "helios_tick"
            elif source == "nexus.quantum":
                event_type = "substrate_result"
            elif source == "nexus.brain":
                event_type = "khaos_tick"
            else:
                event_type = "tick"
            payload = {k: v for k, v in frame.items() if k != "_source"}

        # ── HELIOS tick throttle ───────────────────────────────────────────────
        if source == "nexus.star" and event_type == "helios_tick":
            now       = frame.get("ts", time.time())
            soc       = payload.get("soc") or frame.get("soc")
            budget    = payload.get("energy_budget") or frame.get("energy_budget")
            dt        = now - self._last_helios_stored_ts
            soc_delta = abs((soc or 0.0) - (self._last_soc or 0.0)) if soc is not None else 0.0
            budget_changed = (budget != self._last_budget) and budget is not None

            store = (
                dt >= self.HELIOS_MIN_INTERVAL       # heartbeat: 1/min
                or soc_delta >= self.HELIOS_SOC_DELTA  # meaningful SOC shift
                or budget_changed                        # regime change
            )
            if not store:
                # still update last-seen reference without writing to DB
                self._last[source] = (
                    self._last.get(source, (None, now))[0] or "",
                    now,
                )
                return

            # update throttle state
            self._last_soc = soc
            self._last_helios_stored_ts = now

        node_id = self.add_node(frame, event_type=event_type)
        log.info(f"node {node_id[:8]}  {source}:{event_type}")

        # ── precedes edge (temporal continuity within same source) ─────────────
        if source in self._last:
            prev_id, prev_ts = self._last[source]
            dt = frame.get("ts", time.time()) - prev_ts
            if 0 < dt <= self.PRECEDES_WINDOW_S:
                self.add_edge(prev_id, node_id, "precedes", weight=1.0 - dt / self.PRECEDES_WINDOW_S)

        self._last[source] = (node_id, frame.get("ts", time.time()))

        # ── track energy budget from HELIOS ticks ─────────────────────────────
        # flat format: budget is top-level; wrapped format: inside payload
        if source == "nexus.star" and "energy_budget" in frame:
            self._last_budget = frame["energy_budget"]
        if source == "nexus.star" and "energy_budget" in payload:
            self._last_budget = payload["energy_budget"]

        # ── THEIA response_start: snapshot energy state ────────────────────────
        if source == "nexus.core" and event_type == "response_start":
            snap = _helios_snapshot()
            if snap:
                self.add_snapshot(node_id, snap)
            self._pending_response = node_id

        # ── THEIA response_end: correlate with energy budget ──────────────────
        if source == "nexus.core" and event_type == "response_end":
            if self._pending_response:
                self.add_edge(self._pending_response, node_id, "correlated_with")
                self._pending_response = None

        # ── SUBSTRATE trigger: link to last HELIOS tick ────────────────────────
        if source == "nexus.core" and event_type == "substrate_trigger":
            if "nexus.star" in self._last:
                helios_id, _ = self._last["nexus.star"]
                self.add_edge(helios_id, node_id, "triggered_by")
            snap = _helios_snapshot()
            if snap:
                self.add_snapshot(node_id, snap)

        # ── SUBSTRATE material result: link to quantum node ────────────────────
        if source == "nexus.quantum":
            if "nexus.star" in self._last:
                helios_id, _ = self._last["nexus.star"]
                self.add_edge(node_id, helios_id, "caused_by")


# ── query REPL ─────────────────────────────────────────────────────────────────
def _query_repl(conn: sqlite3.Connection) -> None:
    print("\nLOGOS Query REPL — commands:")
    print("  nodes [N]         last N nodes (default 10)")
    print("  edges [N]         last N edges")
    print("  path <node_id>    trace causal path from node")
    print("  snapshot          last energy snapshot")
    print("  stats             graph statistics")
    print("  sql <query>       raw SQL")
    print("  exit\n")

    while True:
        try:
            cmd = input("logos> ").strip()
        except (EOFError, KeyboardInterrupt):
            break

        if not cmd or cmd == "exit":
            break

        parts = cmd.split(maxsplit=1)
        verb  = parts[0].lower()
        arg   = parts[1] if len(parts) > 1 else ""

        try:
            if verb == "nodes":
                n = int(arg) if arg else 10
                rows = conn.execute(
                    "SELECT id, ts, source, event_type FROM nodes ORDER BY ts DESC LIMIT ?", (n,)
                ).fetchall()
                for r in rows:
                    print(f"  {r['id'][:8]}  {r['source']:20s}  {r['event_type']:20s}  "
                          f"{time.strftime('%H:%M:%S', time.localtime(r['ts']))}")

            elif verb == "edges":
                n = int(arg) if arg else 10
                rows = conn.execute(
                    "SELECT src, dst, relation, weight FROM edges ORDER BY ts DESC LIMIT ?", (n,)
                ).fetchall()
                for r in rows:
                    print(f"  {r['src'][:8]} --[{r['relation']}]--> {r['dst'][:8]}  w={r['weight']:.3f}")

            elif verb == "snapshot":
                row = conn.execute(
                    "SELECT * FROM snapshots ORDER BY ts DESC LIMIT 1"
                ).fetchone()
                if row:
                    print(dict(row))

            elif verb == "stats":
                n_nodes = conn.execute("SELECT COUNT(*) FROM nodes").fetchone()[0]
                n_edges = conn.execute("SELECT COUNT(*) FROM edges").fetchone()[0]
                n_snap  = conn.execute("SELECT COUNT(*) FROM snapshots").fetchone()[0]
                sources = conn.execute(
                    "SELECT source, COUNT(*) as c FROM nodes GROUP BY source"
                ).fetchall()
                print(f"  nodes={n_nodes}  edges={n_edges}  snapshots={n_snap}")
                for s in sources:
                    print(f"    {s['source']:25s} {s['c']}")

            elif verb == "path":
                nid = arg[:8] if arg else ""
                # find full id
                row = conn.execute(
                    "SELECT id FROM nodes WHERE id LIKE ?", (f"{nid}%",)
                ).fetchone()
                if not row:
                    print("  node not found")
                    continue
                full_id = row["id"]
                # walk back via edges
                visited = set()
                frontier = [full_id]
                depth = 0
                while frontier and depth < 10:
                    next_f = []
                    for nid_ in frontier:
                        if nid_ in visited:
                            continue
                        visited.add(nid_)
                        node = conn.execute(
                            "SELECT source, event_type, ts FROM nodes WHERE id=?", (nid_,)
                        ).fetchone()
                        if node:
                            print(f"  {'  '*depth}{nid_[:8]}  {node['source']}:{node['event_type']}  "
                                  f"{time.strftime('%H:%M:%S', time.localtime(node['ts']))}")
                        parents = conn.execute(
                            "SELECT src FROM edges WHERE dst=?", (nid_,)
                        ).fetchall()
                        next_f.extend(p["src"] for p in parents)
                    frontier = next_f
                    depth += 1

            elif verb == "sql":
                rows = conn.execute(arg).fetchall()
                for r in rows:
                    print(dict(r))

            else:
                print("  unknown command")

        except Exception as exc:
            print(f"  error: {exc}")


# ── main ───────────────────────────────────────────────────────────────────────
def main() -> None:
    parser = argparse.ArgumentParser(description="LOGOS — Temporal Knowledge Graph")
    parser.add_argument("--query", action="store_true", help="Drop into query REPL")
    parser.add_argument("--db",    default=str(_DB_PATH), help="SQLite path")
    args = parser.parse_args()

    conn  = _init_db(Path(args.db))
    graph = LogosGraph(conn)

    if args.query:
        _query_repl(conn)
        conn.close()
        return

    sub = NexusSubscriber()
    sub.start()

    log.info("LOGOS kernel running — Ctrl+C to stop")
    try:
        while True:
            try:
                frame = sub.queue.get(timeout=2.0)
                graph.ingest(frame)
            except Exception:
                continue
    except KeyboardInterrupt:
        log.info("LOGOS shutting down...")
    finally:
        sub.stop()
        conn.close()


if __name__ == "__main__":
    main()