Create app.py

app.py

@@ -0,0 +1,449 @@
+import os
+import json
+import time
+import uuid
+import sqlite3
+import hashlib
+from dataclasses import dataclass
+from typing import List, Dict, Any, Optional, Tuple
+
+import gradio as gr
+
+# =========================
+# RFT Memory Receipt Engine
+# =========================
+
+# For HF Spaces w/ persistent storage, set BASE_DIR to /data/rftmem in Space settings (env var)
+BASE_DIR = os.environ.get("RFT_MEM_BASE", "var/rftmem")
+os.makedirs(BASE_DIR, exist_ok=True)
+
+def sha256_bytes(b: bytes) -> str:
+    return hashlib.sha256(b).hexdigest()
+
+def sha256_str(s: str) -> str:
+    return sha256_bytes(s.encode("utf-8"))
+
+def now_ms() -> int:
+    return int(time.time() * 1000)
+
+def atomic_write(path: str, data: bytes) -> None:
+    tmp = path + ".tmp"
+    with open(tmp, "wb") as f:
+        f.write(data)
+        f.flush()
+        os.fsync(f.fileno())
+    os.replace(tmp, path)
+
+@dataclass
+class RetrievalHit:
+    event_id: str
+    role: str
+    text: str
+    ts_ms: int
+    digest: str
+    chain_hash: str
+    score: float
+
+class RFTMemoryStore:
+    """
+    Source of truth: append-only JSONL per session + hash-chained ledger.
+    Index: SQLite (events table + FTS5) for lexical search.
+    Receipts: JSON files saved per assistant turn.
+    """
+
+    def __init__(self, base_dir: str):
+        self.base_dir = base_dir
+        self.db_path = os.path.join(base_dir, "index.sqlite")
+        self._init_db()
+
+    def _init_db(self):
+        os.makedirs(self.base_dir, exist_ok=True)
+        con = sqlite3.connect(self.db_path)
+        cur = con.cursor()
+
+        # Main event table
+        cur.execute("""
+        CREATE TABLE IF NOT EXISTS events (
+            session_id TEXT,
+            event_id   TEXT PRIMARY KEY,
+            ts_ms      INTEGER,
+            role       TEXT,
+            text       TEXT,
+            digest     TEXT,
+            prev_hash  TEXT,
+            chain_hash TEXT,
+            collapse   REAL
+        )
+        """)
+        # FTS (lexical retrieval)
+        cur.execute("""
+        CREATE VIRTUAL TABLE IF NOT EXISTS events_fts USING fts5(
+            event_id,
+            session_id,
+            text,
+            content=''
+        )
+        """)
+        # Receipts table (optional convenience)
+        cur.execute("""
+        CREATE TABLE IF NOT EXISTS receipts (
+            receipt_id TEXT PRIMARY KEY,
+            session_id TEXT,
+            ts_ms      INTEGER,
+            prompt_hash TEXT,
+            response_hash TEXT,
+            receipt_path TEXT
+        )
+        """)
+        con.commit()
+        con.close()
+
+    # -------------
+    # Session layout
+    # -------------
+    def session_dir(self, session_id: str) -> str:
+        d = os.path.join(self.base_dir, "sessions", session_id)
+        os.makedirs(d, exist_ok=True)
+        return d
+
+    def session_log_path(self, session_id: str) -> str:
+        return os.path.join(self.session_dir(session_id), "events.jsonl")
+
+    def receipts_dir(self, session_id: str) -> str:
+        d = os.path.join(self.session_dir(session_id), "receipts")
+        os.makedirs(d, exist_ok=True)
+        return d
+
+    # -------------------
+    # RFT collapse scoring
+    # -------------------
+    def collapse_score(self, session_id: str, role: str, text: str) -> float:
+        """
+        Simple, effective heuristic:
+        - novelty vs last ~20 events (token Jaccard)
+        - role weight (user/tool > assistant)
+        """
+        role_w = {"user": 1.0, "tool": 0.9, "assistant": 0.6}.get(role, 0.7)
+        tokens = set(t.lower() for t in text.split() if t.strip())
+        if not tokens:
+            return 0.0
+
+        recent = self.get_events(session_id, limit=20)
+        recent_tokens = set()
+        for e in recent:
+            recent_tokens |= set(t.lower() for t in e["text"].split() if t.strip())
+
+        # novelty = fraction of tokens not seen recently
+        unseen = len(tokens - recent_tokens)
+        novelty = unseen / max(1, len(tokens))
+
+        # length sanity cap (avoid 1-word spikes)
+        length_factor = min(1.0, len(tokens) / 30.0)
+
+        score = role_w * (0.65 * novelty + 0.35 * length_factor)
+        return float(max(0.0, min(1.0, score)))
+
+    # ----------------
+    # Append-only write
+    # ----------------
+    def append_event(self, session_id: str, role: str, text: str) -> Dict[str, Any]:
+        event_id = uuid.uuid4().hex
+        ts = now_ms()
+        payload = {
+            "session_id": session_id,
+            "event_id": event_id,
+            "ts_ms": ts,
+            "role": role,
+            "text": text
+        }
+        digest = sha256_str(json.dumps(payload, sort_keys=True))
+        prev_hash = self.get_last_chain_hash(session_id) or "0"*64
+        chain_hash = sha256_str(prev_hash + digest)
+
+        collapse = self.collapse_score(session_id, role, text)
+
+        rec = {
+            **payload,
+            "digest": digest,
+            "prev_hash": prev_hash,
+            "chain_hash": chain_hash,
+            "collapse": collapse
+        }
+
+        # 1) Write JSONL source-of-truth
+        log_path = self.session_log_path(session_id)
+        line = (json.dumps(rec, ensure_ascii=False) + "\n").encode("utf-8")
+        with open(log_path, "ab") as f:
+            f.write(line)
+            f.flush()
+            os.fsync(f.fileno())
+
+        # 2) Update SQLite index (synchronous for demo simplicity)
+        con = sqlite3.connect(self.db_path)
+        cur = con.cursor()
+        cur.execute("""
+            INSERT INTO events(session_id,event_id,ts_ms,role,text,digest,prev_hash,chain_hash,collapse)
+            VALUES(?,?,?,?,?,?,?,?,?)
+        """, (session_id, event_id, ts, role, text, digest, prev_hash, chain_hash, collapse))
+        cur.execute("INSERT INTO events_fts(event_id, session_id, text) VALUES(?,?,?)", (event_id, session_id, text))
+        con.commit()
+        con.close()
+
+        return rec
+
+    def get_last_chain_hash(self, session_id: str) -> Optional[str]:
+        con = sqlite3.connect(self.db_path)
+        cur = con.cursor()
+        cur.execute("""
+            SELECT chain_hash FROM events
+            WHERE session_id=?
+            ORDER BY ts_ms DESC
+            LIMIT 1
+        """, (session_id,))
+        row = cur.fetchone()
+        con.close()
+        return row[0] if row else None
+
+    def get_events(self, session_id: str, limit: int = 200) -> List[Dict[str, Any]]:
+        con = sqlite3.connect(self.db_path)
+        cur = con.cursor()
+        cur.execute("""
+            SELECT event_id, ts_ms, role, text, digest, prev_hash, chain_hash, collapse
+            FROM events
+            WHERE session_id=?
+            ORDER BY ts_ms ASC
+            LIMIT ?
+        """, (session_id, limit))
+        rows = cur.fetchall()
+        con.close()
+        out = []
+        for r in rows:
+            out.append({
+                "event_id": r[0],
+                "ts_ms": r[1],
+                "role": r[2],
+                "text": r[3],
+                "digest": r[4],
+                "prev_hash": r[5],
+                "chain_hash": r[6],
+                "collapse": r[7],
+            })
+        return out
+
+    # -------------------------
+    # Hybrid retrieval (lexical)
+    # -------------------------
+    def search_lexical(self, session_id: str, query: str, k: int = 8) -> List[RetrievalHit]:
+        con = sqlite3.connect(self.db_path)
+        cur = con.cursor()
+
+        # FTS5 bm25 ranking (lower is better). We'll invert into a positive score.
+        cur.execute("""
+            SELECT e.event_id, e.role, e.text, e.ts_ms, e.digest, e.chain_hash,
+                   bm25(events_fts) as rank
+            FROM events_fts
+            JOIN events e ON e.event_id = events_fts.event_id
+            WHERE events_fts.text MATCH ? AND e.session_id=?
+            ORDER BY rank ASC
+            LIMIT ?
+        """, (query, session_id, k))
+
+        rows = cur.fetchall()
+        con.close()
+
+        hits = []
+        for (eid, role, text, ts, digest, chain_hash, rank) in rows:
+            score = 1.0 / (1.0 + float(rank if rank is not None else 0.0))
+            hits.append(RetrievalHit(eid, role, text, ts, digest, chain_hash, score))
+        return hits
+
+    # -------------------------
+    # Receipts + verification
+    # -------------------------
+    def write_receipt(self, session_id: str, user_text: str, retrieved: List[RetrievalHit], prompt: str, response: str) -> str:
+        receipt_id = uuid.uuid4().hex
+        ts = now_ms()
+
+        receipt = {
+            "receipt_id": receipt_id,
+            "session_id": session_id,
+            "ts_ms": ts,
+            "query": user_text,
+            "retrieval": [{
+                "event_id": h.event_id,
+                "digest": h.digest,
+                "chain_hash": h.chain_hash,
+                "score": h.score,
+                "role": h.role,
+                "text": h.text
+            } for h in retrieved],
+            "prompt_hash": sha256_str(prompt),
+            "response_hash": sha256_str(response),
+            "engine": {
+                "name": "RFT Memory Receipt Engine",
+                "version": "0.1-demo",
+                "notes": "append-only JSONL + SQLite FTS + hash-chained ledger"
+            }
+        }
+
+        path = os.path.join(self.receipts_dir(session_id), f"{receipt_id}.json")
+        atomic_write(path, json.dumps(receipt, indent=2, ensure_ascii=False).encode("utf-8"))
+
+        con = sqlite3.connect(self.db_path)
+        cur = con.cursor()
+        cur.execute("""
+            INSERT INTO receipts(receipt_id, session_id, ts_ms, prompt_hash, response_hash, receipt_path)
+            VALUES(?,?,?,?,?,?)
+        """, (receipt_id, session_id, ts, receipt["prompt_hash"], receipt["response_hash"], path))
+        con.commit()
+        con.close()
+        return path
+
+    def verify_receipt(self, receipt_json: Dict[str, Any]) -> Tuple[bool, str]:
+        session_id = receipt_json.get("session_id")
+        if not session_id:
+            return False, "Missing session_id."
+
+        # 1) Verify the referenced events exist and digests match
+        con = sqlite3.connect(self.db_path)
+        cur = con.cursor()
+
+        for item in receipt_json.get("retrieval", []):
+            eid = item.get("event_id")
+            expected_digest = item.get("digest")
+            expected_chain = item.get("chain_hash")
+
+            cur.execute("SELECT digest, chain_hash FROM events WHERE event_id=? AND session_id=?", (eid, session_id))
+            row = cur.fetchone()
+            if not row:
+                con.close()
+                return False, f"Event not found: {eid}"
+            if row[0] != expected_digest:
+                con.close()
+                return False, f"Digest mismatch for {eid}"
+            if row[1] != expected_chain:
+                con.close()
+                return False, f"Chain hash mismatch for {eid}"
+
+        con.close()
+        return True, "Receipt verified: referenced events exist and hashes match."
+
+store = RFTMemoryStore(BASE_DIR)
+
+# --------------------
+# Gradio demonstration
+# --------------------
+def new_session():
+    return uuid.uuid4().hex
+
+def format_events(events: List[Dict[str, Any]]) -> str:
+    lines = []
+    for e in events[-200:]:
+        lines.append(
+            f"{time.strftime('%Y-%m-%d %H:%M:%S', time.gmtime(e['ts_ms']/1000))} | {e['role']}\n"
+            f"{e['text']}\n"
+            f"event_id={e['event_id']}  collapse={e['collapse']:.2f}\n"
+            f"digest={e['digest']}\n"
+            f"chain={e['chain_hash']}\n"
+            f"{'-'*60}"
+        )
+    return "\n".join(lines)
+
+def chat_turn(session_id: str, user_msg: str, retrieval_k: int, use_generation: bool):
+    if not session_id:
+        session_id = new_session()
+
+    store.append_event(session_id, "user", user_msg)
+
+    # Lexical retrieval (demo: per-session scope)
+    hits = store.search_lexical(session_id, user_msg, k=retrieval_k)
+
+    memories = "\n".join([f"- ({h.role}) {h.text}" for h in hits]) if hits else "(none)"
+    prompt = (
+        "SYSTEM: You are a transparent assistant. Use retrieved memories if relevant.\n"
+        f"RETRIEVED MEMORIES:\n{memories}\n\n"
+        f"USER:\n{user_msg}\n"
+    )
+
+    # Demo response (deterministic by default). Optional "generation" placeholder.
+    if use_generation:
+        # Keep this lightweight: no external APIs in demo.
+        response = (
+            "I’m running in demo mode (no external model calls). "
+            "Here are the most relevant retrieved memories I would use:\n\n"
+            f"{memories}\n\n"
+            "If you want real generation, plug your local model call here."
+        )
+    else:
+        response = (
+            "RFT Memory Engine (demo): I retrieved the following memory slices:\n\n"
+            f"{memories}\n\n"
+            "I can verify exactly which memories were used via the receipt."
+        )
+
+    store.append_event(session_id, "assistant", response)
+
+    receipt_path = store.write_receipt(session_id, user_msg, hits, prompt, response)
+
+    events = store.get_events(session_id, limit=500)
+    ledger = format_events(events)
+
+    # Show retrieved block + receipt for transparency
+    retrieved_view = "\n".join([f"{h.score:.4f} | {h.role} | {h.text}" for h in hits]) if hits else "(none)"
+    return session_id, response, retrieved_view, ledger, receipt_path
+
+def search_memory(session_id: str, query: str, k: int):
+    if not session_id:
+        return "(no session)"
+    hits = store.search_lexical(session_id, query, k=k)
+    if not hits:
+        return "(no hits)"
+    return "\n".join([f"{h.score:.4f} | {h.role} | {h.text}\n  event_id={h.event_id}\n  digest={h.digest}\n" for h in hits])
+
+def verify_uploaded_receipt(file_obj):
+    if file_obj is None:
+        return "Upload a receipt JSON file."
+    with open(file_obj.name, "r", encoding="utf-8") as f:
+        data = json.load(f)
+    ok, msg = store.verify_receipt(data)
+    return f"{'✅' if ok else '❌'} {msg}"
+
+with gr.Blocks(title="RFT Memory Receipt Engine (Local Persistence Demo)") as demo:
+    gr.Markdown(
+        "# RFT Memory Receipt Engine\n"
+        "Append-only session ledger + SQLite FTS retrieval + cryptographic receipts.\n\n"
+        "**Goal:** demonstrate durable memory + verifiable retrieval lineage."
+    )
+
+    session_id = gr.Textbox(label="Session ID", value=new_session())
+    with gr.Row():
+        retrieval_k = gr.Slider(1, 20, value=8, step=1, label="Retrieval K")
+        use_generation = gr.Checkbox(value=False, label="Show 'generation' placeholder")
+
+    user_msg = gr.Textbox(label="User message", placeholder="Type a message…")
+    send = gr.Button("Send")
+
+    assistant_out = gr.Textbox(label="Assistant output", lines=8)
+    retrieved_out = gr.Textbox(label="Retrieved memory slices (what influenced this turn)", lines=8)
+    ledger_out = gr.Textbox(label="Session Ledger (hash-chained)", lines=16)
+    receipt_path = gr.Textbox(label="Receipt saved at (server path)", lines=1)
+
+    send.click(
+        chat_turn,
+        inputs=[session_id, user_msg, retrieval_k, use_generation],
+        outputs=[session_id, assistant_out, retrieved_out, ledger_out, receipt_path],
+    )
+
+    gr.Markdown("## Manual Search")
+    search_q = gr.Textbox(label="Search query")
+    search_btn = gr.Button("Search")
+    search_results = gr.Textbox(label="Search results", lines=10)
+    search_btn.click(search_memory, inputs=[session_id, search_q, retrieval_k], outputs=[search_results])
+
+    gr.Markdown("## Verify a Receipt")
+    receipt_file = gr.File(label="Upload receipt JSON")
+    verify_btn = gr.Button("Verify Receipt")
+    verify_out = gr.Textbox(label="Verification result")
+    verify_btn.click(verify_uploaded_receipt, inputs=[receipt_file], outputs=[verify_out])
+
+demo.launch()