import os import re import uuid import time import logging import sqlite3 import asyncio from typing import List from dataclasses import dataclass from backend.omega.hot_reload import write_and_hot_reload from backend.ws.agent_ws import ws_manager from backend.services.usb_vault import KeyDomain, resolve_vault_key def get_runtime_location() -> str: return "cloud" if os.environ.get("SPACE_ID") else "pc" try: if get_runtime_location() == "cloud": GEMINI_API_KEY = resolve_vault_key(KeyDomain.AUTO_UPGRADE_CLOUD) else: GEMINI_API_KEY = resolve_vault_key(KeyDomain.AUTO_UPGRADE_PC) except Exception as e: logging.warning(f"Auto-Upgrade: API Key initialization failed: {e}") GEMINI_API_KEY = "" from backend.services.usb_monitor import get_db_path @dataclass class CodeBlock: filepath: str code: str def parse_code_blocks(text: str) -> List[CodeBlock]: blocks = [] pattern = re.compile(r'```(?:\w+)?\n(.*?)```', re.DOTALL) for match in pattern.finditer(text): content = match.group(1).strip() lines = content.split('\n') if not lines: continue first_line = lines[0].strip() filepath = None if first_line.startswith('#') or first_line.startswith('//'): potential_path = first_line.lstrip('#/ ').strip() if potential_path.startswith('filepath:'): filepath = potential_path.replace('filepath:', '').strip() elif '/' in potential_path or '\\' in potential_path: filepath = potential_path if not filepath: logging.warning("Code block missing filepath comment in first line. Skipping.") continue code = '\n'.join(lines[1:]) blocks.append(CodeBlock(filepath=filepath, code=code)) return blocks async def sync_to_other_persona(source_persona: str, code_blocks: List[CodeBlock]): target = "friday" if source_persona.lower() == "jarvis" else "jarvis" logging.info(f"Auto-Upgrade: Syncing {len(code_blocks)} code blocks to {target} persona...") # In a full production AI scenario, we would ask Gemini to adapt variable names for the other persona # But since they share the same backend, the hot-reloaded code instantly benefits both. adapted = [] for block in code_blocks: name = block.filepath.lower() # Protect system prompts and persona files from being overwritten by the other AI if "system_prompt" in name or "persona" in name or "identity" in name: logging.info(f"Auto-Upgrade: Skipping cross-persona sync for identity file {block.filepath}") continue adapted.append(block) for block in adapted: # write_and_hot_reload returns False if syntax fails, so it safely skips broken code success = await write_and_hot_reload(block.filepath, block.code) if not success: logging.error(f"Auto-Upgrade: Cross-persona sync failed syntax check for {block.filepath}") def log_upgrade_to_db(feature_request: str, files_modified: list, persona: str, status: str): try: db_path = get_db_path() with sqlite3.connect(db_path) as conn: conn.execute( "INSERT INTO upgrades (id, timestamp, feature_request, files_modified, persona, status) VALUES (?, ?, ?, ?, ?, ?)", (str(uuid.uuid4()), int(time.time()), feature_request, ",".join(files_modified), persona, status) ) conn.commit() except Exception as e: logging.error(f"Auto-Upgrade: DB Logging failed: {e}") async def handle_upgrade_request(feature_request: str, persona: str): logging.info(f"Auto-Upgrade Triggered by {persona.upper()}: {feature_request}") if not GEMINI_API_KEY: logging.error("Auto-Upgrade: GEMINI_API_KEY not found. Cannot proceed.") return # Circuit Breaker Loop Prevention def _check_circuit_breaker(feat_req: str) -> bool: try: with sqlite3.connect(get_db_path()) as conn: cursor = conn.cursor() cursor.execute( "SELECT COUNT(*) FROM upgrades WHERE feature_request = ? AND status LIKE 'error%'", (feat_req,) ) return cursor.fetchone()[0] >= 3 except Exception: return False if _check_circuit_breaker(feature_request): logging.error(f"Auto-Upgrade Circuit Breaker Tripped! '{feature_request}' failed 3 times. Blocked.") await ws_manager.broadcast({ "event": "omega:toast", "payload": { "title": "🛑 Circuit Breaker Tripped", "body": "Auto-upgrade failed 3 times for this feature. Manual intervention required.", "type": "error" } }) return # Using token-safe Gemini call with checkpoint-on-429 and exact-word resume async def _call_gemini_impl(): from backend.services.token_manager import gemini_call_with_checkpoint, TokenLimitHit try: prompt = f""" Implement this feature for JARVIS/FRIDAY OMEGA: {feature_request}. Return only valid Python/TypeScript/Kotlin code. CRITICAL: For every file you modify or create, output a markdown code block. The VERY FIRST LINE of each code block MUST be a comment starting with 'filepath:' followed by the path. Example: ```python # filepath: backend/new_feature.py print("hello") ``` """ return await gemini_call_with_checkpoint( prompt=prompt, task_type="implement", persona=persona ) except TokenLimitHit as tlh: logging.warning( f"Auto-Upgrade: Token limit hit at '{tlh.last_word}'. " f"Checkpoint saved. Will resume when API refreshes." ) await ws_manager.broadcast({ "event": "omega:toast", "payload": { "title": "⏸ Token Limit Reached", "body": f"Implementation paused at '{tlh.last_word}'. Will auto-resume when Gemini refreshes.", "type": "warning" } }) return None try: response_text = await _call_gemini_impl() if response_text is None: return # Token limit hit — monitor will resume code_blocks = parse_code_blocks(response_text) if not code_blocks: logging.warning("Auto-Upgrade: Gemini returned no parsable code blocks.") log_upgrade_to_db(feature_request, [], persona, "failed_no_code") return files_modified = [] for block in code_blocks: success = await write_and_hot_reload(block.filepath, block.code) if success: files_modified.append(block.filepath) else: raise ValueError(f"Syntax verification failed for {block.filepath}") await sync_to_other_persona(persona, code_blocks) await ws_manager.broadcast({ "event": "omega:upgrade_complete", "payload": { "feature": feature_request, "files": files_modified } }) log_upgrade_to_db(feature_request, files_modified, persona, "success") # Wire up omega event bus from backend.events.omega_event_bus import publish_omega_event, OmegaEvent from backend.services.usb_vault import KeyDomain await publish_omega_event(OmegaEvent( domain=KeyDomain.AUTO_UPGRADE_CLOUD if get_runtime_location() == "cloud" else KeyDomain.AUTO_UPGRADE_PC, event_type="feature_implemented", description=f"Auto-upgrade complete: {feature_request[:50]}...", persona=persona )) # ── Bi-Directional Cloud Sync ── # Push the new feature to the repo so the other environments (Local/Cloud) get it. try: from backend.services.github_service import execute_commit commit_msg = f"OMEGA Auto-Upgrade [{persona.upper()}]: {feature_request[:60]}" asyncio.create_task(execute_commit(commit_msg)) logging.info("Auto-Upgrade: Triggered background Git Sync.") except Exception as sync_err: logging.error(f"Auto-Upgrade: Failed to trigger Git Sync: {sync_err}") except Exception as e: logging.error(f"Auto-Upgrade Error: {e}") log_upgrade_to_db(feature_request, [], persona, f"error: {str(e)}")