# -*- coding: utf-8 -*- import os, json, time, glob, uuid, re from typing import List, Dict, Any, Optional, Tuple from dataclasses import dataclass, field, asdict try: from src.core.event_ledger import EventLedger except ImportError: class EventLedger: # type: ignore def __init__(self, *args: Any, **kwargs: Any) -> None: pass def record_event(self, *args: Any, **kwargs: Any) -> None: pass @dataclass class MissionIntent: project_name: Optional[str] = None project_name_source: Optional[str] = None # EXPLICIT_PROJECT_DECLARATION, EXPLICIT_WORKSPACE_DECLARATION, SAFE_SEMANTIC_INFERENCE workspace_intent: str = "CURRENT" # NEW | CURRENT page_title: Optional[str] = None brand_name: Optional[str] = None requested_files: List[str] = field(default_factory=list) requested_features: List[str] = field(default_factory=list) button_text: Optional[str] = None content_text: Optional[str] = None content_requirements: Dict[str, Any] = field(default_factory=dict) visual_requirements: Dict[str, Any] = field(default_factory=dict) def to_dict(self) -> Dict[str, Any]: return asdict(self) class MissionDispatcher: def __init__(self, workspace_root: Optional[str] = None, ledger: Optional[EventLedger] = None, storage_dir: Optional[str] = None): self.workspace_root = os.path.abspath(workspace_root) if workspace_root else None self.ledger = ledger self._missions_cache: Dict[str, Dict[str, Any]] = {} self._active_mission_id = None self._active_mission_by_conv: Dict[str, str] = {} self._active_mission_by_instance: Dict[str, str] = {} self._last_project_name_source: Optional[str] = None if storage_dir: self.storage_dir = os.path.abspath(storage_dir) else: app_root = r"C:\OptimusProjects\Optimus Developer" self.storage_dir = os.path.join(app_root, "data", "missions") os.makedirs(self.storage_dir, exist_ok=True) def _get_path(self, mission_id: str) -> str: safe_id = "".join(c for c in mission_id if c.isalnum() or c in ("-", "_")) return os.path.join(self.storage_dir, f"{safe_id}.json") def clear_active_mission(self, conv_id: Optional[str] = None, instance_id: Optional[str] = None) -> None: if conv_id and conv_id in self._active_mission_by_conv: del self._active_mission_by_conv[conv_id] if instance_id and instance_id in self._active_mission_by_instance: del self._active_mission_by_instance[instance_id] self._active_mission_id = None def parse_mission_intent(self, prompt: str) -> MissionIntent: intent = MissionIntent() prompt_lower = prompt.lower() q_char = chr(34) + chr(39) # 1. Page Title extraction # Matches: con el título "X", con título "X", cambia el título a "X", título "X" title_pat = r'(?i)(?:con\s+(?:el\s+)?t[ií]tulo|cambia\s+el\s+t[ií]tulo\s+a|t[ií]tulo(?:\s+de\s+(?:la\s+)?p[aá]gina)?)\s*[:=]?\s*[' + q_char + r']([^' + q_char + r']+)[' + q_char + r']' m_title = re.search(title_pat, prompt) if m_title: intent.page_title = m_title.group(1).strip() # 2. Brand Name extraction # Matches: para una marca llamada "X", para una tienda llamada "X", marca "X" brand_pat = r'(?i)(?:para\s+(?:una\s+)?(?:marca|tienda|empresa|negocio)\s+(?:llamada|llamado|nombrada|nombrado|de)?|(?:marca|tienda))\s*[:=]?\s*[' + q_char + r']([^' + q_char + r']+)[' + q_char + r']' m_brand = re.search(brand_pat, prompt) if m_brand: intent.brand_name = m_brand.group(1).strip() # 3. Requested Files extraction # Matches: archivo llamado "X", crea un archivo llamado "X", archivo "X" files_pat = r'(?i)(?:(?:crea|crear)\s+(?:un\s+)?)?(?:archivo|fichero|file)(?:\s+llamado)?\s*[' + q_char + r']([^' + q_char + r']+)[' + q_char + r']' m_files = re.findall(files_pat, prompt) if m_files: intent.requested_files.extend([f.strip() for f in m_files if f.strip()]) m_files_unquoted = re.findall( r'(?i)(?:archivo|fichero|file)(?:\s+llamado)?\s+([a-zA-Z0-9_\-\.]+\.[a-zA-Z0-9]+)', prompt ) for f in m_files_unquoted: if f.strip() not in intent.requested_files: intent.requested_files.append(f.strip()) # 4. Button Text extraction btn_pat = r'(?i)bot[oó]n\s*[' + q_char + r']([^' + q_char + r']+)[' + q_char + r']' m_btn = re.search(btn_pat, prompt) if m_btn: intent.button_text = m_btn.group(1).strip() # 5. Content / Message Text extraction msg_pat = r'(?i)(?:mensaje|texto|agrega\s+el\s+texto|con\s+el\s+mensaje)\s*[' + q_char + r']([^' + q_char + r']+)[' + q_char + r']' m_msg = re.search(msg_pat, prompt) if m_msg: intent.content_text = m_msg.group(1).strip() # 6. Workspace intent check explicit_new_indicators = ( "crea un proyecto nuevo", "crea un nuevo proyecto", "crear un proyecto nuevo", "crear un nuevo proyecto", "nuevo proyecto llamado", "crea proyecto nuevo", "inicia un proyecto nuevo", "inicia un nuevo proyecto", "iniciar un proyecto nuevo", "crea una carpeta nueva", "crear una carpeta nueva", "crea una página web nueva", "crea una pagina web nueva", "crea un sitio web nuevo", "crear una página web nueva", "crear una pagina web nueva", "crear un sitio web nuevo", "proyecto nuevo llamado", "carpeta nueva llamada", "workspace nuevo llamado", "workspace llamado", "crea una página web para", "crea una pagina web para", "crea un sitio web para", "crear una página web para", "crear una pagina web para", "crear un sitio web para" ) if any(k in prompt_lower for k in explicit_new_indicators): intent.workspace_intent = "NEW" else: intent.workspace_intent = "CURRENT" # 7. Check Negation if any(neg in prompt_lower for neg in ('no crees', 'no generes', 'sin crear', 'revisa únicamente', 'revisa unicamente')): intent.project_name = None intent.project_name_source = None return intent # 8. RESOLVE PROJECT NAME ACCORDING TO MANDATORY PRECEDENCE # PRIORIDAD 1: Declaración explícita de proyecto # "crea un proyecto nuevo llamado X", "crea un nuevo proyecto llamado X", "proyecto llamado X", "carpeta nueva llamada X", "workspace llamado X" p1_prefix = ( r'(?i)(?:(?:crea|crear|inicia|iniciar)\s+(?:un\s+)?(?:nuevo\s+proyecto|proyecto\s+nuevo|nueva\s+carpeta|carpeta\s+nueva|nuevo\s+workspace|workspace\s+nuevo)|' r'(?:nuevo\s+proyecto|proyecto\s+nuevo)|' r'(?:nueva\s+carpeta|carpeta\s+nueva)|' r'(?:nuevo\s+workspace|workspace\s+nuevo)|' r'proyecto|carpeta|workspace)\s+(?:llamado|llamada|nombrado|nombrada|titulado|titulada)\s*' ) # 1A: Quoted m_p1_quoted = re.search(p1_prefix + r'[' + q_char + r']([^' + q_char + r']+)[' + q_char + r']', prompt) if m_p1_quoted: cand = m_p1_quoted.group(1).strip() matched_prefix = m_p1_quoted.group(0).lower() if len(cand) >= 2 and cand.lower() not in ('activo', 'actual', 'este', 'nuevo', 'abierto'): intent.project_name = cand intent.project_name_source = "EXPLICIT_WORKSPACE_DECLARATION" if any(w in matched_prefix for w in ('carpeta', 'workspace')) else "EXPLICIT_PROJECT_DECLARATION" intent.workspace_intent = "NEW" return intent # 1B: Unquoted p1_unquoted_pattern = ( p1_prefix + r'([a-zA-Z0-9_\-\u00c0-\u017f]+(?:\s+[a-zA-Z0-9_\-\u00c0-\u017f]+)*?)' + r'(?=\s+(?:y\s+(?:crea|agrega|a[ñn]ade|genera|construye|haz|modifica|dise[ñn]a)|con\s+(?:el\s+)?t[ií]tulo|con\s+t[ií]tulo|con\s+|para\s+(?:una\s+)?(?:marca|tienda|empresa|negocio|cliente)|bot[oó]n|mensaje|texto|t[ií]tulo|titulo)\b|[.,;\r\n]|$)' ) m_p1_unquoted = re.search(p1_unquoted_pattern, prompt) if m_p1_unquoted: cand = m_p1_unquoted.group(1).strip() matched_prefix = m_p1_unquoted.group(0).lower() if len(cand) >= 2 and cand.lower() not in ('activo', 'actual', 'este', 'nuevo', 'abierto'): intent.project_name = cand intent.project_name_source = "EXPLICIT_WORKSPACE_DECLARATION" if any(w in matched_prefix for w in ('carpeta', 'workspace')) else "EXPLICIT_PROJECT_DECLARATION" intent.workspace_intent = "NEW" return intent # PRIORIDAD 2: Nombre explícitamente asociado a proyecto/carpeta/workspace p2_pattern = ( r'(?i)(?:(?:en\s+el\s+|para\s+el\s+|crea\s+el\s+|abrir\s+el\s+)?(?:proyecto|carpeta|workspace))\s+' r'(?:[' + q_char + r']([^' + q_char + r']+)[' + q_char + r']|([a-zA-Z0-9_\-\u00c0-\u017f]+(?:\s+[a-zA-Z0-9_\-\u00c0-\u017f]+)*?))' r'(?=\s+(?:y\s+|con\s+|para\s+|bot[oó]n|mensaje|texto|t[ií]tulo|titulo)\b|[.,;\r\n]|$)' ) m_p2 = re.search(p2_pattern, prompt) if m_p2: cand = (m_p2.group(1) or m_p2.group(2) or "").strip() matched_str = m_p2.group(0).lower() if len(cand) >= 2 and cand.lower() not in ('activo', 'actual', 'este', 'nuevo', 'nueva', 'abierto', 'existente'): intent.project_name = cand intent.project_name_source = "EXPLICIT_WORKSPACE_DECLARATION" if any(w in matched_str for w in ('carpeta', 'workspace')) else "EXPLICIT_PROJECT_DECLARATION" return intent # PRIORIDAD 3: Inferencia contextual SOLO si realmente se está creando un proyecto nuevo y no existe un nombre explícito if intent.workspace_intent == "NEW": m_category_named = re.search( r'(?i)(?:para\s+(?:una\s+|un\s+)?)(panader[íi]a|veterinaria|cl[íi]nica|farmacia|librer[íi]a|tienda|empresa|restaurante)\s+(?:llamada|llamado|de)?\s+([a-zA-Z0-9_\-\u00c0-\u017f]+(?:\s+[a-zA-Z0-9_\-\u00c0-\u017f]+)*?)(?=\s+(?:con|y|para|bot[oó]n|mensaje|texto)\b|[.,;\r\n]|$)', prompt ) if m_category_named: category = m_category_named.group(1).capitalize().replace("í", "i").replace("Í", "I") entity = m_category_named.group(2).strip() cand = f"{category} {entity}" if len(cand) >= 2 and cand.lower() not in ('activo', 'actual', 'este', 'nuevo', 'abierto'): intent.project_name = cand intent.project_name_source = "SAFE_SEMANTIC_INFERENCE" return intent m_named = re.search( r'(?i)(?:para\s+(?:la\s+|el\s+)?)([a-zA-Z0-9_\-\u00c0-\u017f]+(?:\s+[a-zA-Z0-9_\-\u00c0-\u017f]+)*?)(?=\s+(?:con|y|para|bot[oó]n|mensaje|texto)\b|[.,;\r\n]|$)', prompt ) if m_named: cand = m_named.group(1).strip() if len(cand) >= 2 and cand.lower() not in ('activo', 'actual', 'este', 'nuevo', 'abierto'): intent.project_name = cand intent.project_name_source = "SAFE_SEMANTIC_INFERENCE" return intent # PRIORIDAD 4: No inventar nombre a partir de cualquier string citado. return intent def _infer_project_name(self, prompt: str) -> Optional[str]: intent = self.parse_mission_intent(prompt) self._last_project_name_source = intent.project_name_source return intent.project_name def create_mission_from_prompt( self, prompt: str, conv_id: str, active_workspace: Optional[str] = None, explicit_tasks: Optional[List[Dict[str, Any]]] = None, instance_id: Optional[str] = None ) -> Dict[str, Any]: mission_id = f"mission_{int(time.time())}_{uuid.uuid4().hex[:6]}" now = time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime()) current_ws = os.path.abspath(active_workspace or self.workspace_root) if (active_workspace or self.workspace_root) else None intent = self.parse_mission_intent(prompt) inferred = intent.project_name source = intent.project_name_source self._last_project_name_source = source if inferred is not None and intent.workspace_intent == "NEW": ide_root = os.path.abspath(r"C:\OptimusProjects\Optimus Developer") base_projects = os.path.dirname(ide_root) target_workspace = os.path.join(base_projects, inferred) project_name = inferred elif current_ws: target_workspace = current_ws project_name = os.path.basename(current_ws) elif inferred is not None: ide_root = os.path.abspath(r"C:\OptimusProjects\Optimus Developer") base_projects = os.path.dirname(ide_root) target_workspace = os.path.join(base_projects, inferred) project_name = inferred else: ide_root = os.path.abspath(r"C:\OptimusProjects\Optimus Developer") base_projects = os.path.dirname(ide_root) target_workspace = os.path.join(base_projects, "Proyecto Nuevo") project_name = "Proyecto Nuevo" is_user_project = True clean_prompt = prompt.strip().splitlines()[0] title = clean_prompt[:60] + ("..." if len(clean_prompt) > 60 else "") tasks = explicit_tasks if explicit_tasks else self._generate_plan_tasks(prompt, project_name) mission = { "id": mission_id, "conversation_id": conv_id, "title": title, "prompt": prompt, "project_name": project_name, "project_name_source": source, "page_title": intent.page_title, "brand_name": intent.brand_name, "is_user_project": is_user_project, "target_workspace": target_workspace, "status": "in_progress", "created_at": now, "updated_at": now, "tasks": tasks, "completed_tasks": 0, "total_tasks": len(tasks), "percent": 0, "artifacts_created": [] } self._save_mission(mission) self._active_mission_id = mission_id if conv_id: self._active_mission_by_conv[conv_id] = mission_id if instance_id: self._active_mission_by_instance[instance_id] = mission_id self.ledger.record_event(EventLedger.MISSION_CREATED, { "mission_id": mission_id, "title": title, "project_name": project_name, "project_name_source": source, "page_title": intent.page_title, "brand_name": intent.brand_name, "target_workspace": target_workspace, "total_tasks": len(tasks) }, source=EventLedger.SOURCE_AGENT) self.ledger.record_event(EventLedger.PLAN_CREATED, { "mission_id": mission_id, "plan": [t["title"] for t in tasks] }, source=EventLedger.SOURCE_AGENT) return mission def _generate_plan_tasks(self, prompt: str, project_name: Optional[str]) -> List[Dict[str, Any]]: prompt_lower = prompt.lower() is_negated = any(neg in prompt_lower for neg in ("no crees", "no generes", "sin crear", "revisa únicamente", "revisa unicamente")) # Check if an exact file creation is requested (unless negated) if not is_negated: m_file = re.search(r'(?:(?:archivo|fichero|file)(?:\\s+llamado)?\\s+([a-zA-Z0-9_\-\.]+\.[a-zA-Z0-9]+)|(?:crea|create)\\s+([a-zA-Z0-9_\-\.]+\.[a-zA-Z0-9]+))', prompt, re.IGNORECASE) if m_file: fname = (m_file.group(1) or m_file.group(2)).strip() return [ {"id": "task_1", "title": "Analizar requerimientos en Digital Optimus Brain", "status": "pending", "action": "ANALYZE_REQUEST"}, {"id": "task_2", "title": f"Crear y verificar {fname} en workspace activo", "status": "pending", "action": "CREATE_FILE"}, {"id": "task_3", "title": "Verificar SHA-256 físico y registrar evidencia", "status": "pending", "action": "VERIFY_FILES"} ] # Check if bakery website is specifically requested if "panader" in prompt_lower and ("catálogo" in prompt_lower or "catalogo" in prompt_lower): return [ {"id": "task_1", "title": "Crear estructura de proyecto en workspace activo", "status": "pending", "action": "CREATE_STRUCTURE"}, {"id": "task_2", "title": "Crear index.html con estructura semántica, catálogo y carrito", "status": "pending", "action": "CREATE_INDEX"}, {"id": "task_3", "title": "Crear styles.css con diseño responsive y moderno", "status": "pending", "action": "CREATE_STYLES"}, {"id": "task_4", "title": "Crear script.js con catálogo interactivo y checkout a WhatsApp", "status": "pending", "action": "CREATE_SCRIPT"}, {"id": "task_5", "title": "Verificación de archivos y cálculo de hash SHA-256", "status": "pending", "action": "VERIFY_FILES"}, {"id": "task_6", "title": "Validación en Terminal y registro en Event Ledger", "status": "pending", "action": "TERMINAL_VALIDATE"}, {"id": "task_7", "title": "Apertura y verificación de preview web", "status": "pending", "action": "BROWSER_PREVIEW"} ] # Check if python script / hello.py if "hello.py" in prompt_lower or "python" in prompt_lower: return [ {"id": "task_1", "title": "Analizar requerimientos en Digital Optimus Brain", "status": "pending", "action": "ANALYZE_REQUEST"}, {"id": "task_2", "title": "Generar y guardar script en workspace activo", "status": "pending", "action": "CREATE_FILE"}, {"id": "task_3", "title": "Ejecutar y verificar en Terminal", "status": "pending", "action": "TERMINAL_VALIDATE"} ] # Check if explicitly asking for calculator if "calculadora" in prompt_lower: return [ {"id": "task_1", "title": "Analizar requerimientos de calculadora en Digital Optimus Brain", "status": "pending", "action": "ANALYZE_REQUEST"}, {"id": "task_2", "title": "Crear estructura de calculadora", "status": "pending", "action": "CREATE_STRUCTURE"}, {"id": "task_3", "title": "Verificar resultado y registrar evidencia", "status": "pending", "action": "VERIFY_FILES"} ] # General mission / capability execution / analysis return [ {"id": "task_1", "title": "Analizar requerimiento en Digital Optimus Brain", "status": "pending", "action": "ANALYZE_REQUEST"}, {"id": "task_2", "title": "Consultar capacidades del sistema y ejecutar acciones", "status": "pending", "action": "EXECUTE_CAPABILITY"}, {"id": "task_3", "title": "Validar evidencia física y registrar en Event Ledger", "status": "pending", "action": "VERIFY_EVIDENCE"} ] def _save_mission(self, mission: Dict[str, Any]): self._missions_cache[mission["id"]] = mission if not self.storage_dir: return path = self._get_path(mission["id"]) with open(path, "w", encoding="utf-8") as f: json.dump(mission, f, indent=2, ensure_ascii=False) def get_mission(self, mission_id: str) -> Optional[Dict[str, Any]]: if mission_id in self._missions_cache: return self._missions_cache[mission_id] if not self.storage_dir: return None path = self._get_path(mission_id) if os.path.exists(path): try: with open(path, "r", encoding="utf-8") as f: return json.load(f) except Exception: pass return None def get_active_mission(self, conv_id: Optional[str] = None, instance_id: Optional[str] = None) -> Optional[Dict[str, Any]]: # Strictly scoped: filter by conv_id or instance_id and status in_progress. Never mtime newest! if conv_id: m_id = self._active_mission_by_conv.get(conv_id) if m_id: m = self.get_mission(m_id) if m and m.get("status") == "in_progress": return m return None if instance_id: m_id = self._active_mission_by_instance.get(instance_id) if m_id: m = self.get_mission(m_id) if m and m.get("status") == "in_progress": return m return None if self._active_mission_id: m = self.get_mission(self._active_mission_id) if m and m.get("status") == "in_progress": return m return None def update_task(self, mission_id: str, task_id: str, status: str, details: Optional[Dict[str, Any]] = None) -> Optional[Dict[str, Any]]: mission = self.get_mission(mission_id) if not mission: return None for t in mission.get("tasks", []): if t["id"] == task_id: t["status"] = status if details: t["details"] = details break completed = sum(1 for t in mission.get("tasks", []) if t.get("status") == "completed") total = len(mission.get("tasks", [])) mission["completed_tasks"] = completed mission["percent"] = int((completed / total) * 100) if total > 0 else 0 mission["updated_at"] = time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime()) self._save_mission(mission) event_type = EventLedger.TASK_COMPLETED if status == "completed" else (EventLedger.TASK_FAILED if status == "failed" else "TASK_UPDATED") self.ledger.record_event(event_type, { "mission_id": mission_id, "task_id": task_id, "status": status, "completed": completed, "total": total }, source=EventLedger.SOURCE_AGENT) return mission def update_mission_status(self, mission_id: str, status: str) -> Optional[Dict[str, Any]]: mission = self.get_mission(mission_id) if not mission: return None mission["status"] = status mission["updated_at"] = time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime()) if status in ("completed", "done"): mission["percent"] = 100 for t in mission.get("tasks", []): t["status"] = "completed" self._save_mission(mission) return mission