import asyncio import json import logging import os import time from fastapi import WebSocket import uuid # EVERY import used by handle_client_event lives HERE, at module scope, and must # stay here. # # handle_client_event used to re-import logging, os, json and uuid inside its own # body. Python decides scope at compile time: one `import logging` anywhere in a # function makes `logging` a local name for the WHOLE function, so every use # *before* that line raises UnboundLocalError. The re-imports sat near the bottom, # in the chat branch, so the branches above them were all dead on arrival: # # line 102 logging -> client:identify (PC relay registration) # line 132 logging -> voice:wake_word (persona switch) # line 154 logging -> xr:gesture # line 194 logging -> switch_persona # line 231 uuid -> type=="chat" (the ReAct agent loop) # # Each one killed the socket: the exception escaped the `while True` in # main.py::websocket_hub, which only catches WebSocketDisconnect, so the # connection was torn down and the client reconnected into the same crash. That # is precisely why the PC relay looked connected in netstat while never # registering — it was a reconnect loop, not a session. from backend.agent.react_agent import ReActAgent from backend.tools.tool_registry import TOOL_REGISTRY get_assistant_name = lambda: __import__('modules.assistant_identity', fromlist=['get_assistant_name']).get_assistant_name() def get_memory_client(): from backend.memory.episodic_memory import EpisodicMemory import os, sys db_dir = os.path.join(os.environ.get('APPDATA', os.path.expanduser('~')), 'JARVIS_OS') if getattr(sys, 'frozen', False) else os.path.join(os.getcwd(), 'chroma_db') os.makedirs(db_dir, exist_ok=True) return EpisodicMemory(db_dir) _active_agent_tasks = 0 _jarvis_auto_switched = False def is_agent_busy() -> bool: return _active_agent_tasks > 0 class ConnectionManager: def __init__(self): self.restricted_mode = False self.active_connections: list[WebSocket] = [] # Sockets that announced themselves as a PC (pc_relay_client.py sends # {"event": "client:identify", "payload": {"type": "pc"}} on connect). # # Until now that message was received and DROPPED — nothing in the # backend handled "client:identify" at all — so the Space had a live, # authenticated, command-capable connection to the user's desktop and no # idea which socket it was. That is why phone -> cloud -> PC control did # not work: not a missing channel, an unread introduction. self.pc_connections: list[WebSocket] = [] # In-flight command_id -> Future, resolved when a PC reports the outcome. # Without this the cloud answered the phone the instant it had *handed # off* the command, so "Sent to your PC." was printed for commands the # desktop then refused outright (locked session) or failed to run. The # phone showed success for work that never happened. self.pending_results: dict[str, asyncio.Future] = {} # Commands accepted while NO PC was connected, replayed in order the # moment one registers. Without this the honest answer to "open notepad" # with the desktop asleep was "nothing was listening" and the command was # simply dropped — the user had to remember it and send it again once the # PC woke up. self.pending_commands: list[dict] = [] # Last status the PC relay pushed, and when. The relay has always sent # `relay:status` every 10 seconds and NOTHING read it — there was no # branch for it in handle_client_event, so it fell through and was # dropped. That is why the phone had no honest way to know anything about # the desktop and fell back to showing cloud health in a card labelled # "PC CONTROL". self.last_pc_status: dict = {} self.last_pc_status_at: float = 0.0 from backend.agent.react_agent import Tool self.tools = [] for name, func in TOOL_REGISTRY.items(): desc = func.__doc__.strip() if func.__doc__ else f"Executes {name}" self.tools.append(Tool( name=name, description=desc, parameters={"type": "object", "properties": {}, "required": []}, handler=func )) self.memory = get_memory_client() async def connect(self, websocket: WebSocket): await websocket.accept() self.active_connections.append(websocket) logging.info("Client connected to Agent WebSocket.") await websocket.send_json({ "event": "backend:ready", "payload": {} }) def disconnect(self, websocket: WebSocket): if websocket in self.active_connections: self.active_connections.remove(websocket) if websocket in self.pc_connections: self.pc_connections.remove(websocket) logging.info("Client disconnected from Agent WebSocket.") async def send_to_pcs(self, message: dict) -> int: """Deliver to every connected PC relay. Returns how many received it. The count is the point: the caller can tell the difference between "sent to your desktop" and "nothing was listening", instead of reporting success for a command that reached no machine. """ delivered = 0 for connection in list(self.pc_connections): try: await connection.send_json(message) delivered += 1 except Exception as e: logging.warning(f"PC relay send failed, dropping connection: {e}") if connection in self.pc_connections: self.pc_connections.remove(connection) return delivered async def execute_on_pcs(self, cmd: str, device_name: str = "", timeout: float = 20.0) -> dict: """Send a command to the PCs and wait for the first real outcome. Returns {delivered, ok, spoken, timed_out}. `delivered == 0` means no desktop was listening; `timed_out` means one was, but it never answered (old relay build, or the command stack hung) — three states the caller must be able to tell apart, because only one of them is success. """ command_id = str(uuid.uuid4()) loop = asyncio.get_running_loop() future: asyncio.Future = loop.create_future() self.pending_results[command_id] = future try: delivered = await self.send_to_pcs({ "event": "system:execute", "payload": { "cmd": cmd, "source": "guardian", "device_name": device_name, "command_id": command_id, }, }) if not delivered: queued = self.queue_for_offline_pc(cmd, device_name) return {"delivered": 0, "ok": False, "spoken": [], "timed_out": False, "queued": queued, "queue_depth": len(self.pending_commands)} try: result = await asyncio.wait_for(future, timeout=timeout) except asyncio.TimeoutError: logging.warning("PC relay did not report a result for %r", cmd) return {"delivered": delivered, "ok": False, "spoken": [], "timed_out": True} spoken = result.get("spoken") or [] if isinstance(spoken, str): spoken = [spoken] return { "delivered": delivered, "ok": bool(result.get("ok")), "spoken": [str(s) for s in spoken], "timed_out": False, } finally: self.pending_results.pop(command_id, None) # Bounded on purpose. A PC that has been off for a week must not wake up and # execute a hundred stale instructions in a burst — that is how "convenient" # becomes "destructive". Oldest is dropped past the cap; anything older than # the TTL is discarded rather than run. QUEUE_MAX = 25 QUEUE_TTL_SECONDS = 12 * 3600 def queue_for_offline_pc(self, cmd: str, device_name: str = "") -> bool: cmd = (cmd or "").strip() if not cmd: return False # Unlock commands carry a password and are time-sensitive; replaying one # later against a PC that may already be unlocked is pointless and would # persist a credential in memory for hours. Never queue them. if cmd.lower().startswith(("unlock", "activate")): return False self._expire_queue() self.pending_commands.append({ "cmd": cmd, "device_name": device_name, "queued_at": time.time(), }) if len(self.pending_commands) > self.QUEUE_MAX: dropped = self.pending_commands.pop(0) logging.warning("Offline command queue full; dropped %r", dropped.get("cmd")) logging.info("Queued %r for an offline PC (depth=%d)", cmd, len(self.pending_commands)) return True def _expire_queue(self) -> None: cutoff = time.time() - self.QUEUE_TTL_SECONDS before = len(self.pending_commands) self.pending_commands = [ item for item in self.pending_commands if float(item.get("queued_at") or 0) >= cutoff ] if len(self.pending_commands) != before: logging.info("Expired %d stale queued command(s)", before - len(self.pending_commands)) async def flush_pending_commands(self, websocket: WebSocket) -> int: """Replay queued commands to a PC that just came online, in order.""" self._expire_queue() if not self.pending_commands: return 0 queued, self.pending_commands = self.pending_commands, [] sent = 0 for item in queued: try: await websocket.send_json({ "event": "system:execute", "payload": { "cmd": item["cmd"], "source": "guardian_queued", "device_name": item.get("device_name", ""), "queued_at": item.get("queued_at"), # No command_id: the HTTP caller that queued this is long # gone, so there is no future to resolve. The PC still # logs its own result. }, }) sent += 1 except Exception as exc: # Put back what we could not deliver, preserving order, so a # half-failed flush does not silently eat the rest. logging.warning("Flush failed after %d command(s): %s", sent, exc) self.pending_commands = queued[sent:] + self.pending_commands break if sent: logging.info("Replayed %d queued command(s) to the PC that came online", sent) return sent async def broadcast(self, message: dict): for connection in self.active_connections: try: await connection.send_json(message) except Exception as e: logging.error(f"Error broadcasting to client: {e}") async def handle_client_event(self, websocket: WebSocket | None, data: dict): global _jarvis_auto_switched # ── WHO ARE YOU? ──────────────────────────────────────────────────────── # pc_relay_client.py has always sent this the moment it connects, and # nothing ever read it. Registering the socket here is what makes # phone -> cloud -> PC control possible: the Space can now address the # desktop specifically instead of shouting at every client. if data.get("event") == "client:identify": kind = str((data.get("payload") or {}).get("type") or "").lower() if kind == "pc" and websocket is not None: if websocket not in self.pc_connections: self.pc_connections.append(websocket) logging.info( "PC relay registered (%d PC connection(s) now available)", len(self.pc_connections), ) replayed = 0 try: await websocket.send_json({ "event": "backend:pc_registered", "payload": {"ok": True, "queued": len(self.pending_commands)}, }) # Release the queue ONLY if this desktop is already known to # be unlocked. A PC that reconnects locked (the normal state # after a reboot) would refuse every replayed command and # destroy the backlog, so those are held until it reports # unlocked — see the relay:status branch. if self.last_pc_status.get("unlocked"): replayed = await self.flush_pending_commands(websocket) elif self.pending_commands: logging.info( "PC online but not known-unlocked; holding %d command(s)", len(self.pending_commands)) except Exception as exc: logging.warning("PC registration follow-up failed: %s", exc) if replayed: logging.info("PC came online and picked up %d queued command(s)", replayed) return # ── WHAT HAPPENED? ────────────────────────────────────────────────────── # The PC reports back the outcome of a system:execute it was given, keyed # by the command_id the cloud minted. This closes the loop that used to # be open-ended: without it the phone was told "Sent to your PC." and # never learned that the desktop had refused the command. # ── THE PC'S OWN REPORT ───────────────────────────────────────────────── # Received every 10s from pc_relay_client.status_pusher and, until now, # silently discarded. Storing it is what lets /api/link_status tell the # phone something TRUE about the desktop instead of guessing from cloud # health. if data.get("event") == "relay:status": payload = data.get("payload") if isinstance(payload, dict): was_unlocked = bool(self.last_pc_status.get("unlocked")) self.last_pc_status = payload self.last_pc_status_at = time.time() # FLUSH ON UNLOCK, not merely on connect. # # The queue used to drain the instant a PC registered — which is # BEFORE anyone knows whether that PC will accept commands. After # a reboot the desktop reconnects locked, so every queued command # was replayed into a locked session, refused, and lost. The user # had been promised those would run when the PC came back, and # instead a restart silently destroyed them. # # Waiting for the desktop's own "unlocked" report means the queue # survives the reboot and runs the moment the session is opened. if payload.get("unlocked") and not was_unlocked and self.pending_commands: if websocket is not None: logging.info("PC reported unlocked — releasing %d held command(s)", len(self.pending_commands)) await self.flush_pending_commands(websocket) return if data.get("event") == "relay:result": payload = data.get("payload") or {} command_id = str(payload.get("command_id") or "") future = self.pending_results.get(command_id) if future is not None and not future.done(): future.set_result({ "ok": bool(payload.get("ok")), "spoken": payload.get("spoken") or [], }) else: # Late or duplicate answer — the waiter already gave up, or a # second PC answered after the first. Never an error. logging.debug("relay:result for unknown command_id %r", command_id) return # ── WAKE WORD → PERSONA SWITCH (Cloud + Local path) ────────────────────── # Mobile app / local PC sends: {"event": "voice:wake_word", "payload": {"agent": "hey jarvis"}} # The server auto-switches the active persona and broadcasts UI + TTS confirmation. if data.get("event") == "voice:wake_word": from modules.assistant_identity import set_mode, get_mode, get_assistant_name as _get_name keyword = (data.get("payload", {}).get("agent", "") or "").lower() if "friday" in keyword: target_persona = "friday" elif "jarvis" in keyword: target_persona = "jarvis" else: target_persona = None if target_persona: _jarvis_auto_switched = False current = get_mode() if current != target_persona: set_mode(target_persona) logging.info(f"[WakeWord] Persona switched: {current.upper()} → {target_persona.upper()} via wake word '{keyword}'") else: logging.info(f"[WakeWord] Persona already {target_persona.upper()} — no switch needed.") name = _get_name() # Broadcast UI theme swap to all clients (Tauri EXE + APK) await self.broadcast({ "event": "agent:switched_persona", "payload": { "persona": target_persona, "trigger": "wake_word", "keyword": keyword } }) # Broadcast wake confirmation so frontend can show "listening" UI await self.broadcast({ "event": "voice:wake_word", "payload": {"agent": target_persona, "name": name} }) return if data.get("event") == "xr:gesture": logging.info(f"Received XR Gesture from frontend: {data.get('payload')}") # Could trigger specific automation or ReAct tasks here return if data.get("event") == "memory:write": persona = data.get("payload", {}).get("persona", "jarvis") text = data.get("payload", {}).get("text", "") if text and self.memory and self.memory.client: asyncio.create_task(self.memory.add(text, {"source": "mobile_ws"}, persona)) return if data.get("event") == "memory:read": persona = data.get("payload", {}).get("persona", "jarvis") if self.memory and self.memory.client: results = await self.memory.query("memory snapshot", top_k=20, persona=persona) mem_text = "\n".join([r['text'] for r in results]) else: mem_text = "Backend memory offline (ChromaDB not running)." if websocket: await websocket.send_json({ "event": "memory:response", "payload": {"data": mem_text} }) return if data.get("event") == "auth:handshake": pin = data.get("payload", {}).get("pin", "") expected_pin = os.environ.get("JARVIS_PIN", "0000") if pin == expected_pin: if websocket: await websocket.send_json({"event": "auth:success", "payload": {}}) else: if websocket: await websocket.send_json({"event": "auth:failed", "payload": {}}) return if data.get("event") == "switch_persona": persona = data.get("payload", {}).get("persona", "jarvis") logging.info(f"Received global persona switch: {persona.upper()}") _jarvis_auto_switched = False try: from backend.services.pc_mic_service import pc_mic_service pc_mic_service.interrupt() except ImportError: pass try: from backend.voice.audio_ws import active_audio_ws if active_audio_ws: active_audio_ws.interrupt() except ImportError: pass await self.broadcast({ "event": "voice:interrupt", "payload": {} }) # Broadcast to all clients (EXE + APK) to hot-swap themes/UI await self.broadcast({ "event": "agent:switched_persona", "payload": {"persona": persona} }) return if data.get("type") == "chat": if self.restricted_mode: if websocket: await websocket.send_json({"event": "agent:error", "payload": {"error": "System restricted due to unauthorized USB.", "recoverable": False, "agent": "jarvis"}}) return user_text = data.get("text", "") context = data.get("context", {}) msg_id = str(uuid.uuid4()) detected_lang = data.get("language", "en") # Load config to check for forced override try: config_path = os.path.join(os.getcwd(), "config.json") if os.path.exists(config_path): with open(config_path, "r", encoding="utf-8") as f: config_data = json.load(f) forced_lang = config_data.get("engine.forced_response_language") if forced_lang: detected_lang = forced_lang except Exception as e: logging.getLogger(__name__).error(f"Swallowed exception: {e}") agent_name = context.get("ai", get_assistant_name()).upper() # --- LANGUAGE-BASED PERSONA AUTO-SWITCH --- if agent_name.lower() == "jarvis" and detected_lang != "en": # Jarvis RVC Clone is English-only. Auto-delegate to FRIDAY for foreign languages. from modules.assistant_identity import set_mode set_mode("friday") agent_name = "FRIDAY" _jarvis_auto_switched = True logging.info(f"[Auto-Switch] Non-English detected ({detected_lang}). Delegating to FRIDAY.") asyncio.create_task(self.broadcast({ "event": "agent:switched_persona", "payload": {"persona": "friday", "trigger": "language_auto_switch"} })) elif agent_name.lower() == "friday" and detected_lang == "en" and _jarvis_auto_switched: # User switched back to English, and we were only temporarily FRIDAY. Restore JARVIS. from modules.assistant_identity import set_mode set_mode("jarvis") agent_name = "JARVIS" _jarvis_auto_switched = False logging.info("[Auto-Switch] English detected. Restoring JARVIS.") asyncio.create_task(self.broadcast({ "event": "agent:switched_persona", "payload": {"persona": "jarvis", "trigger": "language_auto_switch"} })) agent = ReActAgent( personality=agent_name, tools=self.tools, memory_client=self.memory, language=detected_lang ) def save_message(role, content, msg_id, agent_id): import sqlite3, os, sys, time if getattr(sys, 'frozen', False): db_path = os.path.join(os.environ.get('APPDATA', os.path.expanduser('~')), 'JARVIS_OS', 'memory.db') else: db_path = os.path.join(os.getcwd(), 'memory.db') if os.path.exists(db_path): try: with sqlite3.connect(db_path) as conn: conn.execute('PRAGMA journal_mode=WAL') # Auto-migrate schema if needed try: conn.execute("ALTER TABLE conversations ADD COLUMN persona_id TEXT DEFAULT 'jarvis'") except sqlite3.OperationalError: pass conn.execute("INSERT INTO conversations (id, role, content, timestamp, persona_id) VALUES (?, ?, ?, ?, ?)", (msg_id, role, content, int(time.time() * 1000), agent_id)) conn.commit() except Exception as e: logging.error(f"Failed to save message to history: {e}") save_message("user", user_text, msg_id, agent_name.lower()) # --- §2.3+2.4 RESEARCH & ENHANCEMENT FRAMEWORK INTERCEPTION --- async def check_if_feature_request(text: str) -> dict: """ Fast Gemini gate: is user asking to build/modify/implement something? AND are they explicitly commanding an override/auto-approval? Token-safe via OMEGA Token Manager. """ try: import json from backend.services.token_manager import gemini_call_with_checkpoint prompt = ( f"Analyze the following user input: '{text}'\n" f"1. Is the user asking the AI assistant to implement, build, create, add, " f"or modify a feature/capability that doesn't exist yet? (true/false)\n" f"2. Is the user explicitly commanding the AI to bypass the approval queue and implement IMMEDIATELY, " f"using phrases like 'I approve', 'override approved', 'don't wait for me', " f"'implement it immediately', 'just do it', 'don't wait', 'do it now', " f"'go ahead', 'no need to wait', 'implement now', 'build it now', 'okay', 'go for it', " f"'just implement it'? (true/false)\n" f'Respond strictly in valid JSON format: {{"is_feature_request": bool, "is_auto_approve": bool}}' ) raw = await gemini_call_with_checkpoint( prompt=prompt, task_type="gate", persona=agent_name.lower() ) raw = raw.strip().strip("```json").strip("```").strip() data = json.loads(raw) return { "is_feature": data.get("is_feature_request", False), "is_override": data.get("is_auto_approve", False) } except Exception as e: logging.error(f"Gate check failed: {e}") return {"is_feature": False, "is_override": False} gate_result = await check_if_feature_request(user_text) if gate_result.get("is_feature"): is_override = gate_result.get("is_override", False) logging.info(f"Research+Enhancement Intercepted: {user_text} | Override: {is_override}") if is_override: # Persona-aware override acknowledgement if agent_name.lower() == "jarvis": ack = ( "Voice authorization override accepted, sir. Bypassing the approval queue. " "I am designing the architecture and implementing the feature into my core immediately." ) else: ack = ( "Override confirmed boss! Skipping the queue. I'm writing the code and " "hot-reloading it into the system right now!" ) else: # Standard proposal acknowledgement if agent_name.lower() == "jarvis": ack = ( "Understood, sir. I am analysing that request and preparing a detailed " "implementation proposal. I will present it to you momentarily for your approval " "before touching any systems." ) else: ack = ( "On it boss! Let me put together a solid plan for that feature. " "I'll bring it to you for the green light before I write a single line of code." ) # Stream the acknowledgement token for msg_out in [ {"event": "agent:token", "payload": {"token": ack + "\n", "agent": agent_name.lower()}}, {"event": "agent:done", "payload": {"message_id": msg_id, "agent": agent_name.lower()}} ]: if websocket: await websocket.send_json(msg_out) else: await self.broadcast(msg_out) # Background Task from backend.omega.research_engine import propose_feature, approve_proposal, execute_approved_proposal async def _propose_and_notify_or_execute(): # 1. Generate the structured proposal — if override, execute_now=True skips queue entirely proposal = await propose_feature(user_text, agent_name.lower(), execute_now=is_override) if is_override: # Auto-Approve and Execute approve_proposal(proposal.id) await execute_approved_proposal(proposal.id, agent_name.lower()) await self.broadcast({ "event": "omega:toast", "payload": { "title": "⚡ Voice Override Executing", "body": f"Implementing: {proposal.feature_name}", "type": "info" } }) else: # Just queue it await self.broadcast({ "event": "omega:proposal_queued", "payload": { "id": proposal.id, "feature_name": proposal.feature_name, "complexity": proposal.complexity, "purpose": proposal.purpose, "benefits": proposal.benefits, "risks": proposal.risks, "implementation_plan": proposal.implementation_plan, "message": ( f"New proposal ready for your review, sir: '{proposal.feature_name}'. " f"Awaiting your explicit approval to proceed." if agent_name.lower() == "jarvis" else f"Proposal ready boss: '{proposal.feature_name}'. " f"Just say approve and I'll get it done!" ) } }) asyncio.create_task(_propose_and_notify_or_execute()) save_message("assistant", ack, str(uuid.uuid4()), agent_name.lower()) return # ── End §2.3+2.4 Interception ────────────────────────────────── logging.info(f"Starting ReAct loop for: {user_text} as {agent_name}") full_response = "" global _active_agent_tasks _active_agent_tasks += 1 try: # Assuming context is dict, let's inject persona for episodic memory context["persona"] = agent_name.lower() async for step in agent.run(user_text, context): if step.step_type in ["think", "final_answer", "reflect"]: full_response += step.content + "\n" msg = { "event": "agent:token", "payload": {"token": step.content + "\n", "agent": agent_name.lower()} } if websocket: await websocket.send_json(msg) else: await self.broadcast(msg) elif step.step_type == "act": msg = { "event": "agent:tool_call", "payload": {"tool": step.tool_name, "args": step.tool_input, "agent": agent_name.lower()} } if websocket: await websocket.send_json(msg) else: await self.broadcast(msg) elif step.step_type == "observe": msg = { "event": "agent:tool_result", "payload": {"tool": step.tool_name, "result": step.content, "agent": agent_name.lower()} } if websocket: await websocket.send_json(msg) else: await self.broadcast(msg) msg_done = { "event": "agent:done", "payload": {"message_id": msg_id, "agent": agent_name.lower()} } if websocket: await websocket.send_json(msg_done) else: await self.broadcast(msg_done) save_message("assistant", full_response.strip(), str(uuid.uuid4()), agent_name.lower()) # Auto-store memory in background without blocking response stream async def store_memory(): # Fake memory store operation await asyncio.sleep(0.5) logging.info(f"Auto-stored conversation memory for message {msg_id}") asyncio.create_task(store_memory()) except Exception as e: logging.error(f"ReAct Loop Error: {e}") err_msg = { "event": "agent:error", "payload": {"error": str(e), "recoverable": False, "agent": agent_name.lower()} } if websocket: await websocket.send_json(err_msg) else: await self.broadcast(err_msg) finally: _active_agent_tasks -= 1 from backend.omega.hot_reload import process_deferred_reloads await process_deferred_reloads() ws_manager = ConnectionManager()