From 75f5b72d3b2fd9e063b70bb2741d45c12f01abbb Mon Sep 17 00:00:00 2001 From: Bryce Johnson Date: Sun, 1 Feb 2026 02:02:38 -0500 Subject: [PATCH 1/2] Use SQLite for raw items + inferences (keep JSON fallback) --- README.md | 11 ++- app/config.py | 31 ++++++ app/db/__init__.py | 1 + app/db/sqlite.py | 229 +++++++++++++++++++++++++++++++++++++++++++++ app/main.py | 104 ++++++++++---------- 5 files changed, 327 insertions(+), 49 deletions(-) create mode 100644 app/config.py create mode 100644 app/db/__init__.py create mode 100644 app/db/sqlite.py diff --git a/README.md b/README.md index b8b823d..a12847e 100644 --- a/README.md +++ b/README.md @@ -6,7 +6,8 @@ This application is a "Tinder for Inferences" triage tool. It is designed to hel ## Features * **Triage Interface**: Swipe-like interface to Approve (True) or Reject (False) inferences. * **Context Aware**: Shows the source data (e.g., the text message) alongside the AI's conclusion. -* **JSON Persistence**: Saves your validated knowledge graph components for later use by your "Ares" agent. +* **Durable storage**: Uses a local **SQLite** database for inferences/raw items (with legacy JSON fallback). +* **Export**: Saves your validated knowledge graph components for later use by your "Ares" agent. ## Getting Started @@ -25,13 +26,19 @@ This application is a "Tinder for Inferences" triage tool. It is designed to hel ```bash python scripts/generate_mock_data.py ``` - *This creates a `inferences.json` file with sample connections (e.g., linking iMessage mentions to Instagram profiles).* + *This creates an `inferences.json` file with sample connections (legacy format; it will be imported into SQLite on startup).* 2. **Start the Server**: ```bash uvicorn app.main:app --reload ``` +Optional environment variables: +```bash +export JARVIS_SQLITE_PATH="./jarvis.sqlite3" +export JARVIS_PROCESS_BATCH_SIZE="25" +``` + 3. **Open the App**: Navigate to `http://localhost:8000` in your browser. diff --git a/app/config.py b/app/config.py new file mode 100644 index 0000000..ada7fdb --- /dev/null +++ b/app/config.py @@ -0,0 +1,31 @@ +"""App configuration (env-driven). + +Keep this module dependency-free; it is imported very early. +""" + +from __future__ import annotations + +import os +from pathlib import Path + + +def _env(name: str, default: str) -> str: + v = os.environ.get(name) + return v if v is not None and v != "" else default + + +# Where to store local DB/state (relative paths are resolved from cwd). +DATA_DIR = Path(_env("JARVIS_DATA_DIR", ".")).expanduser().resolve() + +SQLITE_PATH = Path(_env("JARVIS_SQLITE_PATH", str(DATA_DIR / "jarvis.sqlite3"))).expanduser().resolve() + +# Legacy JSON paths (used for import/compat and as a fallback) +INFERENCES_JSON_PATH = Path(_env("JARVIS_INFERENCES_JSON", str(DATA_DIR / "inferences.json"))).expanduser().resolve() +RAW_DATA_JSON_PATH = Path(_env("JARVIS_RAW_DATA_JSON", str(DATA_DIR / "raw_data.json"))).expanduser().resolve() + +# Server +HOST = _env("JARVIS_HOST", "0.0.0.0") +PORT = int(_env("JARVIS_PORT", "8000")) + +# Processing +PROCESS_BATCH_SIZE = int(_env("JARVIS_PROCESS_BATCH_SIZE", "25")) diff --git a/app/db/__init__.py b/app/db/__init__.py new file mode 100644 index 0000000..3f3f0a4 --- /dev/null +++ b/app/db/__init__.py @@ -0,0 +1 @@ +# DB package diff --git a/app/db/sqlite.py b/app/db/sqlite.py new file mode 100644 index 0000000..3463105 --- /dev/null +++ b/app/db/sqlite.py @@ -0,0 +1,229 @@ +"""SQLite persistence layer. + +Design goals: +- zero external deps (stdlib only) +- safe defaults: local file only +- keep API compatible with existing frontend fields + +We store: +- inferences (triage candidates) +- raw_data_items (ingested raw snippets) + +This is not the final schema for the full project, but it is a durable +stepping-stone vs JSON files. +""" + +from __future__ import annotations + +import json +import sqlite3 +from dataclasses import asdict +from datetime import datetime +from pathlib import Path +from typing import Any, Dict, Iterable, List, Optional, Tuple + +from app.models import RawDataItem + + +def connect(path: Path) -> sqlite3.Connection: + path.parent.mkdir(parents=True, exist_ok=True) + con = sqlite3.connect(str(path)) + con.row_factory = sqlite3.Row + return con + + +def init_db(con: sqlite3.Connection) -> None: + con.execute( + """ + CREATE TABLE IF NOT EXISTS raw_items ( + id TEXT PRIMARY KEY, + source TEXT NOT NULL, + content TEXT NOT NULL, + timestamp TEXT NOT NULL, + metadata_json TEXT + ); + """ + ) + + con.execute( + """ + CREATE TABLE IF NOT EXISTS inferences ( + id TEXT PRIMARY KEY, + source TEXT NOT NULL, + content TEXT NOT NULL, + inference TEXT NOT NULL, + confidence REAL NOT NULL, + status TEXT NOT NULL DEFAULT 'pending', + user_notes TEXT, + created_at TEXT NOT NULL + ); + """ + ) + con.commit() + + +# -------------------------- Raw items -------------------------- + +def upsert_raw_items(con: sqlite3.Connection, items: Iterable[RawDataItem]) -> int: + rows = 0 + for item in items: + con.execute( + """ + INSERT INTO raw_items(id, source, content, timestamp, metadata_json) + VALUES(?, ?, ?, ?, ?) + ON CONFLICT(id) DO UPDATE SET + source=excluded.source, + content=excluded.content, + timestamp=excluded.timestamp, + metadata_json=excluded.metadata_json + """, + ( + item.id, + item.source, + item.content, + item.timestamp.isoformat(), + json.dumps(item.metadata or {}, ensure_ascii=False), + ), + ) + rows += 1 + con.commit() + return rows + + +def list_raw_items(con: sqlite3.Connection, limit: int = 1000) -> List[Dict[str, Any]]: + cur = con.execute( + "SELECT id, source, content, timestamp, metadata_json FROM raw_items ORDER BY timestamp DESC LIMIT ?", + (limit,), + ) + out: List[Dict[str, Any]] = [] + for r in cur.fetchall(): + md = {} + try: + md = json.loads(r["metadata_json"]) if r["metadata_json"] else {} + except Exception: + md = {} + out.append( + { + "id": r["id"], + "source": r["source"], + "content": r["content"], + "timestamp": r["timestamp"], + "metadata": md, + } + ) + return out + + +# -------------------------- Inferences -------------------------- + +def insert_inference( + con: sqlite3.Connection, + *, + inference_id: str, + source: str, + content: str, + inference: str, + confidence: float, + status: str = "pending", + user_notes: Optional[str] = None, +) -> None: + con.execute( + """ + INSERT INTO inferences(id, source, content, inference, confidence, status, user_notes, created_at) + VALUES(?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + inference_id, + source, + content, + inference, + float(confidence), + status, + user_notes, + datetime.utcnow().isoformat(), + ), + ) + con.commit() + + +def get_next_pending_inference(con: sqlite3.Connection) -> Optional[Dict[str, Any]]: + cur = con.execute( + """ + SELECT id, source, content, inference, confidence, status, user_notes + FROM inferences + WHERE status='pending' + ORDER BY created_at ASC + LIMIT 1 + """ + ) + row = cur.fetchone() + return dict(row) if row else None + + +def update_inference_status(con: sqlite3.Connection, inference_id: str, status: str, notes: Optional[str]) -> bool: + cur = con.execute( + """ + UPDATE inferences + SET status=?, user_notes=COALESCE(?, user_notes) + WHERE id=? + """, + (status, notes, inference_id), + ) + con.commit() + return cur.rowcount > 0 + + +def list_inferences(con: sqlite3.Connection, status: Optional[str] = None) -> List[Dict[str, Any]]: + if status: + cur = con.execute( + "SELECT id, source, content, inference, confidence, status, user_notes FROM inferences WHERE status=? ORDER BY created_at ASC", + (status,), + ) + else: + cur = con.execute( + "SELECT id, source, content, inference, confidence, status, user_notes FROM inferences ORDER BY created_at ASC" + ) + return [dict(r) for r in cur.fetchall()] + + +# -------------------------- Migration helpers -------------------------- + +def import_legacy_json_inferences(con: sqlite3.Connection, json_path: Path) -> int: + if not json_path.exists(): + return 0 + try: + data = json.loads(json_path.read_text()) + except Exception: + return 0 + if not isinstance(data, list): + return 0 + + inserted = 0 + for item in data: + try: + inference_id = str(item.get("id")) + source = str(item.get("source")) + content = str(item.get("content")) + inference = str(item.get("inference")) + confidence = float(item.get("confidence", 0.0)) + status = str(item.get("status", "pending")) + user_notes = item.get("user_notes") + # Avoid duplicates + cur = con.execute("SELECT 1 FROM inferences WHERE id=?", (inference_id,)) + if cur.fetchone(): + continue + insert_inference( + con, + inference_id=inference_id, + source=source, + content=content, + inference=inference, + confidence=confidence, + status=status, + user_notes=user_notes, + ) + inserted += 1 + except Exception: + continue + + return inserted diff --git a/app/main.py b/app/main.py index b31906c..ef3f589 100644 --- a/app/main.py +++ b/app/main.py @@ -1,20 +1,26 @@ import json import os from typing import List, Optional + from fastapi import FastAPI, HTTPException, Request from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates from fastapi.responses import JSONResponse from pydantic import BaseModel + +from app import config from app.brain import brain # Import the Brain +from app.db import sqlite as db_sqlite from app.ingestors.chatgpt import ChatGPTIngestor from app.ingestors.safari import SafariIngestor app = FastAPI() # --- Configuration --- -DATA_FILE = "inferences.json" -RAW_DATA_FILE = "raw_data.json" +# Prefer SQLite for durability. Keep legacy JSON paths for import/fallback. +SQLITE_PATH = config.SQLITE_PATH +DATA_FILE = str(config.INFERENCES_JSON_PATH) +RAW_DATA_FILE = str(config.RAW_DATA_JSON_PATH) # --- Models --- class IngestRequest(BaseModel): @@ -75,6 +81,12 @@ def update_status(self, inference_id: str, status: str, notes: str = None): self.save_all(data) return found +# --- Persistence --- +con = db_sqlite.connect(SQLITE_PATH) +db_sqlite.init_db(con) +# One-time best-effort import of legacy JSON inferences so existing users keep state. +db_sqlite.import_legacy_json_inferences(con, config.INFERENCES_JSON_PATH) + db = InferencesDB(DATA_FILE) # --- Routes --- @@ -89,7 +101,7 @@ async def read_root(request: Request): @app.get("/api/inference") async def get_next_inference(): - inference = db.get_pending() + inference = db_sqlite.get_next_pending_inference(con) if not inference: return {"message": "No pending inferences"} return inference @@ -97,7 +109,7 @@ async def get_next_inference(): @app.post("/api/triage") async def triage_inference(request: TriageRequest): new_status = "approved" if request.action == "approve" else "rejected" - success = db.update_status(request.id, new_status, request.notes) + success = db_sqlite.update_inference_status(con, request.id, new_status, request.notes) if not success: raise HTTPException(status_code=404, detail="Inference not found") return {"status": "success"} @@ -105,26 +117,27 @@ async def triage_inference(request: TriageRequest): @app.get("/api/export") async def export_consciousness(): """Export all APPROVED inferences for Ares.""" - all_data = db.load_all() - # Filter for approved inferences - spirit_data = [item for item in all_data if item.get("status") == "approved"] + spirit_data = db_sqlite.list_inferences(con, status="approved") return JSONResponse( content=spirit_data, - headers={"Content-Disposition": "attachment; filename=ares_consciousness.json"} + headers={"Content-Disposition": "attachment; filename=ares_consciousness.json"}, ) @app.post("/api/generate") async def trigger_generation(): - """Trigger the Brain to generate a new inference (Manual trigger for now).""" - # In a real app, this would loop through the database. - # Here we just generate one random mock/AI inference. + """Trigger the Brain to generate a new inference (manual trigger for now).""" new_inference = await brain.generate_inference("Random Source", "Random content snippet...") - # Save to DB - all_data = db.load_all() - all_data.append(new_inference) - db.save_all(all_data) + db_sqlite.insert_inference( + con, + inference_id=new_inference["id"], + source=new_inference["source"], + content=new_inference["content"], + inference=new_inference["inference"], + confidence=new_inference.get("confidence", 0.0), + status=new_inference.get("status", "pending"), + ) return {"status": "generated", "id": new_inference["id"]} @@ -140,60 +153,57 @@ async def trigger_ingest(source: str, request: IngestRequest): items = ingestor.ingest() - # Persist to raw data file + # Persist to SQLite (primary) + stored = db_sqlite.upsert_raw_items(con, items) + + # Also persist to legacy raw_data.json for now (compat/debug) raw_data = [] if os.path.exists(RAW_DATA_FILE): try: with open(RAW_DATA_FILE, "r") as f: raw_data = json.load(f) - except: - pass + except Exception: + raw_data = [] - # Convert Pydantic models to dicts new_items = [item.dict() for item in items] - - # Append new items (simple append for now, duplicates possible) - # In a real system, we'd check IDs raw_data.extend(new_items) - with open(RAW_DATA_FILE, "w", default=str) as f: - # custom serializer for datetime if needed, or rely on pydantic .dict() handling it mostly? - # Actually pydantic .dict() keeps datetime objects which json.dump fails on. - # We need to serialize properly. + with open(RAW_DATA_FILE, "w") as f: json.dump(raw_data, f, indent=2, default=str) - return {"status": "success", "source": source, "items_count": len(items)} + return {"status": "success", "source": source, "items_count": len(items), "stored": stored} @app.post("/api/process") async def process_raw_data(): - """Trigger the Brain to process all raw data.""" - if not os.path.exists(RAW_DATA_FILE): - return {"status": "no_data", "inferences_generated": 0} + """Trigger the Brain to process raw data and generate triage inferences.""" - with open(RAW_DATA_FILE, "r") as f: - raw_items = json.load(f) + # Prefer SQLite raw items; fall back to legacy json. + raw_items = db_sqlite.list_raw_items(con, limit=config.PROCESS_BATCH_SIZE) + if not raw_items and os.path.exists(RAW_DATA_FILE): + with open(RAW_DATA_FILE, "r") as f: + raw_items = json.load(f) - generated_count = 0 - new_inferences = [] + if not raw_items: + return {"status": "no_data", "inferences_generated": 0} - # Process batch (limit to 5 for now to avoid freezing) - for item in raw_items[:5]: - # Skip if we already have an inference for this content? (Simplification: process all) + generated_count = 0 - # Call Brain - # Use brain.process_raw_data logic (we need to update brain.py first or inline it here) - # Let's use the existing generate_inference method for now + # Process batch + for item in raw_items[: config.PROCESS_BATCH_SIZE]: inference = await brain.generate_inference(item["source"], item["content"]) - new_inferences.append(inference) + db_sqlite.insert_inference( + con, + inference_id=inference["id"], + source=inference["source"], + content=inference["content"], + inference=inference["inference"], + confidence=inference.get("confidence", 0.0), + status=inference.get("status", "pending"), + ) generated_count += 1 - # Save Inferences - all_data = db.load_all() - all_data.extend(new_inferences) - db.save_all(all_data) - return {"status": "success", "inferences_generated": generated_count} if __name__ == "__main__": import uvicorn - uvicorn.run(app, host="0.0.0.0", port=8000) + uvicorn.run(app, host=config.HOST, port=config.PORT) From 8631ee40a15381bcc09bdad582ded54a47573de3 Mon Sep 17 00:00:00 2001 From: Bryce Johnson Date: Sun, 1 Feb 2026 03:59:25 -0500 Subject: [PATCH 2/2] Add anomaly register + triage ranking (Brenner appendix + objective) --- app/db/sqlite.py | 51 ++++++++++++++++++++++++++++++++++++++++++++++++ app/main.py | 9 ++++++--- app/ranking.py | 47 ++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 104 insertions(+), 3 deletions(-) create mode 100644 app/ranking.py diff --git a/app/db/sqlite.py b/app/db/sqlite.py index 3463105..0a62dfc 100644 --- a/app/db/sqlite.py +++ b/app/db/sqlite.py @@ -59,6 +59,21 @@ def init_db(con: sqlite3.Connection) -> None: ); """ ) + + # Brenner-style "appendix": quarantine anomalies/exceptions without deleting + con.execute( + """ + CREATE TABLE IF NOT EXISTS anomaly_register ( + id TEXT PRIMARY KEY, + inference_id TEXT, + raw_item_id TEXT, + kind TEXT, -- contradiction | exception | data_quality + note TEXT, + created_at TEXT NOT NULL + ); + """ + ) + con.commit() @@ -227,3 +242,39 @@ def import_legacy_json_inferences(con: sqlite3.Connection, json_path: Path) -> i continue return inserted + + +# -------------------------- Anomalies -------------------------- + +def insert_anomaly( + con: sqlite3.Connection, + *, + anomaly_id: str, + inference_id: Optional[str], + raw_item_id: Optional[str], + kind: str, + note: str, +) -> None: + con.execute( + """ + INSERT INTO anomaly_register(id, inference_id, raw_item_id, kind, note, created_at) + VALUES(?, ?, ?, ?, ?, ?) + """, + ( + anomaly_id, + inference_id, + raw_item_id, + kind, + note, + datetime.utcnow().isoformat(), + ), + ) + con.commit() + + +def list_anomalies(con: sqlite3.Connection, limit: int = 200) -> List[Dict[str, Any]]: + cur = con.execute( + "SELECT id, inference_id, raw_item_id, kind, note, created_at FROM anomaly_register ORDER BY created_at DESC LIMIT ?", + (limit,), + ) + return [dict(r) for r in cur.fetchall()] diff --git a/app/main.py b/app/main.py index ef3f589..11c6bb9 100644 --- a/app/main.py +++ b/app/main.py @@ -13,6 +13,7 @@ from app.db import sqlite as db_sqlite from app.ingestors.chatgpt import ChatGPTIngestor from app.ingestors.safari import SafariIngestor +from app.ranking import triage_score app = FastAPI() @@ -101,10 +102,12 @@ async def read_root(request: Request): @app.get("/api/inference") async def get_next_inference(): - inference = db_sqlite.get_next_pending_inference(con) - if not inference: + # Prefer ranking among pending items (high-signal first) + pending = db_sqlite.list_inferences(con, status="pending") + if not pending: return {"message": "No pending inferences"} - return inference + pending.sort(key=triage_score, reverse=True) + return pending[0] @app.post("/api/triage") async def triage_inference(request: TriageRequest): diff --git a/app/ranking.py b/app/ranking.py new file mode 100644 index 0000000..9d96d6e --- /dev/null +++ b/app/ranking.py @@ -0,0 +1,47 @@ +"""Ranking heuristics for triage. + +Brenner-ish objective (adapted): + score ≈ (expected discriminability × option value) / (cost × ambiguity) + +We don't have explicit expected information gain estimates yet, so this is a +simple approximation to improve triage order. + +Notes: +- Works on the current API fields (id/source/content/inference/confidence). +- If richer schema is present (source_ids, type), we can use it. +""" + +from __future__ import annotations + +from typing import Any, Dict, List + + +def triage_score(inf: Dict[str, Any]) -> float: + conf = float(inf.get("confidence", 0.0) or 0.0) + + # Multi-source bonus if available + src_ids = inf.get("source_ids") + multi_bonus = 1.0 + if isinstance(src_ids, list) and len(src_ids) >= 2: + multi_bonus = 1.15 + + # Ambiguity proxy: longer text tends to pack multiple claims. + text = (inf.get("inference") or inf.get("statement") or "") + length = max(len(text), 1) + length_penalty = 1.0 + if length > 220: + length_penalty = 0.80 + elif length > 140: + length_penalty = 0.90 + + # Source bonus (rough) + source = (inf.get("source") or "").lower() + source_bonus = 1.0 + if source in ("chatgpt", "imessage", "safari"): + source_bonus = 1.05 + + return conf * multi_bonus * length_penalty * source_bonus + + +def sort_inferences(infs: List[Dict[str, Any]]) -> List[Dict[str, Any]]: + return sorted(infs, key=triage_score, reverse=True)