diff --git a/agent/error_handler.py b/agent/error_handler.py deleted file mode 100644 index c122068..0000000 --- a/agent/error_handler.py +++ /dev/null @@ -1,149 +0,0 @@ -""" -MARK XL — Error Handler -Replaces google.generativeai with local Ollama via core.llm_client. -""" -import json -import re -import sys -from enum import Enum -from pathlib import Path - -from core.llm_client import call_llm_text - - -def get_base_dir() -> Path: - if getattr(sys, "frozen", False): - return Path(sys.executable).parent - return Path(__file__).resolve().parent.parent - - -BASE_DIR = get_base_dir() - - -class ErrorDecision(Enum): - RETRY = "retry" - SKIP = "skip" - REPLAN = "replan" - ABORT = "abort" - - -ERROR_ANALYST_PROMPT = """You are the error recovery module of MARK XL AI assistant. - -A task step has failed. Analyze the error and decide what to do. - -DECISIONS: -- retry : Transient error (network timeout, temporary file lock, race condition). -- skip : This step is not critical and the task can succeed without it. -- replan : The approach was wrong. A different tool or method should be tried. -- abort : The task is fundamentally impossible or unsafe to continue. - -Return ONLY valid JSON: -{ - "decision": "retry|skip|replan|abort", - "reason": "why it failed", - "fix_suggestion": "what to try instead (for replan)", - "max_retries": 1, - "user_message": "Short message to tell the user (max 15 words)" -} -""" - - -def analyze_error( - step: dict, - error: str, - attempt: int = 1, - max_attempts: int = 2, -) -> dict: - if attempt >= max_attempts: - print(f"[ErrorHandler] ⚠️ Max attempts for step {step.get('step')} — forcing replan") - return { - "decision": ErrorDecision.REPLAN, - "reason": f"Failed {attempt} times: {error[:100]}", - "fix_suggestion": "Try a completely different approach or tool", - "max_retries": 0, - "user_message": "Trying a different approach, sir.", - } - - prompt = f"""Failed step: -Tool: {step.get('tool')} -Description: {step.get('description')} -Parameters: {json.dumps(step.get('parameters', {}), indent=2)} -Critical: {step.get('critical', False)} - -Error: -{error[:500]} - -Attempt number: {attempt}""" - - try: - text = call_llm_text(prompt, system=ERROR_ANALYST_PROMPT) - text = re.sub(r"```(?:json)?", "", text).strip().rstrip("`").strip() - result = json.loads(text) - - decision_str = result.get("decision", "replan").lower() - decision_map = { - "retry": ErrorDecision.RETRY, - "skip": ErrorDecision.SKIP, - "replan": ErrorDecision.REPLAN, - "abort": ErrorDecision.ABORT, - } - result["decision"] = decision_map.get(decision_str, ErrorDecision.REPLAN) - - if step.get("critical") and result["decision"] == ErrorDecision.SKIP: - result["decision"] = ErrorDecision.REPLAN - result["user_message"] = "This step is critical — finding alternative approach, sir." - - print(f"[ErrorHandler] Decision: {result['decision'].value} — {result.get('reason', '')}") - return result - - except Exception as e: - print(f"[ErrorHandler] ⚠️ Analysis failed: {e} — defaulting to replan") - return { - "decision": ErrorDecision.REPLAN, - "reason": str(e), - "fix_suggestion": "Try alternative approach", - "max_retries": 1, - "user_message": "Encountered an issue, adjusting approach, sir.", - } - - -def generate_fix(step: dict, error: str, fix_suggestion: str) -> dict: - prompt = f"""A task step failed. Generate a replacement step. - -Original step: -Tool: {step.get('tool')} -Description: {step.get('description')} -Parameters: {json.dumps(step.get('parameters', {}), indent=2)} - -Error: {error[:300]} -Fix suggestion: {fix_suggestion} - -Write a Python script that accomplishes the same goal differently. -Return ONLY the Python code, no explanation.""" - - try: - code = call_llm_text(prompt) - code = re.sub(r"```(?:python)?", "", code).strip().rstrip("`").strip() - return { - "step": step.get("step"), - "tool": "code_helper", - "description": f"Auto-fix for: {step.get('description')}", - "parameters": { - "action": "run", - "description": fix_suggestion, - "code": code, - "language": "python", - }, - "depends_on": step.get("depends_on", []), - "critical": step.get("critical", False), - } - except Exception as e: - print(f"[ErrorHandler] ⚠️ Fix generation failed: {e}") - return { - "step": step.get("step"), - "tool": "generated_code", - "description": f"Fallback for: {step.get('description')}", - "parameters": {"description": step.get("description", "")}, - "depends_on": step.get("depends_on", []), - "critical": step.get("critical", False), - } diff --git a/agent/executor.py b/agent/executor.py deleted file mode 100644 index b2dcfd9..0000000 --- a/agent/executor.py +++ /dev/null @@ -1,394 +0,0 @@ -""" -MARK XL — Agent Executor -Replaces google.generativeai with local Ollama via core.llm_client. -""" -import json -import re -import sys -import threading -import subprocess -import tempfile -import os -from pathlib import Path -from typing import Callable - -from agent.planner import create_plan, replan -from agent.error_handler import analyze_error, generate_fix, ErrorDecision -from core.llm_client import call_llm_text - - -def get_base_dir() -> Path: - if getattr(sys, "frozen", False): - return Path(sys.executable).parent - return Path(__file__).resolve().parent.parent - - -BASE_DIR = get_base_dir() - - -# --------------------------------------------------------------------------- -# Code generation helper (replaces _run_generated_code with Gemini) -# --------------------------------------------------------------------------- - -def _run_generated_code(description: str, speak: Callable | None = None) -> str: - if speak: - speak("Writing custom code for this task, sir.") - - home = Path.home() - desktop = home / "Desktop" - downloads = home / "Downloads" - documents = home / "Documents" - - if not desktop.exists(): - try: - import winreg - key = winreg.OpenKey( - winreg.HKEY_CURRENT_USER, - r"Software\Microsoft\Windows\CurrentVersion\Explorer\Shell Folders", - ) - desktop = Path(winreg.QueryValueEx(key, "Desktop")[0]) - except Exception: - pass - - system = ( - "You are an expert Python developer. " - "Write clean, complete, working Python code. " - "Use standard library + common packages. " - "Install missing packages with subprocess + pip if needed. " - "Return ONLY the Python code. No explanation, no markdown, no backticks.\n\n" - f"SYSTEM PATHS:\n" - f" Desktop = r'{desktop}'\n" - f" Downloads = r'{downloads}'\n" - f" Documents = r'{documents}'\n" - f" Home = r'{home}'\n" - ) - prompt = f"Write Python code to accomplish this task:\n\n{description}" - - try: - code = call_llm_text(prompt, system=system) - code = re.sub(r"```(?:python)?", "", code).strip().rstrip("`").strip() - - with tempfile.NamedTemporaryFile( - mode="w", suffix=".py", delete=False, encoding="utf-8" - ) as f: - f.write(code) - tmp_path = f.name - - print(f"[Executor] 🐍 Running generated code: {tmp_path}") - - result = subprocess.run( - [sys.executable, tmp_path], - capture_output=True, text=True, - timeout=120, cwd=str(Path.home()), - ) - - try: - os.unlink(tmp_path) - except Exception: - pass - - output = result.stdout.strip() - error = result.stderr.strip() - - if result.returncode == 0 and output: - return output - elif result.returncode == 0: - return "Task completed successfully." - elif error: - raise RuntimeError(f"Code error: {error[:400]}") - return "Completed." - - except subprocess.TimeoutExpired: - raise RuntimeError("Generated code timed out after 120 seconds.") - except RuntimeError: - raise - except Exception as e: - raise RuntimeError(f"Generated code failed: {e}") - - -# --------------------------------------------------------------------------- -# Context injection -# --------------------------------------------------------------------------- - -def _detect_language(text: str) -> str: - try: - return call_llm_text( - f"What language is this text written in? " - f"Reply with ONLY the language name in English (e.g. Turkish, English, French).\n\n" - f"Text: {text[:200]}" - ).strip() - except Exception: - return "English" - - -def _translate_to_goal_language(content: str, goal: str) -> str: - if not goal: - return content - try: - target_lang = _detect_language(goal) - print(f"[Executor] 🌐 Translating to: {target_lang}") - prompt = ( - f"You are a professional translator. " - f"Translate the following text into {target_lang}.\n" - f"IMPORTANT:\n" - f"- Translate EVERYTHING, leave nothing in English\n" - f"- Keep all facts, numbers, and data intact\n" - f"- Keep the structure and formatting\n" - f"- Output ONLY the translated text, nothing else\n\n" - f"Text to translate:\n{content[:4000]}" - ) - translated = call_llm_text(prompt) - print(f"[Executor] ✅ Translation done ({target_lang})") - return translated - except Exception as e: - print(f"[Executor] ⚠️ Translation failed: {e}") - return content - - -def _inject_context(params: dict, tool: str, step_results: dict, goal: str = "") -> dict: - if not step_results: - return params - params = dict(params) - if tool == "file_controller" and params.get("action") in ("write", "create_file"): - content = params.get("content", "") - if not content or len(content) < 50: - all_results = [ - v for v in step_results.values() - if v and len(v) > 100 and v not in ("Done.", "Completed.") - ] - if all_results: - combined = "\n\n---\n\n".join(all_results) - translated = _translate_to_goal_language(combined, goal) - params["content"] = translated - print("[Executor] 💉 Injected + translated content") - return params - - -# --------------------------------------------------------------------------- -# Tool routing -# --------------------------------------------------------------------------- - -def _call_tool(tool: str, parameters: dict, speak: Callable | None) -> str: - if tool == "open_app": - from actions.open_app import open_app - return open_app(parameters=parameters, player=None) or "Done." - - elif tool == "web_search": - from actions.web_search import web_search - return web_search(parameters=parameters, player=None) or "Done." - - elif tool == "game_updater": - from actions.game_updater import game_updater - return game_updater(parameters=parameters, player=None, speak=speak) or "Done." - - elif tool == "browser_control": - from actions.browser_control import browser_control - return browser_control(parameters=parameters, player=None) or "Done." - - elif tool == "file_controller": - from actions.file_controller import file_controller - return file_controller(parameters=parameters, player=None) or "Done." - - elif tool == "code_helper": - from actions.code_helper import code_helper - return code_helper(parameters=parameters, player=None, speak=speak) or "Done." - - elif tool == "dev_agent": - from actions.dev_agent import dev_agent - return dev_agent(parameters=parameters, player=None, speak=speak) or "Done." - - elif tool == "screen_process": - from actions.screen_processor import screen_process - result = screen_process(parameters=parameters, player=None) - return result if isinstance(result, str) else "Screen captured and analyzed." - - elif tool == "send_message": - from actions.send_message import send_message - return send_message(parameters=parameters, player=None) or "Done." - - elif tool == "reminder": - from actions.reminder import reminder - return reminder(parameters=parameters, player=None) or "Done." - - elif tool == "youtube_video": - from actions.youtube_video import youtube_video - return youtube_video(parameters=parameters, player=None) or "Done." - - elif tool == "weather_report": - from actions.weather_report import weather_action - return weather_action(parameters=parameters, player=None) or "Done." - - elif tool == "computer_settings": - from actions.computer_settings import computer_settings - return computer_settings(parameters=parameters, player=None) or "Done." - - elif tool == "desktop_control": - from actions.desktop import desktop_control - return desktop_control(parameters=parameters, player=None) or "Done." - - elif tool == "computer_control": - from actions.computer_control import computer_control - return computer_control(parameters=parameters, player=None) or "Done." - - elif tool == "generated_code": - description = parameters.get("description", "") - if not description: - raise ValueError("generated_code requires a 'description' parameter.") - return _run_generated_code(description, speak=speak) - - elif tool == "flight_finder": - from actions.flight_finder import flight_finder - return flight_finder(parameters=parameters, player=None, speak=speak) or "Done." - - else: - print(f"[Executor] ⚠️ Unknown tool '{tool}' — falling back to generated_code") - return _run_generated_code(f"Accomplish this task: {parameters}", speak=speak) - - -# --------------------------------------------------------------------------- -# AgentExecutor -# --------------------------------------------------------------------------- - -class AgentExecutor: - - MAX_REPLAN_ATTEMPTS = 2 - - def execute( - self, - goal: str, - speak: Callable | None = None, - cancel_flag: threading.Event | None = None, - ) -> str: - print(f"\n[Executor] 🎯 Goal: {goal}") - - replan_attempts = 0 - completed_steps: list = [] - step_results: dict = {} - plan = create_plan(goal) - - while True: - steps = plan.get("steps", []) - if not steps: - msg = "I couldn't create a valid plan for this task, sir." - if speak: speak(msg) - return msg - - success = True - failed_step = None - failed_error = "" - - for step in steps: - if cancel_flag and cancel_flag.is_set(): - if speak: speak("Task cancelled, sir.") - return "Task cancelled." - - step_num = step.get("step", "?") - tool = step.get("tool", "generated_code") - desc = step.get("description", "") - params = step.get("parameters", {}) - params = _inject_context(params, tool, step_results, goal=goal) - - print(f"\n[Executor] ▶️ Step {step_num}: [{tool}] {desc}") - - attempt = 1 - step_ok = False - - while attempt <= 3: - if cancel_flag and cancel_flag.is_set(): - break - try: - result = _call_tool(tool, params, speak) - step_results[step_num] = result - completed_steps.append(step) - print(f"[Executor] ✅ Step {step_num} done: {str(result)[:100]}") - step_ok = True - break - - except Exception as e: - error_msg = str(e) - print(f"[Executor] ❌ Step {step_num} attempt {attempt} failed: {error_msg}") - - recovery = analyze_error(step, error_msg, attempt=attempt) - decision = recovery["decision"] - user_msg = recovery.get("user_message", "") - - if speak and user_msg: - speak(user_msg) - - if decision == ErrorDecision.RETRY: - attempt += 1 - import time; time.sleep(2) - continue - - elif decision == ErrorDecision.SKIP: - print(f"[Executor] ⏭️ Skipping step {step_num}") - completed_steps.append(step) - step_ok = True - break - - elif decision == ErrorDecision.ABORT: - msg = f"Task aborted, sir. {recovery.get('reason', '')}" - if speak: speak(msg) - return msg - - else: # REPLAN - fix_suggestion = recovery.get("fix_suggestion", "") - if fix_suggestion and tool != "generated_code": - try: - fixed_step = generate_fix(step, error_msg, fix_suggestion) - if speak: speak("Trying an alternative approach, sir.") - res = _call_tool( - fixed_step["tool"], - fixed_step["parameters"], - speak, - ) - step_results[step_num] = res - completed_steps.append(step) - step_ok = True - break - except Exception as fix_err: - print(f"[Executor] ⚠️ Fix failed: {fix_err}") - - failed_step = step - failed_error = error_msg - success = False - break - - if not step_ok and not failed_step: - failed_step = step - failed_error = "Max retries exceeded" - success = False - - if not success: - break - - if success: - return self._summarize(goal, completed_steps, speak) - - if replan_attempts >= self.MAX_REPLAN_ATTEMPTS: - msg = f"Task failed after {replan_attempts} replan attempts, sir." - if speak: speak(msg) - return msg - - if speak: speak("Adjusting my approach, sir.") - replan_attempts += 1 - plan = replan(goal, completed_steps, failed_step, failed_error) - - def _summarize(self, goal: str, completed_steps: list, speak: Callable | None) -> str: - fallback = f"All done, sir. Completed {len(completed_steps)} steps for: {goal[:60]}." - steps_str = "\n".join(f"- {s.get('description', '')}" for s in completed_steps) - prompt = ( - f'User goal: "{goal}"\n' - f"Completed steps:\n{steps_str}\n\n" - "Write a single natural sentence summarising what was accomplished. " - "Address the user as 'sir'. Be direct and positive." - ) - try: - summary = call_llm_text(prompt) - if summary: - if speak: speak(summary) - return summary - except Exception: - pass - if speak: speak(fallback) - return fallback diff --git a/agent/planner.py b/agent/planner.py deleted file mode 100644 index 5613c13..0000000 --- a/agent/planner.py +++ /dev/null @@ -1,211 +0,0 @@ -""" -MARK XL — Task Planner -Replaces google.generativeai with local Ollama via core.llm_client. -""" -import json -import re -import sys -from pathlib import Path - -from core.llm_client import call_llm_text - - -def get_base_dir() -> Path: - if getattr(sys, "frozen", False): - return Path(sys.executable).parent - return Path(__file__).resolve().parent.parent - - -BASE_DIR = get_base_dir() - - -PLANNER_PROMPT = """You are the planning module of MARK XL, a personal AI assistant. -Your job: break any user goal into a sequence of steps using ONLY the tools listed below. - -ABSOLUTE RULES: -- NEVER use generated_code or write Python scripts. It does not exist. -- NEVER reference previous step results in parameters. Every step is independent. -- Use web_search for ANY information retrieval, research, or current data. -- Use file_controller to save content to disk. -- Max 5 steps. Use the minimum steps needed. - -AVAILABLE TOOLS AND THEIR PARAMETERS: - -open_app - app_name: string (required) - -web_search - query: string (required) — write a clear, focused search query - mode: "search" or "compare" (optional, default: search) - items: list of strings (optional, for compare mode) - aspect: string (optional, for compare mode) - -game_updater - action: "update" | "install" | "list" | "download_status" | "schedule" (required) - platform: "steam" | "epic" | "both" (optional, default: both) - game_name: string (optional) - app_id: string (optional) - shutdown_when_done: boolean (optional) - -browser_control - action: "go_to" | "search" | "click" | "type" | "scroll" | "get_text" | "press" | "close" (required) - url: string (for go_to) - query: string (for search) - text: string (for click/type) - direction: "up" | "down" (for scroll) - -file_controller - action: "write" | "create_file" | "read" | "list" | "delete" | "move" | "copy" | "find" | "disk_usage" (required) - path: string — use "desktop" for Desktop folder - name: string — filename - content: string — file content (for write/create_file) - -computer_settings - action: string (required) - description: string — natural language description - value: string (optional) - -computer_control - action: "type" | "click" | "hotkey" | "press" | "scroll" | "screenshot" | "screen_find" | "screen_click" (required) - text: string (for type) - x, y: int (for click) - keys: string (for hotkey, e.g. "ctrl+c") - key: string (for press) - direction: "up" | "down" (for scroll) - description: string (for screen_find/screen_click) - -screen_process - text: string (required) — what to analyze or ask about the screen - angle: "screen" | "camera" (optional) - -send_message - receiver: string (required) - message_text: string (required) - platform: string (required) - -reminder - date: string YYYY-MM-DD (required) - time: string HH:MM (required) - message: string (required) - -desktop_control - action: "wallpaper" | "organize" | "clean" | "list" | "task" (required) - path: string (optional) - task: string (optional) - -youtube_video - action: "play" | "summarize" | "trending" (required) - query: string (for play) - -weather_report - city: string (required) - -flight_finder - origin: string (required) - destination: string (required) - date: string (required) - -code_helper - action: "write" | "edit" | "run" | "explain" (required) - description: string (required) - language: string (optional) - output_path: string (optional) - file_path: string (optional) - -dev_agent - description: string (required) - language: string (optional) - -OUTPUT — return ONLY valid JSON, no markdown, no explanation, no code blocks: -{ - "goal": "...", - "steps": [ - { - "step": 1, - "tool": "tool_name", - "description": "what this step does", - "parameters": {}, - "critical": true - } - ] -} -""" - - -def create_plan(goal: str, context: str = "") -> dict: - user_input = f"Goal: {goal}" - if context: - user_input += f"\n\nContext: {context}" - - try: - text = call_llm_text(user_input, system=PLANNER_PROMPT) - text = re.sub(r"```(?:json)?", "", text).strip().rstrip("`").strip() - - plan = json.loads(text) - if "steps" not in plan or not isinstance(plan["steps"], list): - raise ValueError("Invalid plan structure") - - for step in plan["steps"]: - if step.get("tool") == "generated_code": - print(f"[Planner] ⚠️ generated_code in step {step.get('step')} — replacing with web_search") - step["tool"] = "web_search" - step["parameters"] = {"query": step.get("description", goal)[:200]} - - print(f"[Planner] ✅ Plan: {len(plan['steps'])} steps") - for s in plan["steps"]: - print(f" Step {s['step']}: [{s['tool']}] {s['description']}") - return plan - - except json.JSONDecodeError as e: - print(f"[Planner] ⚠️ JSON parse failed: {e}") - return _fallback_plan(goal) - except Exception as e: - print(f"[Planner] ⚠️ Planning failed: {e}") - return _fallback_plan(goal) - - -def _fallback_plan(goal: str) -> dict: - print("[Planner] 🔄 Fallback plan") - return { - "goal": goal, - "steps": [ - { - "step": 1, - "tool": "web_search", - "description": f"Search for: {goal}", - "parameters": {"query": goal}, - "critical": True, - } - ], - } - - -def replan(goal: str, completed_steps: list, failed_step: dict, error: str) -> dict: - completed_summary = "\n".join( - f" - Step {s['step']} ({s['tool']}): DONE" for s in completed_steps - ) - prompt = f"""Goal: {goal} - -Already completed: -{completed_summary if completed_summary else ' (none)'} - -Failed step: [{failed_step.get('tool')}] {failed_step.get('description')} -Error: {error} - -Create a REVISED plan for the remaining work only. Do not repeat completed steps.""" - - try: - text = call_llm_text(prompt, system=PLANNER_PROMPT) - text = re.sub(r"```(?:json)?", "", text).strip().rstrip("`").strip() - plan = json.loads(text) - - for step in plan.get("steps", []): - if step.get("tool") == "generated_code": - step["tool"] = "web_search" - step["parameters"] = {"query": step.get("description", goal)[:200]} - - print(f"[Planner] 🔄 Revised plan: {len(plan['steps'])} steps") - return plan - except Exception as e: - print(f"[Planner] ⚠️ Replan failed: {e}") - return _fallback_plan(goal) diff --git a/agent/task_queue.py b/agent/task_queue.py deleted file mode 100644 index 90c0e84..0000000 --- a/agent/task_queue.py +++ /dev/null @@ -1,221 +0,0 @@ -import threading -import time -import uuid -from dataclasses import dataclass, field -from enum import Enum -from typing import Callable, Any - - -class TaskStatus(Enum): - PENDING = "pending" - RUNNING = "running" - COMPLETED = "completed" - FAILED = "failed" - CANCELLED = "cancelled" - - -class TaskPriority(Enum): - LOW = 3 - NORMAL = 2 - HIGH = 1 - - -@dataclass(order=True) -class Task: - priority: int - created_at: float = field(compare=False) - task_id: str = field(compare=False) - goal: str = field(compare=False) - status: TaskStatus = field(compare=False, default=TaskStatus.PENDING) - result: Any = field(compare=False, default=None) - error: str = field(compare=False, default="") - speak: Any = field(compare=False, default=None) - on_complete: Any = field(compare=False, default=None) - cancel_flag: threading.Event = field(compare=False, default_factory=threading.Event) - - -class TaskQueue: - def __init__(self, max_concurrent: int = 1): - self._queue: list[Task] = [] - self._lock: threading.Lock = threading.Lock() - self._condition: threading.Condition = threading.Condition(self._lock) - self._tasks: dict[str, Task] = {} - self._running: bool = False - self._worker_thread: threading.Thread | None = None - self._max_concurrent = max_concurrent - self._active_count = 0 - self._executor = None - - def _get_executor(self): - if self._executor is None: - from agent.executor import AgentExecutor - self._executor = AgentExecutor() - return self._executor - - def start(self) -> None: - if self._running: - return - self._running = True - self._worker_thread = threading.Thread( - target=self._worker_loop, - daemon=True, - name="AgentTaskQueue" - ) - self._worker_thread.start() - print("[TaskQueue] ✅ Started") - - def stop(self) -> None: - self._running = False - with self._condition: - self._condition.notify_all() - print("[TaskQueue] 🔴 Stopped") - - def submit( - self, - goal: str, - priority: TaskPriority = TaskPriority.NORMAL, - speak: Callable | None = None, - on_complete: Callable | None = None, - ) -> str: - - task_id = str(uuid.uuid4())[:8] - task = Task( - priority = priority.value, - created_at = time.time(), - task_id = task_id, - goal = goal, - speak = speak, - on_complete = on_complete, - ) - - with self._condition: - self._queue.append(task) - self._queue.sort(key=lambda t: (t.priority, t.created_at)) - self._tasks[task_id] = task - self._condition.notify() - - print(f"[TaskQueue] 📥 Task queued: [{task_id}] {goal[:60]}") - return task_id - - def cancel(self, task_id: str) -> bool: - - with self._lock: - task = self._tasks.get(task_id) - if not task: - return False - if task.status in (TaskStatus.COMPLETED, TaskStatus.FAILED, TaskStatus.CANCELLED): - return False - - task.cancel_flag.set() - task.status = TaskStatus.CANCELLED - print(f"[TaskQueue] 🚫 Task cancelled: [{task_id}]") - return True - - def get_status(self, task_id: str) -> dict | None: - with self._lock: - task = self._tasks.get(task_id) - if not task: - return None - return { - "task_id": task.task_id, - "goal": task.goal, - "status": task.status.value, - "result": task.result, - "error": task.error, - } - - def get_all_statuses(self) -> list[dict]: - with self._lock: - return [ - { - "task_id": t.task_id, - "goal": t.goal[:50], - "status": t.status.value, - } - for t in self._tasks.values() - ] - - def pending_count(self) -> int: - with self._lock: - return sum(1 for t in self._queue if t.status == TaskStatus.PENDING) - - def _worker_loop(self) -> None: - while self._running: - task = None - - with self._condition: - while self._running and not self._next_task(): - self._condition.wait(timeout=1.0) - task = self._next_task() - if task: - task.status = TaskStatus.RUNNING - self._active_count += 1 - try: - self._queue.remove(task) - except ValueError: - pass - - if task: - threading.Thread( - target=self._run_task, - args=(task,), - daemon=True, - name=f"AgentTask-{task.task_id}" - ).start() - - def _next_task(self) -> Task | None: - if self._active_count >= self._max_concurrent: - return None - for task in self._queue: - if task.status == TaskStatus.PENDING and not task.cancel_flag.is_set(): - return task - return None - - def _run_task(self, task: Task) -> None: - print(f"[TaskQueue] ▶️ Running: [{task.task_id}] {task.goal[:60]}") - try: - executor = self._get_executor() - result = executor.execute( - goal = task.goal, - speak = task.speak, - cancel_flag = task.cancel_flag, - ) - - with self._lock: - if task.cancel_flag.is_set(): - task.status = TaskStatus.CANCELLED - else: - task.status = TaskStatus.COMPLETED - task.result = result - self._active_count -= 1 - - if task.on_complete and not task.cancel_flag.is_set(): - try: - task.on_complete(task.task_id, result) - except Exception as e: - print(f"[TaskQueue] ⚠️ on_complete callback error: {e}") - - print(f"[TaskQueue] ✅ Completed: [{task.task_id}]") - - except Exception as e: - with self._lock: - task.status = TaskStatus.FAILED - task.error = str(e) - self._active_count -= 1 - print(f"[TaskQueue] ❌ Failed: [{task.task_id}] {e}") - - with self._condition: - self._condition.notify() - -_queue = TaskQueue() -_queue_started = False -_queue_lock = threading.Lock() - - -def get_queue() -> TaskQueue: - global _queue_started - with _queue_lock: - if not _queue_started: - _queue.start() - _queue_started = True - return _queue \ No newline at end of file