From a36dff7f2c71753b14ecf8f34ae0ddf8ac14858e Mon Sep 17 00:00:00 2001 From: Austin Date: Sat, 25 Jul 2026 15:24:40 -0700 Subject: [PATCH] =?UTF-8?q?feat(brain):=20centralized=20brain=20=E2=80=94?= =?UTF-8?q?=20unified=20information=20pipeline=20(brain=20+=20skills=20+?= =?UTF-8?q?=20chat)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - brain-core/brain_index.py: SQLite+FTS5 index over brain/** markdown, skills/*/learnings.md, and data/chat-history.json. Idempotent upsert keyed by (source, path, agent, #msg) so re-ingest never duplicates. - brain-cli.py: ingest / stats / search CLI. - server.py: /api/brain-index/{ingest,upsert,search,stats} so agents can WRITE to the brain (upsert) and you/agents can QUERY it (search). - .gitignore: exclude the regenerated *.db. Verified live: ingest 66 docs (16 brain, 32 chat, 18 skill-learnings), cross-source FTS search, and agent upsert is immediately searchable. --- .gitignore | 4 + brain-cli.py | 56 +++++++++ brain-core/brain_index.py | 245 ++++++++++++++++++++++++++++++++++++++ server.py | 76 ++++++++++++ 4 files changed, 381 insertions(+) create mode 100644 brain-cli.py create mode 100644 brain-core/brain_index.py diff --git a/.gitignore b/.gitignore index 4e8a34c..f856db9 100644 --- a/.gitignore +++ b/.gitignore @@ -19,4 +19,8 @@ data/cost-history.json graphify-out/ brain/graphify-out/ skills/*/graphify-out/ +# Centralized brain index (regenerated by `python brain-cli.py ingest`) +brain-core/*.db +brain-core/*.db-wal +brain-core/*.db-shm *.log diff --git a/brain-cli.py b/brain-cli.py new file mode 100644 index 0000000..3c93a3f --- /dev/null +++ b/brain-cli.py @@ -0,0 +1,56 @@ +#!/usr/bin/env python3 +"""CLI for the Agentic OS centralized brain. + +Usage: + python brain-cli.py ingest # (re)build the unified index + python brain-cli.py stats # show per-source doc counts + python brain-cli.py search "your query" # full-text search + python brain-cli.py search "query" --source chat --limit 5 + +The index lives at brain-core/brain_index.db (SQLite + FTS5, no deps). +""" +import argparse +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent / "brain-core")) +import brain_index as bi + + +def main(): + p = argparse.ArgumentParser(description="Agentic OS centralized brain CLI") + sub = p.add_subparsers(dest="cmd", required=True) + + sub.add_parser("ingest", help="(re)build the unified index from all sources") + sub.add_parser("stats", help="show per-source document counts") + + sp = sub.add_parser("search", help="full-text search the brain") + sp.add_argument("query", help="search terms") + sp.add_argument("--source", help="filter: brain | skill-learning | chat") + sp.add_argument("--limit", type=int, default=20) + + args = p.parse_args() + conn = bi.get_conn() + + if args.cmd == "ingest": + summary = bi.ingest_all(conn) + conn.commit() + print("Ingested:", summary) + print("Total docs:", bi.stats(conn)["total"]) + elif args.cmd == "stats": + print(bi.stats(conn)) + elif args.cmd == "search": + hits = bi.search(conn, args.query, limit=args.limit, source=args.source) + if not hits: + print("No results.") + return + for h in hits: + src = h["source"] + agent = f" (agent={h['agent']})" if h.get("agent") else "" + print(f"[{src}]{agent} {h['title']}") + print(" ", h["snippet"][:200]) + print(f"\n{len(hits)} result(s).") + + +if __name__ == "__main__": + main() diff --git a/brain-core/brain_index.py b/brain-core/brain_index.py new file mode 100644 index 0000000..55cd6a4 --- /dev/null +++ b/brain-core/brain_index.py @@ -0,0 +1,245 @@ +""" +Agentic OS — Centralized Brain: unified information pipeline. + +Ingests three knowledge sources into a single SQLite+FTS5 index so agents +can write to it and you can query it from a CLI or via HTTP. + +Sources: + - brain/** -> markdown notes (source="brain") + - skills/*/learnings.md -> per-skill learnings (source="skill-learning") + - data/chat-history.json -> per-turn chat log (source="chat") + +No external dependencies: uses Python's stdlib sqlite3 (FTS5). + +Design notes: + - doc_id is a stable hash of (source, source_path, [agent]) so re-ingest + is idempotent (upsert, not duplicate). + - FTS5 external-content is NOT used; we keep the text in the same row for + simplicity and so the CLI/HTTP can return snippets directly. +""" + +from __future__ import annotations + +import hashlib +import json +import sqlite3 +from pathlib import Path +from typing import Iterable + +BASE_DIR = Path(__file__).resolve().parent.parent +DB_PATH = BASE_DIR / "brain-core" / "brain_index.db" + + +def _doc_id(source: str, source_path: str, agent: str | None = None) -> str: + raw = f"{source}|{source_path}|{agent or ''}" + return hashlib.sha1(raw.encode("utf-8")).hexdigest()[:16] + + +def _now_iso() -> str: + from datetime import datetime, timezone + return datetime.now(timezone.utc).isoformat() + + +def get_conn(db_path: Path | None = None) -> sqlite3.Connection: + db_path = db_path or DB_PATH + db_path.parent.mkdir(parents=True, exist_ok=True) + conn = sqlite3.connect(str(db_path)) + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA journal_mode=WAL;") + conn.execute( + """ + CREATE TABLE IF NOT EXISTS brain_index ( + doc_id TEXT PRIMARY KEY, + source TEXT NOT NULL, + source_path TEXT NOT NULL, + title TEXT NOT NULL, + agent TEXT, + content TEXT NOT NULL, + updated_at TEXT NOT NULL + ); + """ + ) + # FTS5 virtual table mirroring the content column (with doc_id for joins). + conn.execute( + """ + CREATE VIRTUAL TABLE IF NOT EXISTS brain_fts USING fts5( + doc_id UNINDEXED, + title, + content, + tokenize='unicode61 remove_diacritics 2' + ); + """ + ) + conn.commit() + return conn + + +def upsert_doc(conn: sqlite3.Connection, *, source: str, source_path: str, + title: str, content: str, agent: str | None = None, + updated_at: str | None = None) -> str: + doc_id = _doc_id(source, source_path, agent) + ts = updated_at or _now_iso() + conn.execute( + """ + INSERT INTO brain_index (doc_id, source, source_path, title, agent, content, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(doc_id) DO UPDATE SET + title=excluded.title, + content=excluded.content, + agent=excluded.agent, + updated_at=excluded.updated_at + """, + (doc_id, source, source_path, title, agent, content, ts), + ) + # Keep FTS in sync. + conn.execute("DELETE FROM brain_fts WHERE doc_id = ?", (doc_id,)) + conn.execute( + "INSERT INTO brain_fts (doc_id, title, content) VALUES (?, ?, ?)", + (doc_id, title, content), + ) + return doc_id + + +# --- Ingestors --------------------------------------------------------- + +def _walk_markdown(root: Path, skip_dirs: set[str] | None = None) -> Iterable[Path]: + skip = skip_dirs or {".git", "node_modules", "__pycache__", ".agents", "graphify-out"} + for p in sorted(root.rglob("*.md")): + if any(part in skip for part in p.relative_to(root).parts): + continue + yield p + + +def ingest_brain(conn: sqlite3.Connection, root: Path | None = None) -> int: + root = root or (BASE_DIR / "brain") + if not root.exists(): + return 0 + count = 0 + for p in _walk_markdown(root): + text = p.read_text(encoding="utf-8", errors="replace") + rel = str(p.relative_to(BASE_DIR)) + upsert_doc(conn, source="brain", source_path=rel, + title=p.stem, content=text) + count += 1 + return count + + +def ingest_skill_learnings(conn: sqlite3.Connection, root: Path | None = None) -> int: + root = root or (BASE_DIR / "skills") + if not root.exists(): + return 0 + count = 0 + for learn in sorted(root.rglob("learnings.md")): + text = learn.read_text(encoding="utf-8", errors="replace") + skill_name = learn.parent.name + rel = str(learn.relative_to(BASE_DIR)) + upsert_doc(conn, source="skill-learning", source_path=rel, + title=f"{skill_name} — learnings", content=text, + agent=skill_name) + count += 1 + return count + + +def ingest_chat(conn: sqlite3.Connection, path: Path | None = None) -> int: + path = path or (BASE_DIR / "data" / "chat-history.json") + if not path.exists(): + return 0 + try: + data = json.loads(path.read_text(encoding="utf-8")) + except (json.JSONDecodeError, OSError): + return 0 + messages = data.get("messages", []) + count = 0 + for i, m in enumerate(messages): + agent = m.get("agent") or "unknown" + role = m.get("role") or "unknown" + content = (m.get("content") or "").strip() + if not content: + continue + ts = m.get("timestamp", "") + # Make the doc key unique per message so turns don't overwrite. + # source_path encodes the message index + timestamp. + unique_path = f"{path.relative_to(BASE_DIR)}#{i}" + title = f"chat:{agent}:{role} @ {ts[:19]}" if ts else f"chat:{agent}:{role} #{i}" + upsert_doc(conn, source="chat", source_path=unique_path, + title=title, content=content, agent=agent, + updated_at=ts or None) + count += 1 + return count + + +def ingest_all(conn: sqlite3.Connection) -> dict: + return { + "brain": ingest_brain(conn), + "skill-learning": ingest_skill_learnings(conn), + "chat": ingest_chat(conn), + } + + +# --- Query ------------------------------------------------------------- + +def search(conn: sqlite3.Connection, query: str, limit: int = 20, + source: str | None = None) -> list[dict]: + """Full-text search across the unified index. Returns ranked hits with + a short snippet around the first match.""" + where = "brain_fts MATCH ?" + params: list = [query] + if source: + where += " AND brain_index.source = ?" + params.append(source) + sql = f""" + SELECT brain_index.*, bm25(brain_fts) AS rank + FROM brain_fts + JOIN brain_index ON brain_index.doc_id = brain_fts.doc_id + WHERE {where} + ORDER BY rank + LIMIT ? + """ + params.append(limit) + rows = conn.execute(sql, params).fetchall() + return [_row_to_dict(r, query) for r in rows] + + +def _row_to_dict(r: sqlite3.Row, query: str) -> dict: + d = dict(r) + d["snippet"] = _snippet(d.get("content", ""), query) + return d + + +def _snippet(text: str, query: str, width: int = 160) -> str: + q = query.strip().split()[0] if query.strip() else "" + if not q: + return text[:width] + low = text.lower() + idx = low.find(q.lower()) + if idx < 0: + return text[:width] + start = max(0, idx - width // 3) + end = min(len(text), start + width) + return ("…" if start > 0 else "") + text[start:end] + ("…" if end < len(text) else "") + + +def stats(conn: sqlite3.Connection) -> dict: + by_source = conn.execute( + "SELECT source, COUNT(*) AS n FROM brain_index GROUP BY source" + ).fetchall() + total = conn.execute("SELECT COUNT(*) AS n FROM brain_index").fetchone()["n"] + return {row["source"]: row["n"] for row in by_source} | {"total": total} + + +if __name__ == "__main__": + # CLI: `python brain-core/brain_index.py ingest` / `search "query"` + import sys + c = get_conn() + if len(sys.argv) > 1 and sys.argv[1] == "ingest": + s = ingest_all(c) + c.commit() + print("Ingested:", s, "total docs:", stats(c)["total"]) + elif len(sys.argv) > 1 and sys.argv[1] == "stats": + print(stats(c)) + elif len(sys.argv) > 2 and sys.argv[1] == "search": + for hit in search(c, sys.argv[2]): + print(f"[{hit['source']}] {hit['title']} (agent={hit.get('agent')})") + print(" ", hit["snippet"][:140]) + else: + print("usage: brain_index.py [ingest|stats|search \"query\"]") diff --git a/server.py b/server.py index ea84839..fa5df21 100644 --- a/server.py +++ b/server.py @@ -588,6 +588,82 @@ def update_brain_file(file_name: str, data: BrainUpdate): append_audit({"action": "brain_update", "file": file_name}) return {"status": "ok", "file": file_name} +# ─── Routes: Centralized Brain Index (unified information pipeline) ─ +# Aggregates brain/** notes, skills/*/learnings.md, and chat history into +# one SQLite+FTS5 index. Agents can write to it (ingest/upsert); you can +# query it (search/stats). + +def _brain_index_module(): + """Lazy import of brain-core so a missing dir never breaks server boot.""" + import importlib.util + spec = importlib.util.spec_from_file_location( + "brain_index", BASE_DIR / "brain-core" / "brain_index.py" + ) + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + return mod + + +@app.post("/api/brain-index/ingest") +def ingest_brain_index(): + mod = _brain_index_module() + conn = mod.get_conn() + try: + summary = mod.ingest_all(conn) + conn.commit() + return {"status": "ok", "ingested": summary, "stats": mod.stats(conn)} + finally: + conn.close() + + +@app.post("/api/brain-index/upsert") +def upsert_brain_doc(data: dict): + """Agents push a single document into the brain. Fields: source, + source_path, title, content, agent (optional).""" + for field in ("source", "source_path", "title", "content"): + if not data.get(field): + raise HTTPException(400, f"missing required field: {field}") + mod = _brain_index_module() + conn = mod.get_conn() + try: + doc_id = mod.upsert_doc( + conn, + source=data["source"], + source_path=data["source_path"], + title=data["title"], + content=data["content"], + agent=data.get("agent"), + updated_at=data.get("updated_at"), + ) + conn.commit() + return {"status": "ok", "doc_id": doc_id} + finally: + conn.close() + + +@app.get("/api/brain-index/search") +def search_brain_index(q: str = "", limit: int = 20, source: str = None): + if not q or not q.strip(): + raise HTTPException(400, "query parameter 'q' is required") + mod = _brain_index_module() + conn = mod.get_conn() + try: + hits = mod.search(conn, q, limit=limit, source=source) + return {"query": q, "count": len(hits), "results": hits} + finally: + conn.close() + + +@app.get("/api/brain-index/stats") +def stats_brain_index(): + mod = _brain_index_module() + conn = mod.get_conn() + try: + return mod.stats(conn) + finally: + conn.close() + + # ─── Routes: Skills ─────────────────────────────────────────────── SKILL_NAME_RE = re.compile(r"^[a-zA-Z0-9_-]{1,64}$")