from gevent import monkey monkey.patch_all() import json import time from pathlib import Path import gevent from flask import Flask, request, jsonify, render_template_string from flask_socketio import SocketIO, emit import threading from infj_bot.core.cognitive_orchestrator import CognitiveOrchestrator # Observatory integration — hive_mind is an external symlinked dependency try: import sys as _sys _hive_path = str(Path(__file__).resolve().parent / "hive_mind") if _hive_path not in _sys.path: _sys.path.insert(0, _hive_path) from drift_bridge import DriftBridge as _DriftBridge _obs_drift = _DriftBridge() _OBSERVATORY_ENABLED = True except Exception: _OBSERVATORY_ENABLED = False from infj_bot.core.brain import DriftBrain from infj_bot.core.commands import BotState, handle_command from infj_bot.core.plugins.growth import growth_profile from infj_bot.core.history import ChatHistory from infj_bot.core.memory import DriftMemory from infj_bot.core.plugins.goals import GoalsDB from infj_bot.core.config import DATA_DIR, DEFAULT_AUTHORIZED_TARGETS STATE_ROOT = DATA_DIR from infj_bot.core.plugins.documents import DocumentStore from infj_bot.core.prompt_builder import build_chat_prompt from infj_bot.core.plugins.preferences import PreferenceStore from infj_bot.core.plugins.scheduler import TaskScheduler from infj_bot.core.plugins.self_eval import SelfEvaluator from infj_bot.core.gen_cache import DiskGenCache SESSIONS_DIR = STATE_ROOT / "sessions" SESSIONS_DIR.mkdir(parents=True, exist_ok=True) sessions = {} sessions_lock = threading.Lock() class SessionResources: def __init__(self, session_id: str): self.session_id = session_id if session_id == "global": self.session_dir = STATE_ROOT / "global_session" else: self.session_dir = SESSIONS_DIR / session_id self.session_dir.mkdir(parents=True, exist_ok=True) # 1. State (isolated PreferenceStore and TaskScheduler) self.state = BotState( authorized_targets=set(DEFAULT_AUTHORIZED_TARGETS), prefs=PreferenceStore(db_path=self.session_dir / "preferences.db"), scheduler=TaskScheduler(db_path=self.session_dir / "scheduler.db") ) # 2. Memory self.memory = DriftMemory( persist_directory=str(self.session_dir / "memory") ) # 3. Document Store self.doc_store = DocumentStore( persist_directory=str(self.session_dir / "chroma_db") ) # 4. Goals DB self.goals_db = GoalsDB( db_path=self.session_dir / "goals.db" ) # 5. History self.history = ChatHistory( path=self.session_dir / "history.jsonl" ) # 6. Brain (with custom evaluator and disk cache) evaluator = SelfEvaluator(db_path=self.session_dir / "self_eval.db") try: disk_cache = DiskGenCache( path=self.session_dir / "drift_gen_cache.sqlite3" ) except Exception: disk_cache = None self.brain = DriftBrain(evaluator=evaluator, disk_cache=disk_cache) self.last_accessed = time.time() def close(self): # Close disk cache connection to release file descriptors if hasattr(self.brain, "_disk_cache") and self.brain._disk_cache: try: self.brain._disk_cache.close() except Exception: pass def get_session(session_id: str = None) -> SessionResources: if not session_id: session_id = "global" with sessions_lock: if session_id not in sessions: sessions[session_id] = SessionResources(session_id) else: sessions[session_id].last_accessed = time.time() return sessions[session_id] def get_request_session_id(): """Extracts session_id from JSON payload or cookie fallback.""" session_id = None if request.is_json: try: payload = request.json or {} session_id = payload.get("session_id") except Exception: pass if not session_id: session_id = request.cookies.get("drift_session_id") return session_id def prune_and_cleanup_sessions(): """Prunes inactive in-memory sessions and cleans up old disk folders.""" while True: try: now = time.time() # 1. Prune in-memory sessions (30 mins inactivity) with sessions_lock: inactive_keys = [] for sid, sres in sessions.items(): if sid == "global": continue if now - sres.last_accessed > 1800: # 30 mins inactive_keys.append(sid) for sid in inactive_keys: print(f"Pruning in-memory session: {sid}") sres = sessions.pop(sid) sres.close() # 2. Cleanup old session folders on disk (not accessed for over 7 days) if SESSIONS_DIR.exists(): import shutil for p in SESSIONS_DIR.iterdir(): if p.is_dir(): with sessions_lock: is_loaded = p.name in sessions if not is_loaded: try: mtime = p.stat().st_mtime for filepath in p.rglob("*"): if filepath.is_file(): mtime = max(mtime, filepath.stat().st_mtime) if now - mtime > 7 * 86400: print(f"Deleting expired session folder on disk: {p.name}") shutil.rmtree(p, ignore_errors=True) except Exception: pass except Exception as e: print(f"Error in pruning sessions: {e}") gevent.sleep(300) def background_cognitive_loop(): """Background loop to tick homeostasis, shadow, and embodiment for the active sessions.""" from infj_bot.core.being import get_being from infj_bot.core.homeostasis import get_homeostasis from infj_bot.core.shadow import get_shadow from infj_bot.core.dii_tracker import get_dii_tracker from infj_bot.core.global_workspace import get_workspace being = get_being() homeostasis = get_homeostasis() shadow = get_shadow() tracker = get_dii_tracker() workspace = get_workspace() while True: try: # Update embodiment / heartbeat / breath with some dynamic variations embodiment_plugin = cognitive_orchestrator.arch.get_plugin("embodiment") embodiment = embodiment_plugin.instance if embodiment_plugin else None if embodiment: import random bpm = getattr(embodiment, "heartbeat_bpm", 72) bpm += random.uniform(-1.5, 1.5) bpm = max(60, min(100, bpm)) setattr(embodiment, "heartbeat_bpm", bpm) phase = getattr(embodiment, "breath_phase", "exhale") depth = getattr(embodiment, "breath_depth", 0.65) depth += random.uniform(-0.05, 0.05) depth = max(0.3, min(0.95, depth)) setattr(embodiment, "breath_depth", depth) if random.random() < 0.2: setattr(embodiment, "breath_phase", "inhale" if phase == "exhale" else "exhale") # Let homeostasis and shadow run their background cycles if hasattr(shadow, "background_tick"): shadow.background_tick(being=being) if hasattr(homeostasis, "background_cycle"): homeostasis.background_cycle(being=being) # Compute tracker metrics if hasattr(tracker, "compute"): tracker.compute( being=being, workspace=workspace, homeostasis=homeostasis, shadow=shadow, orchestrator=cognitive_orchestrator, ) except Exception as e: print(f"Error in background cognitive cycle: {e}") gevent.sleep(4.0) _MAX_PAYLOAD = 1_048_576 # 1 MB INDEX_HTML = """ DRIFT // DASHBOARD

DRIFT // NEURAL INTERFACE

ONLINE // SESSION ACTIVE
""" def chat_reply(message, session_res: SessionResources): prompt, emotion, dissonance = build_chat_prompt( message, session_res.state, session_res.memory, goals_db=session_res.goals_db, doc_store=session_res.doc_store, prefs=session_res.state.prefs, ) output = session_res.brain.agent_turn(prompt, tools_enabled=True, raw_user_input=message) try: session_res.brain.evaluate_last(prompt, output) except Exception: pass importance = min( 0.95, 0.45 + emotion["intensity"] * 0.3 + dissonance["score"] * 0.15 ) session_res.memory.save_interaction( message, output, mode=session_res.state.mode, emotion=emotion, importance=importance, dissonance=dissonance, ) session_res.history.append(message, output, session_res.state.mode, emotion, dissonance) session_res.state.turns += 1 return output app = Flask(__name__) import secrets, os app.config["SECRET_KEY"] = os.environ.get("DRIFT_SECRET_KEY", secrets.token_hex(32)) socketio = SocketIO( app, async_mode="gevent", cors_allowed_origins="*", websocket_compression=True ) broadcast_interval = 0.35 total_bytes_raw = 0 total_bytes_compressed = 0 cognitive_orchestrator = CognitiveOrchestrator() # Sandbox / Trial Management trial_sessions = {} # {session_id: start_time} def is_trial_active(session_id): if not session_id or session_id not in trial_sessions: return False # 30 minute limit (1800 seconds) if time.time() - trial_sessions[session_id] > 1800: return False return True @app.route("/api/v1/system/governor-metrics", methods=["GET"]) def governor_metrics(): session_id = get_request_session_id() or "global" session_res = get_session(session_id) brain = session_res.brain gov = getattr(brain, "_governor", None) if gov is None: return jsonify({ "status": "error", "detail": "RateGovernor not initialized. Call self.init_governor() in DriftBrain.__init__." }), 503 return jsonify({"status": "online", **gov.metrics_report()}) @app.route("/expo") def start_expo(): """Alias for /trial — clean URL for sharing demo sessions.""" return start_trial() @app.route("/trial") def start_trial(): import uuid from flask import make_response session_id = str(uuid.uuid4()) trial_sessions[session_id] = time.time() # Simple hack to inject the session into the frontend trial_html = INDEX_HTML.replace( "document.querySelector('#form').onsubmit", f"const DRIFT_SESSION_ID = '{session_id}';\n" + "document.querySelector('#form').onsubmit", ).replace( "async function post(path, body={}) {", "async function post(path, body={}) {\n if(typeof DRIFT_SESSION_ID !== 'undefined') body.session_id = DRIFT_SESSION_ID;", ) resp = make_response(trial_html) resp.set_cookie("drift_session_id", session_id, max_age=1800, httponly=True, samesite="Lax") return resp def broadcast_observatory_state(): global broadcast_interval, total_bytes_raw, total_bytes_compressed while True: try: delta = cognitive_orchestrator.get_delta_state() if len(delta) > 1: raw_size = len(json.dumps(delta)) total_bytes_raw += raw_size total_bytes_compressed += int(raw_size * 0.3) delta["network_stats"] = { "raw_kb": round(total_bytes_raw / 1024, 2), "comp_kb": round(total_bytes_compressed / 1024, 2), "interval_ms": int(broadcast_interval * 1000), } socketio.emit("observatory_delta", delta) except Exception: pass gevent.sleep(broadcast_interval) threading.Thread(target=broadcast_observatory_state, daemon=True).start() @socketio.on("latency_ping") def handle_latency_ping(data): emit( "latency_pong", {"server_time": time.time(), "client_timestamp": data.get("timestamp")}, ) @socketio.on("auto_adjust_rate") def handle_adjust_rate(data): global broadcast_interval target_interval = data.get("interval", 0.35) broadcast_interval = max(0.2, min(target_interval, 1.5)) @app.route("/") def index(): from flask import make_response session_id = request.cookies.get("drift_session_id") had_cookie = True if not session_id: import uuid session_id = str(uuid.uuid4()) had_cookie = False # Inject session_id into JS html = INDEX_HTML.replace( "document.querySelector('#form').onsubmit", f"const DRIFT_SESSION_ID = '{session_id}';\n" + "document.querySelector('#form').onsubmit", ).replace( "async function post(path, body={}) {", "async function post(path, body={}) {\n if(typeof DRIFT_SESSION_ID !== 'undefined') body.session_id = DRIFT_SESSION_ID;", ) resp = make_response(html) if not had_cookie: resp.set_cookie("drift_session_id", session_id, max_age=30*86400, httponly=True, samesite="Lax") return resp @app.route("/api/growth", methods=["GET"]) def get_growth(): session_id = get_request_session_id() session_res = get_session(session_id) return jsonify(growth_profile(session_res.memory, session_res.state.turns)) @app.route("/api/tags", methods=["GET"]) def ollama_tags(): return jsonify( { "models": [ { "name": "infj_bot:latest", "model": "infj_bot:latest", "modified_at": "2023-11-04T14:56:49.277302595-07:00", "size": 7323310500, "digest": "9f438cb9cd581fc025612d27f7c1a6669ff83a8bb0ed86c94fcf4c5440555697", } ] } ) @app.route("/api/chat", methods=["POST"]) def api_chat(): payload = request.json or {} session_id = get_request_session_id() # Check for trial session if session_id and session_id in trial_sessions: if not is_trial_active(session_id): return jsonify( {"error": "Trial session expired. Please start a new session at /trial"} ), 403 session_res = get_session(session_id) # Check if this is an Ollama-style request (from Reins) if "messages" in payload: messages = payload.get("messages", []) user_message = "" for msg in reversed(messages): if msg.get("role") == "user": user_message = msg.get("content", "") break if not user_message: return jsonify({"error": "No user message found"}), 400 reply_text = chat_reply(user_message, session_res) return jsonify( { "model": payload.get("model", "infj_bot:latest"), "created_at": time.strftime("%Y-%m-%dT%H:%M:%S.000Z", time.gmtime()), "message": {"role": "assistant", "content": reply_text}, "done": True, } ) # Standard INFJ Bot UI request message = payload.get("message", "") if isinstance(message, str): message = message.strip() if not message: return jsonify({"error": "message is required"}), 400 return jsonify({"reply": chat_reply(message, session_res)}) @app.route("/v1/chat/completions", methods=["POST", "OPTIONS"]) def openai_chat_completions(): if request.method == "OPTIONS": return "", 200 payload = request.json or {} session_id = get_request_session_id() # Check for trial session expiration if session_id and session_id in trial_sessions: if not is_trial_active(session_id): return jsonify( {"error": "Trial session expired. Please start a new session at /trial"} ), 403 session_res = get_session(session_id) messages = payload.get("messages", []) # Extract the last user message user_message = "" for msg in reversed(messages): if msg.get("role") == "user": user_message = msg.get("content", "") break if not user_message: return jsonify({"error": "No user message found"}), 400 # Get reply from infj_bot reply_text = chat_reply(user_message, session_res) # Format as OpenAI response import uuid response = { "id": f"chatcmpl-{uuid.uuid4().hex}", "object": "chat.completion", "created": int(time.time()), "model": payload.get("model", "infj_bot"), "choices": [ { "index": 0, "message": {"role": "assistant", "content": reply_text}, "finish_reason": "stop", } ], "usage": {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0}, } return jsonify(response) @app.route("/api/command", methods=["POST"]) def api_command(): payload = request.json or {} session_id = get_request_session_id() session_res = get_session(session_id) reply = handle_command( payload.get("command", ""), payload.get("args", ""), session_res.state, session_res.brain, session_res.memory, session_res.history, session_res.goals_db, session_res.doc_store, ) return jsonify({"reply": reply}) @app.route("/api/email", methods=["POST"]) def api_email(): # Email sending is not implemented; no send_email backend available. return jsonify( { "sent": False, "error": "Email sending not implemented (no backend configured).", } ), 501 @app.route("/observatory") def observatory(): try: path = Path(__file__).resolve().parent / "templates" / "observatory.html" if not path.exists(): path = Path("/home/crexs/templates/observatory.html") with open(path, "r", encoding="utf-8") as f: content = f.read() from flask import make_response resp = make_response(content) resp.headers["Content-Type"] = "text/html" return resp except Exception: app.logger.exception("observatory render failed") return "Internal server error", 500 @app.route("/glyph") @app.route("/phi-glyph") def glyph(): try: path = Path(__file__).resolve().parent / "templates" / "phi_glyph_system.html" if not path.exists(): path = Path("/home/crexs/Downloads/PHI Glyph System.html") with open(path, "r", encoding="utf-8") as f: content = f.read() from flask import make_response resp = make_response(content) resp.headers["Content-Type"] = "text/html" return resp except Exception: app.logger.exception("glyph render failed") return "Internal server error", 500 def main(): import os print("🚀 DRIFT Web App: Gevent + Compression + Delta Logic Active") # Spawn background greenlets for cognitive and session management gevent.spawn(background_cognitive_loop) gevent.spawn(prune_and_cleanup_sessions) port = int(os.getenv("PORT", 7860)) socketio.run(app, host="0.0.0.0", port=port, debug=False, use_reloader=False) if __name__ == "__main__": main()