"""
Magi Bot — Multi-Bot Workspace & Messaging Subsystem
====================================================
Features:
1. Multi-bot manager (up to 5 bots), each with an isolated workspace folder named Magi(bot_name).
Dynamically resolved across any PC/OS without hardcoded paths.
2. Model requirement: Only local Ollama models >= 7B parameters are allowed.
3. Natural conversation + Actions: Bots write complete, direct messages to the user and can also
perform workspace actions (create, edit, delete, move, execute files).
4. Custom bot configuration: Name, Model, Profile Color, and Custom System Prompt
(separated from internal tool command prompts).
5. User confirmation popup only when a bot wants to move a file outside its Magi(name) workspace.
6. Step-by-step chaining: Exactly ONE action per message; Granite 4.1:3b analyzes the task
and previous action in the background to formulate and trigger the next step.
7. iMessage-style messaging UI with left contact bar and right chat view.
8. Strict emoji-free styling and messaging.
"""
import base64
import datetime
import html
import json
import os
import random
import re
import shutil
import ssl
import subprocess
import sys
import threading
import time
import urllib.parse
import urllib.request
import uuid
import xml.etree.ElementTree as ET
import requests
from PyQt6.QtWidgets import (
QWidget, QVBoxLayout, QHBoxLayout, QLabel, QLineEdit, QTextEdit,
QPushButton, QFrame, QScrollArea, QDialog, QMessageBox,
QSizePolicy, QSplitter, QComboBox, QFileDialog, QTabWidget, QSpinBox
)
from PyQt6.QtCore import Qt, pyqtSignal, QObject, QThread, QTimer, QSize, QEventLoop
from PyQt6.QtGui import QCursor, QFont, QPixmap, QColor, QPainter, QBrush, QPen
from formatter import format_markdown_and_latex
OLLAMA_BASE_URL = "http://localhost:11434"
GRANITE_DECISION_MODEL = "granite4.1:3b"
MIN_BOT_MODEL_BILLIONS = 8.0
MAX_BOTS = 5
TEAM_COORDINATOR_MODEL = "lfm2.5:8b"
def find_installed_lfm_model() -> str:
"""
Finds installed LFM2.5 model in Ollama, fallback to lfm2.5:8b.
"""
try:
resp = requests.get(f"{OLLAMA_BASE_URL}/api/tags", timeout=3)
if resp.status_code == 200:
models = resp.json().get("models", [])
for m in models:
name = m.get("name", "")
if "lfm" in name.lower() and "2.5" in name.lower():
return name
except Exception:
pass
return "lfm2.5:8b"
def get_solo_coordinator_model() -> str:
"""
Solo bot coordinator always uses granite4.1:3b.
"""
return GRANITE_DECISION_MODEL
def get_team_coordinator_model() -> str:
"""
Returns the team coordinator backend model (LFM2.5:8b).
"""
return find_installed_lfm_model()
def get_active_coordinator_model() -> str:
"""
Backwards compatibility alias for team coordinator model.
"""
return get_team_coordinator_model()
DEFAULT_PALETTE = [
"#007aff", # Apple Blue
"#5856d6", # Indigo
"#af52de", # Purple
"#30b0c7", # Teal
"#34c759", # Green
"#ff9500", # Orange
"#ff3b30", # Red
"#64748b", # Slate
]
# ============================================================================
MIN_BOT_MODEL_BILLIONS = 7.0
def parse_parameter_size_billions(param_str: str, name: str) -> float:
"""
Parses parameter size into billions of parameters (float).
Examples: '7.6B' -> 7.6, '8.2B' -> 8.2, '14.8B' -> 14.8, '354.48M' -> 0.354, 'smollm2:360m' -> 0.36
"""
if param_str:
p = param_str.strip().upper()
if p.endswith("B"):
try:
return float(p[:-1])
except ValueError:
pass
elif p.endswith("M"):
try:
return float(p[:-1]) / 1000.0
except ValueError:
pass
match = re.search(r'(\d+(?:\.\d+)?)\s*([bmBM])', name)
if match:
val = float(match.group(1))
unit = match.group(2).upper()
return val if unit == "B" else val / 1000.0
return 0.0
def get_eligible_bot_models() -> list[str]:
"""
Returns only local Ollama models >= 7.0 Billion parameters.
Filters out models under 7B and guardian safety models.
"""
eligible = []
try:
resp = requests.get(f"{OLLAMA_BASE_URL}/api/tags", timeout=3)
if resp.status_code == 200:
models = resp.json().get("models", [])
# Priority sorting: put qwen2.5-coder:7b first if installed
for m in models:
name = m.get("name", "")
lower = name.lower()
if "guardian" in lower:
continue
param_str = m.get("details", {}).get("parameter_size", "")
size_b = parse_parameter_size_billions(param_str, name)
if size_b >= MIN_BOT_MODEL_BILLIONS:
if name not in eligible:
if "qwen2.5-coder" in lower:
eligible.insert(2, name)
else:
eligible.append(name)
except Exception as e:
print(f"[MagiBot] Error fetching Ollama models: {e}")
# Fallback to standard >= 7B models if no local found
if len(eligible) <= 2:
for fallback in ["qwen2.5-coder:7b", "deepseek-r1:8b", "lfm2.5:8b"]:
if fallback not in eligible:
eligible.append(fallback)
return eligible
# ============================================================================
# 2. BOT WORKSPACE & STORAGE MANAGER
# ============================================================================
def get_base_bots_dir() -> str:
"""
Returns the root directory where all Magi bot workspaces reside.
Dynamic across Windows, macOS, and Linux without hardcoding.
"""
base = os.path.join(os.path.expanduser("~"), "Magi_Bots")
os.makedirs(base, exist_ok=True)
return base
def get_bot_workspace_dir(bot_name: str) -> str:
"""
Returns the dynamic workspace folder named Magi(bot_name).
Ensures directory exists.
"""
safe_name = "".join(c for c in bot_name if c.isalnum() or c in (" ", "_", "-")).strip()
if not safe_name:
safe_name = "Bot"
folder_name = f"Magi({safe_name})"
bot_dir = os.path.join(get_base_bots_dir(), folder_name)
os.makedirs(bot_dir, exist_ok=True)
return bot_dir
class MagiBotManager:
"""
Manages up to 5 bots, their configurations, custom prompts, and message history.
"""
CONFIG_FILE = os.path.join(get_base_bots_dir(), "bots_config.json")
def __init__(self):
self.bots: list[dict] = []
self.load()
def load(self):
if os.path.exists(self.CONFIG_FILE):
try:
with open(self.CONFIG_FILE, "r", encoding="utf-8") as f:
self.bots = json.load(f)
except Exception as e:
print(f"[MagiBotManager] Error loading config: {e}")
self.bots = []
if not self.bots:
# Create 2 default starter bots if empty
self.create_default_bots()
else:
# Upgrade existing bots with < 7B models to eligible >= 7B models
eligible = get_eligible_bot_models()
for b in self.bots:
cur_model = b.get("model", "")
if cur_model not in eligible:
b["model"] = eligible[2] if len(eligible) > 2 else (eligible[0] if eligible else "qwen2.5-coder:7b")
self.save()
def save(self):
try:
with open(self.CONFIG_FILE, "w", encoding="utf-8") as f:
json.dump(self.bots, f, ensure_ascii=False, indent=2)
except Exception as e:
print(f"[MagiBotManager] Error saving config: {e}")
def create_default_bots(self):
self.bots = [
{
"id": str(uuid.uuid4()),
"name": "DevBot",
"model": "qwen2.5-coder:7b",
"color": "#007aff",
"custom_system_prompt": "You are DevBot, an expert software engineer. You discuss code clearly, answer questions, and automate development tasks step by step in your workspace.",
"created_at": time.time(),
"messages": []
},
{
"id": str(uuid.uuid4()),
"name": "Assistant",
"model": "deepseek-r1:8b",
"color": "#5856d6",
"custom_system_prompt": "You are Assistant, an intelligent companion. You talk naturally with the user, provide thorough explanations, and manage workspace files methodically.",
"created_at": time.time() + 1,
"messages": []
}
]
self.save()
def get_all_bots(self) -> list[dict]:
# Sync workspace path dynamically on every retrieval
for b in self.bots:
b["workspace_dir"] = get_bot_workspace_dir(b.get("name", "Bot"))
return self.bots
def get_bot(self, bot_id: str) -> dict | None:
for b in self.bots:
if b.get("id") == bot_id:
b["workspace_dir"] = get_bot_workspace_dir(b.get("name", "Bot"))
return b
return None
def create_bot(self, name: str, model: str, color: str, custom_system_prompt: str) -> tuple[bool, str, dict | None]:
if len(self.bots) >= MAX_BOTS:
return False, f"Maximum limit of {MAX_BOTS} bots reached. Delete an existing bot to create a new one.", None
clean_name = name.strip()
if not clean_name:
return False, "Bot name cannot be empty.", None
# Ensure unique name
for b in self.bots:
if b.get("name", "").lower() == clean_name.lower():
return False, f"A bot named '{clean_name}' already exists.", None
new_bot = {
"id": str(uuid.uuid4()),
"name": clean_name,
"model": model,
"color": color if color else DEFAULT_PALETTE[len(self.bots) % len(DEFAULT_PALETTE)],
"custom_system_prompt": custom_system_prompt.strip(),
"created_at": time.time(),
"messages": []
}
# Pre-create workspace folder
get_bot_workspace_dir(clean_name)
self.bots.append(new_bot)
self.save()
return True, "Bot created successfully.", new_bot
def update_bot(self, bot_id: str, name: str, model: str, color: str, custom_system_prompt: str) -> tuple[bool, str]:
bot = self.get_bot(bot_id)
if not bot:
return False, "Bot not found."
clean_name = name.strip()
if not clean_name:
return False, "Bot name cannot be empty."
old_name = bot.get("name", "")
# If name changed, rename workspace folder if possible
if clean_name.lower() != old_name.lower():
for b in self.bots:
if b.get("id") != bot_id and b.get("name", "").lower() == clean_name.lower():
return False, f"A bot named '{clean_name}' already exists."
old_dir = get_bot_workspace_dir(old_name)
new_dir = get_bot_workspace_dir(clean_name)
if os.path.exists(old_dir) and old_dir != new_dir:
try:
# Move contents if new_dir is empty
if not os.listdir(new_dir):
os.rmdir(new_dir)
shutil.move(old_dir, new_dir)
except Exception as e:
print(f"[MagiBotManager] Could not rename workspace: {e}")
bot["name"] = clean_name
bot["model"] = model
bot["color"] = color
bot["custom_system_prompt"] = custom_system_prompt.strip()
self.save()
return True, "Bot updated successfully."
def delete_bot(self, bot_id: str) -> bool:
initial_len = len(self.bots)
self.bots = [b for b in self.bots if b.get("id") != bot_id]
if len(self.bots) < initial_len:
self.save()
return True
return False
def add_message(self, bot_id: str, sender: str, text: str, action: dict = None, action_result: str = None):
bot = self.get_bot(bot_id)
if not bot:
return
if "messages" not in bot:
bot["messages"] = []
msg_obj = {
"id": str(uuid.uuid4()),
"sender": sender, # "user" | "bot" | "system"
"text": text,
"timestamp": datetime.datetime.now().strftime("%H:%M"),
"action": action,
"action_result": action_result
}
bot["messages"].append(msg_obj)
self.save()
def clear_messages(self, bot_id: str):
bot = self.get_bot(bot_id)
if bot:
bot["messages"] = []
self.save()
# Global manager instance
bot_manager = MagiBotManager()
# ============================================================================
# 3. WORKSPACE FILE & EXECUTION TOOLS
# ============================================================================
class BotWorkspaceTools:
"""
Safe file and execution operations strictly scoped within the bot workspace.
"""
@staticmethod
def resolve_safe_path(workspace_dir: str, rel_path: str) -> str:
# Prevent directory traversal attacks
clean_rel = rel_path.lstrip("/\\")
full_path = os.path.abspath(os.path.join(workspace_dir, clean_rel))
return full_path
@staticmethod
def resolve_move_destination(workspace_dir: str, dst_path: str) -> str:
raw = os.path.expanduser((dst_path or "").strip())
if os.path.isabs(raw):
return os.path.abspath(raw)
return os.path.abspath(os.path.join(workspace_dir, raw.lstrip("/\\")))
@staticmethod
def path_leaves_workspace(workspace_dir: str, path: str) -> bool:
ws = os.path.abspath(workspace_dir)
target = BotWorkspaceTools.resolve_move_destination(workspace_dir, path)
try:
return os.path.commonpath([ws, target]) != ws
except ValueError:
return True
@staticmethod
def move_leaves_workspace(workspace_dir: str, action: dict) -> bool:
name = str(action.get("action", "")).strip().lower()
if name not in ("move_file", "rename_file", "mv"):
return False
dst = action.get("destination") or action.get("dst") or action.get("to") or ""
return BotWorkspaceTools.path_leaves_workspace(workspace_dir, dst)
@staticmethod
def create_file(workspace_dir: str, rel_path: str, content: str) -> tuple[bool, str]:
try:
full_path = BotWorkspaceTools.resolve_safe_path(workspace_dir, rel_path)
os.makedirs(os.path.dirname(full_path), exist_ok=True)
with open(full_path, "w", encoding="utf-8") as f:
f.write(content)
return True, f"File '{rel_path}' created successfully ({len(content)} chars written)."
except Exception as e:
return False, f"Failed to create file '{rel_path}': {str(e)}"
@staticmethod
def create_folder(workspace_dir: str, rel_path: str) -> tuple[bool, str]:
try:
full_path = BotWorkspaceTools.resolve_safe_path(workspace_dir, rel_path)
os.makedirs(full_path, exist_ok=True)
return True, f"Folder '{rel_path}' created successfully in workspace."
except Exception as e:
return False, f"Failed to create folder '{rel_path}': {str(e)}"
@staticmethod
def edit_file(workspace_dir: str, rel_path: str, content: str = "", mode: str = "overwrite", target: str = "", replacement: str = "") -> tuple[bool, str]:
try:
full_path = BotWorkspaceTools.resolve_safe_path(workspace_dir, rel_path)
if not os.path.exists(full_path) and mode != "overwrite":
return False, f"File '{rel_path}' does not exist."
os.makedirs(os.path.dirname(full_path), exist_ok=True)
if mode == "replace" and target:
if not os.path.exists(full_path):
return False, f"File '{rel_path}' does not exist for search & replace."
with open(full_path, "r", encoding="utf-8", errors="replace") as f:
old_text = f.read()
if target not in old_text:
return False, f"Target text to replace was not found in '{rel_path}'."
new_text = old_text.replace(target, replacement if replacement else content)
with open(full_path, "w", encoding="utf-8") as f:
f.write(new_text)
return True, f"Replaced target text in '{rel_path}' successfully."
elif mode == "append":
with open(full_path, "a", encoding="utf-8") as f:
f.write(content)
return True, f"File '{rel_path}' appended successfully."
else:
with open(full_path, "w", encoding="utf-8") as f:
f.write(content)
return True, f"File '{rel_path}' updated successfully ({len(content)} chars written)."
except Exception as e:
return False, f"Failed to edit file '{rel_path}': {str(e)}"
@staticmethod
def delete_file(workspace_dir: str, rel_path: str) -> tuple[bool, str]:
try:
full_path = BotWorkspaceTools.resolve_safe_path(workspace_dir, rel_path)
if not os.path.exists(full_path):
return False, f"File or folder '{rel_path}' does not exist."
if os.path.isdir(full_path):
shutil.rmtree(full_path)
return True, f"Directory '{rel_path}' and its contents deleted."
else:
os.remove(full_path)
return True, f"File '{rel_path}' deleted."
except Exception as e:
return False, f"Failed to delete '{rel_path}': {str(e)}"
@staticmethod
def move_file(workspace_dir: str, src_path: str, dst_path: str) -> tuple[bool, str]:
try:
full_src = BotWorkspaceTools.resolve_safe_path(workspace_dir, src_path)
full_dst = BotWorkspaceTools.resolve_move_destination(workspace_dir, dst_path)
if BotWorkspaceTools.path_leaves_workspace(workspace_dir, src_path):
return False, f"Source '{src_path}' must stay inside the workspace."
if not os.path.exists(full_src):
return False, f"Source '{src_path}' does not exist."
os.makedirs(os.path.dirname(full_dst), exist_ok=True)
shutil.move(full_src, full_dst)
return True, f"Moved '{src_path}' to '{dst_path}' successfully."
except Exception as e:
return False, f"Failed to move '{src_path}' to '{dst_path}': {str(e)}"
@staticmethod
def read_file(workspace_dir: str, rel_path: str) -> tuple[bool, str]:
try:
full_path = BotWorkspaceTools.resolve_safe_path(workspace_dir, rel_path)
if not os.path.exists(full_path):
# Search case-insensitively in workspace
clean_name = os.path.basename(rel_path).strip().lower()
for root, _, files in os.walk(workspace_dir):
for f in files:
if f.lower() == clean_name:
full_path = os.path.join(root, f)
break
if os.path.exists(full_path):
break
if not os.path.exists(full_path):
return False, f"File '{rel_path}' not found in workspace '{workspace_dir}'."
with open(full_path, "r", encoding="utf-8", errors="replace") as f:
content = f.read(50000) # limit 50KB for context
return True, f"Content of '{rel_path}' ({len(content)} chars):\n\n{content}"
except Exception as e:
return False, f"Failed to read file '{rel_path}': {str(e)}"
@staticmethod
def list_files(workspace_dir: str, rel_path: str = ".") -> tuple[bool, str]:
try:
target_dir = BotWorkspaceTools.resolve_safe_path(workspace_dir, rel_path)
if not os.path.exists(target_dir):
return False, f"Directory '{rel_path}' does not exist."
items = []
for root, dirs, files in os.walk(target_dir):
rel_root = os.path.relpath(root, workspace_dir)
for d in dirs:
p = os.path.normpath(os.path.join(rel_root, d)).replace("\\", "/")
items.append(f"[DIR] {p}")
for f in files:
p = os.path.normpath(os.path.join(rel_root, f)).replace("\\", "/")
try:
sz = os.path.getsize(os.path.join(root, f))
items.append(f"[FILE] {p} ({sz} bytes)")
except Exception:
items.append(f"[FILE] {p}")
if not items:
return True, "Workspace is currently empty."
return True, "\n".join(items)
except Exception as e:
return False, f"Failed to list files: {str(e)}"
SUPPORTED_PACKAGES = {
"pygame": "pygame",
"numpy": "numpy",
"pandas": "pandas",
"pillow": "pillow",
"pyqt6": "PyQt6",
"customtkinter": "customtkinter"
}
@staticmethod
def get_venv_python(workspace_dir: str) -> str:
venv_dir = os.path.join(workspace_dir, ".venv")
if sys.platform == "win32":
py_path = os.path.join(venv_dir, "Scripts", "python.exe")
else:
py_path = os.path.join(venv_dir, "bin", "python")
return py_path if os.path.exists(py_path) else sys.executable
@staticmethod
def ensure_venv(workspace_dir: str) -> tuple[bool, str]:
venv_dir = os.path.join(workspace_dir, ".venv")
if not os.path.exists(venv_dir):
try:
cmd = [sys.executable, "-m", "venv", venv_dir]
res = subprocess.run(cmd, capture_output=True, text=True, timeout=60)
if res.returncode != 0:
return False, f"Failed to create .venv: {res.stderr.strip()}"
except Exception as e:
return False, f"Exception creating .venv: {str(e)}"
return True, venv_dir
@staticmethod
def install_package(workspace_dir: str, package_name: str) -> tuple[bool, str]:
clean_pkg = package_name.strip().lower()
if clean_pkg not in BotWorkspaceTools.SUPPORTED_PACKAGES:
allowed = ", ".join(BotWorkspaceTools.SUPPORTED_PACKAGES.values())
return False, f"Package '{package_name}' is not supported. Currently allowed packages for .venv: {allowed}"
target_pkg = BotWorkspaceTools.SUPPORTED_PACKAGES[clean_pkg]
ok, venv_info = BotWorkspaceTools.ensure_venv(workspace_dir)
if not ok:
return False, venv_info
venv_py = BotWorkspaceTools.get_venv_python(workspace_dir)
try:
cmd = [venv_py, "-m", "pip", "install", target_pkg]
proc = subprocess.Popen(
cmd,
cwd=workspace_dir,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
)
try:
stdout, stderr = proc.communicate(timeout=120)
except subprocess.TimeoutExpired:
proc.kill()
return False, f"Installation of '{target_pkg}' timed out."
if proc.returncode == 0:
return True, f"Successfully installed '{target_pkg}' into workspace .venv."
else:
return False, f"Installation of '{target_pkg}' failed:\n{stderr.strip()}"
except Exception as e:
return False, f"Error installing package '{target_pkg}': {str(e)}"
@staticmethod
def execute_file(workspace_dir: str, rel_path: str, args: list = None) -> tuple[bool, str]:
try:
full_path = BotWorkspaceTools.resolve_safe_path(workspace_dir, rel_path)
if not os.path.exists(full_path):
return False, f"File to execute '{rel_path}' does not exist."
ext = os.path.splitext(full_path)[1].lower()
cmd = []
if ext == ".py":
venv_py = BotWorkspaceTools.get_venv_python(workspace_dir)
cmd = [venv_py, full_path]
elif ext in (".bat", ".cmd"):
cmd = [full_path]
elif ext == ".ps1":
cmd = ["powershell", "-ExecutionPolicy", "Bypass", "-File", full_path]
elif ext == ".js":
cmd = ["node", full_path]
else:
cmd = [full_path]
if args and isinstance(args, list):
cmd.extend([str(a) for a in args])
# Strictly hide console and GUI windows from the user
startupinfo = None
creationflags = 0
if sys.platform == "win32":
startupinfo = subprocess.STARTUPINFO()
startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW
startupinfo.wShowWindow = 0 # SW_HIDE
creationflags = 0x08000000 # CREATE_NO_WINDOW
env = os.environ.copy()
env["PYTHONUNBUFFERED"] = "1"
env["SDL_VIDEODRIVER"] = "dummy"
env["SDL_AUDIODRIVER"] = "dummy"
env["PYGAME_HIDE_SUPPORT_PROMPT"] = "1"
env["QT_QPA_PLATFORM"] = "offscreen"
env["MPLBACKEND"] = "Agg"
env["TK_SILENT"] = "1"
proc = subprocess.Popen(
cmd,
cwd=workspace_dir,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
shell=(ext in (".bat", ".cmd")),
startupinfo=startupinfo,
creationflags=creationflags,
env=env
)
# Start background window hider in case a framework creates a visible HWND
if sys.platform == "win32":
def hide_windows_quick(p_pid):
try:
import win32gui, win32process, win32con
for _ in range(30):
time.sleep(0.04)
def enum_cb(hwnd, _):
if win32gui.IsWindow(hwnd):
try:
_, w_pid = win32process.GetWindowThreadProcessId(hwnd)
if w_pid == p_pid:
win32gui.ShowWindow(hwnd, win32con.SW_HIDE)
win32gui.SetWindowPos(hwnd, 0, -32000, -32000, 0, 0, win32con.SWP_NOSIZE | win32con.SWP_NOZORDER | win32con.SWP_NOACTIVATE)
except Exception:
pass
return True
win32gui.EnumWindows(enum_cb, None)
except Exception:
pass
threading.Thread(target=hide_windows_quick, args=(proc.pid,), daemon=True).start()
try:
stdout, stderr = proc.communicate(timeout=30)
except subprocess.TimeoutExpired:
proc.kill()
return False, "Execution timed out after 30 seconds."
out_text = ""
if stdout:
out_text += f"STDOUT:\n{stdout}\n"
if stderr:
out_text += f"STDERR:\n{stderr}\n"
out_text += f"Exit Code: {proc.returncode}"
success = (proc.returncode == 0)
return success, out_text.strip()
except Exception as e:
return False, f"Execution failed: {str(e)}"
@staticmethod
def web_search(query: str, max_results: int = 3) -> tuple[bool, str]:
clean_query = query.strip().strip("\"' ")
if not clean_query:
return False, "Search query cannot be empty."
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/123.0.0.0 Safari/537.36"
}
# Try Bing RSS live search
results = []
try:
url = f"https://www.bing.com/search?q={urllib.parse.quote(clean_query)}&format=rss"
resp = requests.get(url, headers=headers, timeout=8)
if resp.status_code == 200:
root = ET.fromstring(resp.text)
items = root.findall(".//item")
for it in items[:max_results]:
title = html.unescape(it.findtext("title", "")).strip()
link = it.findtext("link", "").strip()
desc = html.unescape(it.findtext("description", "")).strip()
results.append({
"title": title,
"url": link,
"snippet": desc
})
except Exception as e:
print(f"[MagiBot] RSS search error: {e}")
# Fallback to Wikipedia OpenSearch if needed
if not results:
try:
wiki_url = f"https://en.wikipedia.org/w/api.php?action=opensearch&search={urllib.parse.quote(clean_query)}&limit={max_results}&namespace=0&format=json"
w_resp = requests.get(wiki_url, headers={"User-Agent": "MagiBot/1.0 (Educational)"}, timeout=6)
if w_resp.status_code == 200:
data = w_resp.json()
titles = data[1]
snippets = data[2]
urls = data[3]
for i in range(len(titles)):
results.append({
"title": titles[i],
"url": urls[i],
"snippet": snippets[i] if i < len(snippets) else ""
})
except Exception as e:
print(f"[MagiBot] Wiki fallback error: {e}")
if not results:
return False, f"No web search results found for: '{clean_query}'"
out_lines = [f"Web Search Results for: \"{clean_query}\" (Top {len(results)} sources):\n"]
for idx, r in enumerate(results[:max_results], 1):
out_lines.append(f"{idx}. {r['title']}")
out_lines.append(f" URL: {r['url']}")
if r['snippet']:
out_lines.append(f" Snippet: {r['snippet']}")
out_lines.append("")
return True, "\n".join(out_lines).strip()
@staticmethod
def fetch_web(url: str, max_words: int = 2500) -> tuple[bool, str]:
clean_url = url.strip().strip("<>[]()\"' ")
if not clean_url:
return False, "URL cannot be empty."
if not clean_url.startswith(("http://", "https://")):
clean_url = "https://" + clean_url
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/123.0.0.0 Safari/537.36",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Accept-Language": "en-US,en;q=0.9"
}
raw_html = ""
try:
session = requests.Session()
resp = session.get(clean_url, headers=headers, timeout=15)
resp.raise_for_status()
raw_html = resp.text
except Exception:
try:
req = urllib.request.Request(clean_url, headers=headers)
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
with urllib.request.urlopen(req, context=ctx, timeout=15) as response:
raw_html = response.read().decode("utf-8", errors="replace")
except Exception as e:
return False, f"Failed to fetch webpage content from '{clean_url}': {str(e)}"
try:
# Strip script, style, nav, header, footer, svg, noscript tags
cleaned = re.sub(r'<(script|style|nav|header|footer|aside|svg|noscript|iframe)[^>]*>[\s\S]*?', ' ', raw_html, flags=re.IGNORECASE)
# Strip all HTML tags
text = re.sub(r'<[^>]+>', ' ', cleaned)
# Unescape all HTML entities
text = html.unescape(text)
# Clean and normalize whitespace
lines = [line.strip() for line in text.splitlines() if line.strip()]
full_text = "\n".join(lines)
words = full_text.split()
word_count = len(words)
if word_count > max_words:
truncated = " ".join(words[:max_words])
result_str = f"Fetched {max_words} words from {clean_url} (Total words: {word_count}):\n\n{truncated}\n\n[Truncated at {max_words} words]"
else:
result_str = f"Fetched {word_count} words from {clean_url}:\n\n{full_text}"
return True, result_str
except Exception as e:
return False, f"Failed to parse webpage content from '{clean_url}': {str(e)}"
@staticmethod
def launch_headless_app(workspace_dir: str, rel_path: str, args: list = None) -> tuple[bool, str]:
ok, msg, _ = headless_app_manager.launch_app(workspace_dir, rel_path, args)
return ok, msg
@staticmethod
def inspect_headless_app(app_id: str) -> tuple[bool, str]:
return headless_app_manager.inspect_app(app_id)
@staticmethod
def interact_headless_app(app_id: str, action_type: str = "click", target: str = "", text: str = "", x: int = 0, y: int = 0) -> tuple[bool, str]:
return headless_app_manager.interact_app(app_id, action_type, target, text, x, y)
@staticmethod
def close_headless_app(app_id: str) -> tuple[bool, str]:
return headless_app_manager.close_app(app_id)
@staticmethod
def get_youtube_transcript(url: str) -> tuple[bool, str]:
return False, "YouTube transcripts are not included. Magi Bot is local-only (Ollama)."
# ============================================================================
# 3B. HEADLESS APP MANAGER & OCR / UI AUTOMATION
# ============================================================================
class HeadlessAppManager:
"""
Manages background/headless Python applications running over the workspace .venv.
Features:
1. Spawns scripts completely in the background without opening or flashing visible windows on the user desktop.
2. Inspects UI elements (buttons, textboxes, labels) and terminal stdout/stderr using Win32 UI control enumeration and OCR analysis.
3. Interacts with background apps (click buttons, set text, send stdin) without stealing mouse cursor or window focus.
4. Cleanly closes and tracks background application lifecycles.
"""
def __init__(self):
self._apps: dict[str, dict] = {}
self._lock = threading.Lock()
self._counter = 0
def launch_app(self, workspace_dir: str, rel_path: str, args: list = None) -> tuple[bool, str, str]:
try:
full_path = BotWorkspaceTools.resolve_safe_path(workspace_dir, rel_path)
if not os.path.exists(full_path):
return False, f"Script '{rel_path}' not found in workspace.", ""
venv_py = BotWorkspaceTools.get_venv_python(workspace_dir)
cmd = [venv_py, full_path]
if args and isinstance(args, list):
cmd.extend([str(a) for a in args])
startupinfo = None
creationflags = 0
if sys.platform == "win32":
startupinfo = subprocess.STARTUPINFO()
startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW
startupinfo.wShowWindow = 0 # SW_HIDE
creationflags = 0x08000000 # CREATE_NO_WINDOW
env = os.environ.copy()
env["PYTHONUNBUFFERED"] = "1"
env["SDL_VIDEODRIVER"] = "dummy"
env["SDL_AUDIODRIVER"] = "dummy"
env["PYGAME_HIDE_SUPPORT_PROMPT"] = "1"
env["QT_QPA_PLATFORM"] = "offscreen"
env["MPLBACKEND"] = "Agg"
env["TK_SILENT"] = "1"
proc = subprocess.Popen(
cmd,
cwd=workspace_dir,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
bufsize=1,
startupinfo=startupinfo,
creationflags=creationflags,
env=env
)
with self._lock:
self._counter += 1
app_id = f"app_{self._counter}"
app_entry = {
"id": app_id,
"pid": proc.pid,
"path": rel_path,
"proc": proc,
"start_time": time.time(),
"stdout_lines": [],
"stderr_lines": [],
"workspace_dir": workspace_dir,
"is_running": True
}
self._apps[app_id] = app_entry
# Continuous background window hider
if sys.platform == "win32":
def continuous_window_hider(p_pid, entry):
try:
import win32gui, win32process, win32con
while entry.get("is_running", True) and proc.poll() is None:
def enum_cb(hwnd, _):
if win32gui.IsWindow(hwnd):
try:
_, w_pid = win32process.GetWindowThreadProcessId(hwnd)
if w_pid == p_pid:
win32gui.ShowWindow(hwnd, win32con.SW_HIDE)
win32gui.SetWindowPos(hwnd, 0, -32000, -32000, 0, 0, win32con.SWP_NOSIZE | win32con.SWP_NOZORDER | win32con.SWP_NOACTIVATE)
except Exception:
pass
return True
win32gui.EnumWindows(enum_cb, None)
time.sleep(0.1)
except Exception:
pass
threading.Thread(target=continuous_window_hider, args=(proc.pid, app_entry), daemon=True).start()
# Start background stdout / stderr reader threads
def read_stream(stream, target_list):
try:
for line in iter(stream.readline, ''):
if line:
target_list.append(line.rstrip("\r\n"))
if len(target_list) > 200:
target_list.pop(0)
else:
break
except Exception:
pass
t_out = threading.Thread(target=read_stream, args=(proc.stdout, app_entry["stdout_lines"]), daemon=True)
t_err = threading.Thread(target=read_stream, args=(proc.stderr, app_entry["stderr_lines"]), daemon=True)
t_out.start()
t_err.start()
# Give the app a moment to spin up
time.sleep(0.4)
return True, f"Headless Python app '{rel_path}' started in background with App ID: '{app_id}' (PID: {proc.pid}). No window opened on user screen.", app_id
except Exception as e:
return False, f"Failed to launch headless app '{rel_path}': {str(e)}", ""
def _get_hwnds_for_pid(self, pid: int) -> list[int]:
hwnds = []
if sys.platform != "win32":
return hwnds
try:
import win32gui, win32process
def enum_cb(hwnd, _):
if win32gui.IsWindow(hwnd):
try:
_, win_pid = win32process.GetWindowThreadProcessId(hwnd)
if win_pid == pid:
hwnds.append(hwnd)
except Exception:
pass
return True
win32gui.EnumWindows(enum_cb, None)
except Exception:
pass
return hwnds
def _inspect_child_controls(self, hwnd: int) -> list[dict]:
controls = []
if sys.platform != "win32":
return controls
try:
import win32gui, win32con
def enum_child_cb(child, _):
try:
c_class = win32gui.GetClassName(child)
c_text = win32gui.GetWindowText(child)
rect = win32gui.GetWindowRect(child)
w = rect[2] - rect[0]
h = rect[3] - rect[1]
c_id = win32gui.GetWindowLong(child, win32con.GWL_ID)
ctrl_type = "CONTROL"
class_lower = c_class.lower()
if "button" in class_lower:
ctrl_type = "BUTTON"
elif "edit" in class_lower or "entry" in class_lower or "text" in class_lower:
ctrl_type = "TEXTBOX"
elif "static" in class_lower or "label" in class_lower:
ctrl_type = "LABEL"
elif "list" in class_lower:
ctrl_type = "LIST"
elif "combo" in class_lower:
ctrl_type = "COMBOBOX"
controls.append({
"type": ctrl_type,
"hwnd": hex(child),
"hwnd_int": child,
"class": c_class,
"text": c_text,
"id": c_id,
"x": rect[0],
"y": rect[1],
"width": w,
"height": h
})
except Exception:
pass
return True
win32gui.EnumChildWindows(hwnd, enum_child_cb, None)
except Exception:
pass
return controls
def inspect_app(self, app_id: str) -> tuple[bool, str]:
with self._lock:
app = self._apps.get(app_id)
if not app:
return False, f"Headless App '{app_id}' not found. Active apps: {list(self._apps.keys())}"
proc = app["proc"]
poll_res = proc.poll()
is_running = (poll_res is None)
uptime = round(time.time() - app["start_time"], 1)
report = [
f"=== HEADLESS APP INSPECTION REPORT: {app_id} ===",
f"Script: {app['path']} | Status: {'RUNNING' if is_running else f'TERMINATED (Exit Code: {poll_res})'}",
f"PID: {app['pid']} | Uptime: {uptime}s\n"
]
# 1. UI Controls & Elements detection
hwnds = self._get_hwnds_for_pid(app["pid"])
all_controls = []
for h in hwnds:
ctrls = self._inspect_child_controls(h)
all_controls.extend(ctrls)
if all_controls:
report.append("--- DETECTED UI CONTROLS (Buttons, Textfields, Outputs) ---")
for idx, c in enumerate(all_controls, 1):
label_str = f'"{c["text"]}"' if c["text"] else "(Empty)"
report.append(f"{idx}. [{c['type']}] Text: {label_str} | Pos: (x: {c['x']}, y: {c['y']}, w: {c['width']}, h: {c['height']}) | Handle: {c['hwnd']}")
report.append("")
elif hwnds:
report.append(f"--- WINDOW DETECTED --- (Handle: {hex(hwnds[0])}) — No standard Win32 child handles; rendered via offscreen buffer.")
report.append("")
# 2. Latest stdout logs
stdout_tail = app["stdout_lines"][-15:]
if stdout_tail:
report.append("--- RECENT STDOUT OUTPUT ---")
report.extend(stdout_tail)
report.append("")
# 3. Latest stderr logs
stderr_tail = app["stderr_lines"][-10:]
if stderr_tail:
report.append("--- RECENT STDERR LOGS ---")
report.extend(stderr_tail)
report.append("")
if not all_controls and not stdout_tail and not stderr_tail:
report.append("(No UI controls or terminal output detected yet)")
return True, "\n".join(report).strip()
def interact_app(self, app_id: str, action_type: str, target: str = "", text: str = "", x: int = 0, y: int = 0) -> tuple[bool, str]:
with self._lock:
app = self._apps.get(app_id)
if not app:
return False, f"Headless App '{app_id}' not found."
proc = app["proc"]
if proc.poll() is not None:
return False, f"Headless App '{app_id}' is no longer running (Exit Code: {proc.poll()})."
act = action_type.lower().strip()
hwnds = self._get_hwnds_for_pid(app["pid"])
controls = []
for h in hwnds:
controls.extend(self._inspect_child_controls(h))
if act == "click":
# 1. Try to find control by text or handle or ID
target_clean = str(target).strip().lower()
matched_ctrl = None
for c in controls:
if (c["text"] and c["text"].lower() == target_clean) or c["hwnd"].lower() == target_clean or str(c["id"]) == target_clean:
matched_ctrl = c
break
if matched_ctrl:
try:
import win32gui, win32con
hwnd_target = matched_ctrl["hwnd_int"]
win32gui.SendMessage(hwnd_target, win32con.BM_CLICK, 0, 0)
time.sleep(0.2)
return True, f"Successfully clicked button/control [{matched_ctrl['type']}] '{matched_ctrl['text']}' (Handle: {matched_ctrl['hwnd']}) in background."
except Exception as e:
return False, f"Click failed on control: {e}"
# 2. Try click by coordinate on main window HWND
if hwnds and (x > 0 or y > 0):
try:
import win32gui, win32con, win32api
lparam = win32api.MAKELONG(x, y)
main_hwnd = hwnds[0]
win32gui.PostMessage(main_hwnd, win32con.WM_LBUTTONDOWN, win32con.MK_LBUTTON, lparam)
win32gui.PostMessage(main_hwnd, win32con.WM_LBUTTONUP, 0, lparam)
time.sleep(0.2)
return True, f"Sent mouse click to ({x}, {y}) on background window ({hex(main_hwnd)})."
except Exception as e:
return False, f"Coordinate click failed: {e}"
return False, f"Could not find matching button/control for target '{target}'."
elif act in ("type", "set_text"):
# Set text in edit control
target_clean = str(target).strip().lower()
matched_ctrl = None
for c in controls:
if c["type"] == "TEXTBOX" or (c["text"] and c["text"].lower() == target_clean) or c["hwnd"].lower() == target_clean:
matched_ctrl = c
break
if matched_ctrl:
try:
import win32gui, win32con
win32gui.SendMessage(matched_ctrl["hwnd_int"], win32con.WM_SETTEXT, 0, text)
time.sleep(0.1)
return True, f"Set text '{text}' into [{matched_ctrl['type']}] (Handle: {matched_ctrl['hwnd']})."
except Exception as e:
return False, f"Failed to set text in control: {e}"
# If no control, write to stdin
if proc.stdin:
try:
proc.stdin.write(text + "\n")
proc.stdin.flush()
time.sleep(0.2)
return True, f"Sent '{text}' to app stdin."
except Exception as e:
return False, f"Failed to write to stdin: {e}"
return False, f"Could not find input control or stdin for target '{target}'."
elif act == "send_stdin":
if proc.stdin:
try:
proc.stdin.write(text + "\n")
proc.stdin.flush()
time.sleep(0.2)
return True, f"Sent '{text}' to stdin of headless app '{app_id}'."
except Exception as e:
return False, f"Failed to write to stdin: {e}"
return False, "App stdin is not available."
elif act == "wait":
wait_sec = min(float(text if text else 1.0), 10.0)
time.sleep(wait_sec)
return True, f"Waited {wait_sec}s for headless app."
else:
return False, f"Unknown interaction action: '{act}'"
def close_app(self, app_id: str) -> tuple[bool, str]:
with self._lock:
app = self._apps.get(app_id)
if not app:
return False, f"Headless App '{app_id}' not found."
proc = app["proc"]
try:
hwnds = self._get_hwnds_for_pid(app["pid"])
import win32gui, win32con
for h in hwnds:
try:
win32gui.PostMessage(h, win32con.WM_CLOSE, 0, 0)
except Exception:
pass
except Exception:
pass
try:
proc.terminate()
proc.wait(timeout=2.0)
except Exception:
try:
proc.kill()
except Exception:
pass
with self._lock:
self._apps.pop(app_id, None)
return True, f"Headless App '{app_id}' closed successfully."
def list_running_apps(self) -> list[dict]:
res = []
with self._lock:
for aid, app in list(self._apps.items()):
poll_res = app["proc"].poll()
res.append({
"id": aid,
"pid": app["pid"],
"path": app["path"],
"is_running": (poll_res is None),
"uptime": round(time.time() - app["start_time"], 1)
})
return res
headless_app_manager = HeadlessAppManager()
# ============================================================================
# 4. ACTION PARSER & SYSTEM PROMPT GENERATOR
# ============================================================================
ACTION_SYSTEM_INSTRUCTION = """
You are an intelligent AI assistant operating inside your dedicated workspace folder:
{workspace_path}
Files in your workspace:
{workspace_files}
HOW TO COMMUNICATE:
1. CHAT & QUESTIONS:
If the user is chatting, asking questions, or discussing ideas, respond normally in plain text without any action block.
2. PERFORMING ACTIONS (Files, Web, Running Code, Testing):
Explain what you are doing in your message, and append EXACTLY ONE action block at the VERY END using this format:
```action
{{"action": "ACTION_NAME", "param1": "value1"}}
```
AVAILABLE ACTIONS:
- Read File:
```action
{{"action": "read_file", "path": "filename.py"}}
```
- Create / Write File:
```action
{{"action": "create_file", "path": "filename.py", "content": "print('hello')"}}
```
- Edit / Update File:
```action
{{"action": "edit_file", "path": "filename.py", "content": "updated code", "mode": "overwrite"}}
```
- Execute Python File (.venv):
```action
{{"action": "execute_file", "path": "filename.py"}}
```
- Read Webpage / URL (reads live text from any link):
```action
{{"action": "fetch_web", "url": "https://example.com/docs"}}
```
- Search Web (searches Google/web for info & documentation):
```action
{{"action": "web_search", "query": "python socket programming"}}
```
- Install Package (.venv):
```action
{{"action": "install_package", "package": "pygame"}}
```
- Create Folder:
```action
{{"action": "create_folder", "path": "src"}}
```
- Delete File or Folder:
```action
{{"action": "delete_file", "path": "temp.txt"}}
```
- Move / Rename File:
```action
{{"action": "move_file", "source": "old.py", "destination": "new.py"}}
```
- Get YouTube Transcript:
```action
{{"action": "get_youtube_transcript", "url": "https://www.youtube.com/watch?v=..."}}
```
- Launch Headless App (.venv GUI test):
```action
{{"action": "launch_headless_app", "path": "app.py"}}
```
- Inspect Headless App:
```action
{{"action": "inspect_headless_app", "app_id": "app_1"}}
```
- Interact Headless App:
```action
{{"action": "interact_headless_app", "app_id": "app_1", "action_type": "click", "target": "Btn"}}
```
- Close Headless App:
```action
{{"action": "close_headless_app", "app_id": "app_1"}}
```
- Ask Helper Bot for Advice:
```action
{{"action": "ask_helper", "question": "Question text", "model": "lfm2.5:8b"}}
```
- Task Done:
```action
{{"action": "done", "summary": "All steps completed successfully."}}
```
RULES:
- Emit at most ONE action per message at the VERY END.
- NEVER use emojis in your response.
"""
def _parse_action_json_or_dict(s: str) -> dict | None:
"""Attempts to parse string into a dictionary using json, ast, or field regex."""
if not s:
return None
s_clean = s.strip()
# 1. Standard JSON with trailing comma removal
cleaned_json = re.sub(r',\s*([\}\]])', r'\1', s_clean)
try:
data = json.loads(cleaned_json)
if isinstance(data, dict):
return data
except Exception:
pass
# 2. ast.literal_eval for single quoted dicts
try:
import ast
data = ast.literal_eval(s_clean)
if isinstance(data, dict):
return data
except Exception:
pass
# 3. Regex field extraction fallback for malformed JSON (e.g. unescaped newlines/quotes in code)
action_match = re.search(r'["\']?(?:action|tool|command|function|name)["\']?\s*:\s*["\']([^"\']+)["\']', s_clean, re.IGNORECASE)
if action_match:
data = {"action": action_match.group(1)}
path_match = re.search(r'["\']?(?:path|file|filename|target|filepath)["\']?\s*:\s*["\']([^"\']+)["\']', s_clean, re.IGNORECASE)
if path_match:
data["path"] = path_match.group(1)
url_match = re.search(r'["\']?(?:url|link|target_url|uri)["\']?\s*:\s*["\']([^"\']+)["\']', s_clean, re.IGNORECASE)
if url_match:
data["url"] = url_match.group(1)
query_match = re.search(r'["\']?(?:query|search|search_query|prompt)["\']?\s*:\s*["\']([^"\']+)["\']', s_clean, re.IGNORECASE)
if query_match:
data["query"] = query_match.group(1)
pkg_match = re.search(r'["\']?(?:package|pkg|library)["\']?\s*:\s*["\']([^"\']+)["\']', s_clean, re.IGNORECASE)
if pkg_match:
data["package"] = pkg_match.group(1)
content_match = re.search(r'["\']?(?:content|code|text|source)["\']?\s*:\s*["\']([\s\S]*?)["\']\s*[,}]', s_clean, re.IGNORECASE)
if content_match:
data["content"] = content_match.group(1)
return data
return None
def _is_valid_action_dict(d: dict) -> bool:
if not isinstance(d, dict):
return False
keys = ["action", "tool", "command", "function", "name", "type"]
return any(k in d and bool(d[k]) for k in keys)
def _normalize_action_dict(d: dict) -> dict:
normalized = dict(d)
# Action key
for k in ("tool", "command", "function", "name", "type"):
if "action" not in normalized and k in normalized:
normalized["action"] = normalized[k]
# Path key
for k in ("file", "filename", "file_path", "target", "filepath"):
if "path" not in normalized and k in normalized:
normalized["path"] = normalized[k]
# Content key
for k in ("code", "text", "body", "source"):
if "content" not in normalized and k in normalized:
normalized["content"] = normalized[k]
# URL key
for k in ("link", "target_url", "uri"):
if "url" not in normalized and k in normalized:
normalized["url"] = normalized[k]
# Query key
for k in ("search", "prompt", "search_query"):
if "query" not in normalized and k in normalized:
normalized["query"] = normalized[k]
# Package key
for k in ("pkg", "library"):
if "package" not in normalized and k in normalized:
normalized["package"] = normalized[k]
return normalized
def _extract_balanced_json_objects(text: str) -> list[tuple[int, int, str]]:
"""
Finds top-level balanced JSON/dict objects { ... } with their (start, end, content).
Properly handles nested braces, string quotes, and escape slashes.
"""
results = []
i = 0
n = len(text)
while i < n:
if text[i] == '{':
start = i
depth = 0
in_string = False
quote_char = ''
escape = False
for j in range(i, n):
c = text[j]
if escape:
escape = False
continue
if c == '\\':
escape = True
continue
if in_string:
if c == quote_char:
in_string = False
else:
if c in ('"', "'"):
in_string = True
quote_char = c
elif c == '{':
depth += 1
elif c == '}':
depth -= 1
if depth == 0:
results.append((start, j + 1, text[start:j + 1]))
i = j
break
i += 1
return results
def extract_action_from_response(text: str) -> tuple[str, dict | None]:
"""
Extracts structured action JSON block from the bot's response with multi-format resilience.
Supports:
- ```action { ... } ``` or ```json { ... } ``` or ```tool { ... } ```
- Bare balanced JSON objects: {"action": "...", ...}
- Single quotes / Python dicts: {'action': '...', ...}
- XML tags: ..., , ...
- Regex field fallback for unescaped newlines/quotes inside string values.
Returns (cleaned_text, action_dict).
"""
if not text:
return "", None
# Strategy 1: Fenced code block (```action ... ``` or ```json ... ``` or ``` ... ```)
fenced_matches = list(re.finditer(r"```(?:action|json|tool)?\s*([\s\S]*?)\s*```", text, flags=re.IGNORECASE))
for m in reversed(fenced_matches):
block_content = m.group(1).strip()
parsed = _parse_action_json_or_dict(block_content)
if parsed and _is_valid_action_dict(parsed):
cleaned_text = (text[:m.start()] + text[m.end():]).strip()
return cleaned_text, _normalize_action_dict(parsed)
# Strategy 2: XML-style tags ( ... or ... or )
xml_matches = list(re.finditer(r"<(?:action|tool|tool_call)(?:\s+([^>]*?))?>([\s\S]*?)", text, flags=re.IGNORECASE))
for m in reversed(xml_matches):
tag_attrs = m.group(1) or ""
inner_content = m.group(2).strip()
if inner_content:
parsed = _parse_action_json_or_dict(inner_content)
if parsed and _is_valid_action_dict(parsed):
cleaned_text = (text[:m.start()] + text[m.end():]).strip()
return cleaned_text, _normalize_action_dict(parsed)
if tag_attrs:
attr_dict = {}
for k, v in re.findall(r'(\w+)=["\']([^"\']*)["\']', tag_attrs):
attr_dict[k] = v
if _is_valid_action_dict(attr_dict):
cleaned_text = (text[:m.start()] + text[m.end():]).strip()
return cleaned_text, _normalize_action_dict(attr_dict)
# Strategy 3: Self-closing XML tag ()
self_closing = re.search(r'<(?:action|tool)\s+([^>]+?)\s*/>', text, flags=re.IGNORECASE)
if self_closing:
attr_dict = {}
for k, v in re.findall(r'(\w+)=["\']([^"\']*)["\']', self_closing.group(1)):
attr_dict[k] = v
if _is_valid_action_dict(attr_dict):
cleaned_text = (text[:self_closing.start()] + text[self_closing.end():]).strip()
return cleaned_text, _normalize_action_dict(attr_dict)
# Strategy 4: Bare balanced JSON or dict object (handles nested braces, f-strings, dicts inside content)
balanced_blocks = _extract_balanced_json_objects(text)
for start_pos, end_pos, block_str in reversed(balanced_blocks):
parsed = _parse_action_json_or_dict(block_str)
if parsed and _is_valid_action_dict(parsed):
cleaned_text = (text[:start_pos] + text[end_pos:]).strip()
return cleaned_text, _normalize_action_dict(parsed)
return text.strip(), None
# ============================================================================
# 5. STEP-BY-STEP WORKER & GRANITE 4.1:3b DECISION ENGINE
# ============================================================================
class MagiBotWorker(QThread):
"""
Executes a single conversational response or workspace step for the active bot.
- If pure conversation: writes message directly without tool popup or Granite looping.
- If action found: requests user confirmation, executes action, then queries granite4.1:3b
in the background to evaluate the next step.
"""
chunk_received = pyqtSignal(str)
step_finished = pyqtSignal(dict) # result payload
confirmation_requested = pyqtSignal(dict) # action payload for popup
error_occurred = pyqtSignal(str)
def __init__(
self,
bot_info: dict,
user_original_prompt: str,
current_step_prompt: str,
step_number: int,
gemini_key: str = "",
mistral_key: str = "",
parent=None
):
super().__init__(parent)
self.bot_info = bot_info
self.user_original_prompt = user_original_prompt
self.current_step_prompt = current_step_prompt
self.step_number = step_number
self.gemini_key = gemini_key
self.mistral_key = mistral_key
self._is_cancelled = False
# Confirmation sync
self._confirmation_event = threading.Event()
self._confirmation_approved = False
def cancel(self):
self._is_cancelled = True
self._confirmation_approved = False
self._confirmation_event.set()
def set_confirmation_response(self, approved: bool):
self._confirmation_approved = approved
self._confirmation_event.set()
def run(self):
try:
# 1. Fetch current workspace files
ws_dir = self.bot_info.get("workspace_dir", get_bot_workspace_dir(self.bot_info.get("name", "Bot")))
_, ws_files = BotWorkspaceTools.list_files(ws_dir)
# 2. Build system instructions
custom_sys = self.bot_info.get("custom_system_prompt", "").strip()
tool_sys = ACTION_SYSTEM_INSTRUCTION.format(
workspace_path=ws_dir,
workspace_files=ws_files
)
full_system = f"{custom_sys}\n\n{tool_sys}".strip()
prompt_content = self.current_step_prompt
# 3. Build messages list
messages = [
{"role": "system", "content": full_system},
{"role": "user", "content": prompt_content}
]
# 4. Call local Ollama model (>= 7B)
model_name = self.bot_info.get("model", "deepcoder:14b")
raw_response = self.call_model(model_name, messages)
if self._is_cancelled:
return
# 5. Extract text and action
clean_text, action = extract_action_from_response(raw_response)
action_result = None
action_status = "none"
# Check if action is a real workspace tool action
if action and action.get("action") not in ("none", "", "done"):
act_name = action.get("action")
# Confirm only when a move would take a file outside Magi(name).
if BotWorkspaceTools.move_leaves_workspace(ws_dir, action):
self._confirmation_event.clear()
self.confirmation_requested.emit({
"bot_name": self.bot_info.get("name", "Bot"),
"workspace_dir": ws_dir,
"action": action,
"step_number": self.step_number
})
# Wait for user input in GUI thread
self._confirmation_event.wait()
if self._is_cancelled:
return
if self._confirmation_approved:
# Execute confirmed move action
success, res_str = self.execute_action(ws_dir, action)
action_status = "success" if success else "failed"
action_result = res_str
else:
action_status = "rejected"
action_result = "Move action was cancelled / rejected by the user."
elif act_name == "ask_helper":
# Bot-to-bot collaboration: ask helper model (default: lfm2.5:8b)
helper_model = action.get("model") or "lfm2.5:8b"
question = action.get("question", "")
bot_name = self.bot_info.get("name", "Bot")
helper_system = (
f"You are an expert AI assistant and collaborator (model: {helper_model}). "
f"Another bot named '{bot_name}' is working in the workspace '{ws_dir}' and is asking for your assistance.\n\n"
f"Current Workspace Files:\n{ws_files}\n\n"
f"Original User Goal:\n{self.user_original_prompt}\n\n"
f"YOUR CAPABILITIES:\n"
f"1. Explain the solution, debug the issue, or provide code/advice in your text response.\n"
f"2. You can also take ONE direct workspace action if needed (create_file, edit_file, delete_file, execute_file, install_package, move_file).\n"
f"If taking an action, append ONE action JSON block at the end:\n"
f"```action\n"
f'{{"action": "create_file", "path": "helper_script.py", "content": "..."}}\n'
f"```\n"
f"CRITICAL: Never use emojis anywhere in your response."
)
helper_messages = [
{"role": "system", "content": helper_system},
{"role": "user", "content": f"Bot '{bot_name}' asks for your help:\n{question}"}
]
try:
raw_helper_res = self.call_model(helper_model, helper_messages)
h_text, h_action = extract_action_from_response(raw_helper_res)
if h_action and h_action.get("action") not in ("none", "", "done"):
h_act_name = h_action.get("action")
if BotWorkspaceTools.move_leaves_workspace(ws_dir, h_action):
self._confirmation_event.clear()
self.confirmation_requested.emit({
"bot_name": f"{bot_name} via {helper_model}",
"workspace_dir": ws_dir,
"action": h_action,
"step_number": self.step_number
})
self._confirmation_event.wait()
if self._is_cancelled:
return
if self._confirmation_approved:
h_ok, h_res = self.execute_action(ws_dir, h_action)
else:
h_ok, h_res = False, "Helper move action was rejected by the user."
else:
h_ok, h_res = self.execute_action(ws_dir, h_action)
action_status = "success" if h_ok else "failed"
action_result = f"Helper ({helper_model}) Response:\n{h_text}\n\nHelper Action [{h_act_name}]: {h_res}"
else:
action_status = "success"
action_result = f"Helper ({helper_model}) Advice:\n{h_text if h_text else raw_helper_res}"
except Exception as e:
action_status = "failed"
action_result = f"Failed to consult helper ({helper_model}): {str(e)}"
else:
# All other actions (create_file, edit_file, delete_file, execute_file, install_package, read_file, list_files)
# execute automatically and autonomously without popup!
success, res_str = self.execute_action(ws_dir, action)
action_status = "success" if success else "failed"
action_result = res_str
# 7. Action was executed: Query granite4.1:3b in the background for next step decision
next_step_info = self.query_granite_decision(
ws_dir,
self.user_original_prompt,
action,
action_status,
action_result
)
is_completed = next_step_info.get("is_completed", True)
next_prompt = next_step_info.get("next_prompt", "")
granite_summary = next_step_info.get("summary", "")
else:
# Conversational response / no workspace action needed
action = None
action_status = "conversational"
is_completed = True
next_prompt = ""
granite_summary = ""
if self._is_cancelled:
return
# Final bot text fallback
final_bot_text = clean_text if clean_text else (raw_response if not action else "Action executed.")
# Emit final step payload
payload = {
"step_number": self.step_number,
"bot_text": final_bot_text,
"action": action,
"action_status": action_status,
"action_result": action_result,
"is_task_completed": is_completed,
"next_prompt": next_prompt,
"granite_summary": granite_summary
}
self.step_finished.emit(payload)
except Exception as e:
if not self._is_cancelled:
self.error_occurred.emit(f"Error in step execution: {str(e)}")
def call_model(self, model_name: str, messages: list) -> str:
return self.call_ollama_model(model_name, messages)
def call_ollama_model(self, model_name: str, messages: list) -> str:
clean_model_name = model_name.replace(" (Cloud)", "").strip()
# Sanitize messages list to prevent 400 Bad Request
clean_msgs = []
for m in messages:
role = str(m.get("role", "user")).strip().lower()
if role not in ("system", "user", "assistant"):
role = "user"
content = str(m.get("content", ""))
if not content.strip():
content = " "
clean_msgs.append({"role": role, "content": content})
options = {"temperature": 0.3}
# Enforce 32k context for Qwen2.5-Coder
if any(k in clean_model_name.lower() for k in ("qwen2.5-coder", "qwen", "coder")):
options["num_ctx"] = 32768
elif any(k in clean_model_name.lower() for k in ("deepseek-r1", "lfm")):
options["num_ctx"] = 16384
url = f"{OLLAMA_BASE_URL}/api/chat"
payload = {
"model": clean_model_name,
"messages": clean_msgs,
"stream": True,
"options": options
}
full_text = []
try:
resp = requests.post(url, json=payload, stream=True, timeout=120)
if resp.status_code != 200:
err_text = ""
try:
err_text = resp.text
err_json = resp.json()
err_text = err_json.get("error", err_text)
except Exception:
pass
raise RuntimeError(f"Ollama call ({clean_model_name}) failed [{resp.status_code}]: {err_text}")
for line in resp.iter_lines():
if self._is_cancelled:
break
if line:
chunk = json.loads(line.decode("utf-8"))
content = chunk.get("message", {}).get("content", "")
if content:
full_text.append(content)
self.chunk_received.emit(content)
except Exception as e:
raise RuntimeError(f"Ollama call ({clean_model_name}) failed: {e}")
return "".join(full_text)
def call_cloud_model(self, provider_or_model: str, messages: list) -> str:
raise RuntimeError("Magi Bot is local-only. Use an Ollama model (>= 7B) at localhost:11434.")
def execute_action(self, workspace_dir: str, action: dict) -> tuple[bool, str]:
raw_act = str(action.get("action", "")).strip().lower()
# Web fetch aliases (web_fetch, fetch_web, fetch_url, read_url, browse_url, scrape_url, get_url)
if raw_act in ("fetch_web", "web_fetch", "fetch_url", "read_url", "browse_url", "scrape_url", "get_url", "fetch_webpage"):
target_url = action.get("url") or action.get("link") or action.get("target") or action.get("path") or ""
return BotWorkspaceTools.fetch_web(target_url)
# Web search aliases
elif raw_act in ("web_search", "search_web", "search", "google_search", "bing_search", "websearch"):
query = action.get("query") or action.get("search") or action.get("text") or action.get("prompt") or ""
return BotWorkspaceTools.web_search(query)
# YouTube transcript aliases
elif raw_act in ("get_youtube_transcript", "fetch_youtube_transcript", "youtube_transcript", "video_transcript"):
yt_url = action.get("url") or action.get("video_id") or action.get("link") or ""
return BotWorkspaceTools.get_youtube_transcript(yt_url)
# File operations
elif raw_act in ("create_file", "write_file", "new_file", "make_file"):
path = action.get("path") or action.get("file") or action.get("filename") or ""
content = action.get("content") if "content" in action else (action.get("code") or action.get("text") or "")
return BotWorkspaceTools.create_file(workspace_dir, path, content)
elif raw_act in ("create_folder", "make_folder", "mkdir", "create_dir"):
path = action.get("path") or action.get("folder") or action.get("name") or ""
return BotWorkspaceTools.create_folder(workspace_dir, path)
elif raw_act in ("edit_file", "modify_file", "update_file"):
path = action.get("path") or action.get("file") or action.get("filename") or ""
content = action.get("content") if "content" in action else (action.get("code") or action.get("text") or "")
return BotWorkspaceTools.edit_file(
workspace_dir,
path,
content,
action.get("mode", "overwrite"),
action.get("target", ""),
action.get("replacement", "")
)
elif raw_act in ("delete_file", "remove_file", "rm_file", "delete_folder"):
path = action.get("path") or action.get("file") or action.get("filename") or ""
return BotWorkspaceTools.delete_file(workspace_dir, path)
elif raw_act in ("move_file", "rename_file", "mv"):
src = action.get("source") or action.get("src") or action.get("from") or ""
dst = action.get("destination") or action.get("dst") or action.get("to") or ""
return BotWorkspaceTools.move_file(workspace_dir, src, dst)
elif raw_act in ("read_file", "view_file", "cat_file", "inspect_file", "get_file"):
path = action.get("path") or action.get("file") or action.get("filename") or ""
return BotWorkspaceTools.read_file(workspace_dir, path)
elif raw_act in ("execute_file", "run_file", "run_script", "exec_file", "execute_script"):
path = action.get("path") or action.get("file") or action.get("script") or ""
args = action.get("args", [])
return BotWorkspaceTools.execute_file(workspace_dir, path, args)
elif raw_act in ("install_package", "pip_install", "install"):
pkg = action.get("package") or action.get("name") or action.get("pkg") or ""
return BotWorkspaceTools.install_package(workspace_dir, pkg)
elif raw_act in ("launch_headless_app", "launch_app", "run_headless"):
path = action.get("path") or action.get("file") or ""
args = action.get("args", [])
return BotWorkspaceTools.launch_headless_app(workspace_dir, path, args)
elif raw_act in ("inspect_headless_app", "inspect_app"):
app_id = action.get("app_id") or action.get("id", "")
return BotWorkspaceTools.inspect_headless_app(app_id)
elif raw_act in ("interact_headless_app", "interact_app"):
app_id = action.get("app_id") or action.get("id", "")
return BotWorkspaceTools.interact_headless_app(
app_id,
action.get("action_type", "click"),
action.get("target", ""),
action.get("text", ""),
int(action.get("x", 0)),
int(action.get("y", 0))
)
elif raw_act in ("close_headless_app", "close_app"):
app_id = action.get("app_id") or action.get("id", "")
return BotWorkspaceTools.close_headless_app(app_id)
elif raw_act in ("list_files", "ls", "dir"):
path = action.get("path", ".")
return BotWorkspaceTools.list_files(workspace_dir, path)
elif raw_act in ("done", "finish", "complete", "task_done", "finished"):
summary = action.get("summary") or action.get("text") or "Task completed."
return True, summary
elif raw_act in ("ask_helper", "ask_bot", "consult_helper"):
q = action.get("question") or action.get("prompt") or action.get("text") or ""
target_model = action.get("model", "lfm2.5:8b")
return BotWorkspaceTools.ask_helper_bot(q, target_model)
else:
return False, f"Unknown action: {raw_act}"
def query_granite_decision(
self,
workspace_dir: str,
original_user_prompt: str,
last_action: dict | None,
last_status: str,
last_result: str | None
) -> dict:
"""
Uses granite4.1:3b in the background to evaluate the user's task and previous action,
determining if more steps are needed and formulating the next prompt.
"""
if not last_action or last_status in ("rejected", "conversational", "none"):
return {
"is_completed": True,
"next_prompt": "",
"summary": "Task processing finished."
}
# List updated workspace files
_, ws_files = BotWorkspaceTools.list_files(workspace_dir)
granite_system = (
"You are an intelligent task planner. You decide what single step an AI bot must perform next "
"to fulfill the user's goal, or decide that the task is finished. "
"Respond in pure valid JSON without markdown formatting or backticks. NEVER use emojis."
)
granite_user = f"""
Original User Goal:
{original_user_prompt}
Last Action Executed:
{json.dumps(last_action, ensure_ascii=False)}
Action Execution Status: {last_status}
Action Result Output:
{last_result if last_result else 'None'}
Current Workspace Files:
{ws_files}
Instructions:
1. Did the last action fulfill the entire user goal? (is_completed: true/false)
- SPECIAL RULE FOR RETRIEVAL & INSPECTION ACTIONS (read_file, web_search, fetch_web, web_fetch, get_youtube_transcript, inspect_headless_app):
If the last action was "read_file", "web_search", "fetch_web", "web_fetch", "get_youtube_transcript", or "inspect_headless_app", data has just been retrieved in "Action Result Output".
The user's goal is NOT completed until the bot explains, presents, or summarizes this content to answer the user's request.
In this case, is_completed MUST be false, and next_step MUST be:
"Present, explain, analyze, or summarize the retrieved content or UI state to completely answer the user's question."
- SPECIAL RULE FOR HEADLESS APP LAUNCH (launch_headless_app):
If the last action was "launch_headless_app", next_step should usually be:
"Inspect the background app UI and outputs using inspect_headless_app."
2. If NO, describe the EXACT single next action the bot should do.
3. Formulate the JSON response:
{{
"is_completed": false,
"next_step": "Precise instruction for the next single action",
"summary": "Brief status update"
}}
"""
try:
resp = requests.post(
f"{OLLAMA_BASE_URL}/api/chat",
json={
"model": get_solo_coordinator_model(),
"messages": [
{"role": "system", "content": granite_system},
{"role": "user", "content": granite_user}
],
"stream": False,
"options": {"temperature": 0.1}
},
timeout=30
)
raw = resp.json().get("message", {}).get("content", "").strip()
# Clean possible markdown json fences
raw_clean = re.sub(r"^```(?:json)?\s*", "", raw, flags=re.MULTILINE)
raw_clean = re.sub(r"\s*```$", "", raw_clean, flags=re.MULTILINE).strip()
data = json.loads(raw_clean)
is_completed = data.get("is_completed", True)
next_step = data.get("next_step", "")
summary = data.get("summary", "")
if not is_completed and next_step:
# Step 4 Requirement: next prompt combines user original prompt + last action + next action
formulated_prompt = (
f"User Goal: {original_user_prompt}\n"
f"Previous Action: {json.dumps(last_action)} (Result: {last_result})\n"
f"Next Step to perform: {next_step}\n\n"
f"Please perform this single step now."
)
return {
"is_completed": False,
"next_prompt": formulated_prompt,
"summary": summary
}
else:
return {
"is_completed": True,
"next_prompt": "",
"summary": summary if summary else "All steps completed successfully."
}
except Exception as e:
print(f"[MagiBotWorker] Granite decision error: {e}")
# Fallback completion
return {
"is_completed": True,
"next_prompt": "",
"summary": "Task completed."
}
# ============================================================================
# 5B. MAGI BOT TEAM — MULTI-BOT COLLABORATIVE SWARM ENGINE
# ============================================================================
class TeamBoard:
"""
Shared blackboard & coordination state for Magi Bot TEAM.
Maintains:
- Global team goal
- Subtasks queue with status & assigned bot
- File locking registry (prevents simultaneous or duplicate edits)
- Completed actions history (anti-duplication)
- Shared knowledge & notes
"""
def __init__(self, goal: str, workspace_dir: str, bots: list[dict]):
self.goal = goal
self.workspace_dir = workspace_dir
self.bots = bots # Up to 3 bots (id, name, model, role, color)
self.tasks: list[dict] = [] # [{"id": 1, "title": "...", "assigned_to": "BotA", "status": "pending|in_progress|completed", "result": ""}]
self.locked_files: dict[str, str] = {} # {"main.py": "BotName"}
self.completed_actions: list[dict] = []
self.shared_notes: list[str] = []
self.activity_log: list[dict] = []
self.lock = threading.Lock()
def set_tasks(self, task_titles: list[dict]):
with self.lock:
self.tasks = []
for idx, t in enumerate(task_titles, 1):
self.tasks.append({
"id": idx,
"title": t.get("title", f"Task {idx}"),
"assigned_to": t.get("assigned_to", ""),
"status": "pending",
"result": ""
})
def add_task(self, title: str, assigned_to: str = "") -> dict:
with self.lock:
new_id = len(self.tasks) + 1
t = {
"id": new_id,
"title": title,
"assigned_to": assigned_to,
"status": "pending",
"result": ""
}
self.tasks.append(t)
return t
def get_pending_task_for_bot(self, bot_name: str) -> dict | None:
with self.lock:
# 1. First priority: tasks assigned directly to this bot
for t in self.tasks:
if t["status"] == "pending" and t["assigned_to"] == bot_name:
return t
# 2. Second priority: unassigned pending tasks
for t in self.tasks:
if t["status"] == "pending" and not t["assigned_to"]:
return t
# Never steal other bots' specialized tasks!
return None
def claim_task(self, bot_name: str, task_id: int) -> bool:
with self.lock:
for t in self.tasks:
if t["id"] == task_id:
if t["status"] == "pending" or t["assigned_to"] == bot_name:
t["status"] = "in_progress"
t["assigned_to"] = bot_name
return True
return False
def complete_task(self, task_id: int, result: str):
with self.lock:
for t in self.tasks:
if t["id"] == task_id:
t["status"] = "completed"
t["result"] = result
break
def is_all_completed(self) -> bool:
with self.lock:
if not self.tasks:
return True
return all(t["status"] == "completed" for t in self.tasks)
def lock_file(self, bot_name: str, file_path: str) -> tuple[bool, str]:
clean_file = os.path.basename(file_path).strip().lower()
with self.lock:
if clean_file in self.locked_files:
owner = self.locked_files[clean_file]
if owner != bot_name:
return False, f"File '{clean_file}' is currently locked by {owner}. Choose a different task or file."
self.locked_files[clean_file] = bot_name
return True, f"File '{clean_file}' locked by {bot_name}."
def unlock_file(self, bot_name: str, file_path: str):
clean_file = os.path.basename(file_path).strip().lower()
with self.lock:
if self.locked_files.get(clean_file) == bot_name:
del self.locked_files[clean_file]
def add_shared_note(self, note: str):
with self.lock:
if note and note not in self.shared_notes:
self.shared_notes.append(note)
def log_activity(self, bot_name: str, bot_color: str, action: str, details: str):
with self.lock:
self.activity_log.append({
"timestamp": datetime.datetime.now().strftime("%H:%M:%S"),
"bot": bot_name,
"color": bot_color,
"action": action,
"details": details
})
def get_summary_text(self) -> str:
with self.lock:
lines = [f"TEAM GOAL: {self.goal}\n", "LIVE TEAM TASK BOARD & ARTIFACTS:"]
for t in self.tasks:
status_tag = f"[{t['status'].upper()}]"
assignee = f"(Assigned: {t['assigned_to']})" if t['assigned_to'] else "(Unassigned)"
lines.append(f"{t['id']}. {status_tag} {t['title']} {assignee}")
if t["result"]:
lines.append(f" Completed Result ({t.get('assigned_to', 'Teammate')}): {t['result'][:400]}")
if self.locked_files:
lines.append("\nLOCKED FILES (In Use): " + ", ".join(f"{f} ({b})" for f, b in self.locked_files.items()))
if self.shared_notes:
lines.append("\nSHARED TEAM NOTES:\n- " + "\n- ".join(self.shared_notes[-5:]))
return "\n".join(lines)
class MagiBotTeamCoordinator:
"""
Orchestrates initial multi-bot task decomposition using LFM2.5:8b.
Tasks are generated ONCE at the start and remain fixed on the board until all are completed.
"""
@staticmethod
def decompose_team_goal(goal: str, workspace_dir: str, bots: list[dict]) -> list[dict]:
_, ws_files = BotWorkspaceTools.list_files(workspace_dir)
bot_descriptions = "\n".join(
f"- Bot '{b.get('name')}' (Role: {b.get('role', 'Developer')}, Model: {b.get('model')}, Quota: {b.get('max_actions_per_turn', 1)} actions/turn)"
for b in bots
)
bot_names_list = [b.get('name') for b in bots if b.get('name')]
total_actions = sum(int(b.get("max_actions_per_turn", 1)) for b in bots)
target_tasks = min(max(total_actions, len(bots)), 8)
sys_prompt = (
"You are the Team Coordinator (model: LFM2.5:8b). Decompose the user's software/engineering goal into a "
f"structured, sequential list of {target_tasks} distinct, non-overlapping subtasks assigned STRICTLY to the team bots based on their specialized roles.\n\n"
f"AVAILABLE TEAM BOTS (assign ONLY to these exact names: {', '.join(bot_names_list)}):\n"
f"{bot_descriptions}\n\n"
"STRICT ROLE SPECIALIZATION RULES:\n"
"1. 'Architect' / 'Lead' / 'Designer' roles MUST receive: System architecture, modular blueprints, data schemas, README.md or project structure.\n"
"2. 'Developer' / 'Coder' / 'Frontend' / 'Backend' roles MUST receive: Implementation of application logic, main components, classes, and UI modules.\n"
"3. 'Tester' / 'QA' / 'Verification' / 'Reviewer' roles MUST receive: Writing test scripts (test_*.py), executing code in .venv with execute_file, verifying output, and bug fixing.\n"
"4. 'Researcher' / 'Web Analyst' roles MUST receive: Online research and documentation retrieval with web_search/fetch_web.\n\n"
"ANTI-DUPLICATION & PIPELINE RULES:\n"
"- Subtasks must follow a sequential engineering pipeline: Architecture -> Implementation -> Verification/Testing.\n"
"- DO NOT assign the same file or duplicate work to multiple bots.\n"
"- 'assigned_to' MUST match the EXACT name of one of the available bots in the roster.\n"
"- Each task title must name the concrete file or script to create/edit/test (e.g. 'Create data models in models.py', 'Implement game loop in main.py', 'Execute test_suite.py in .venv and verify results').\n\n"
"Output pure valid JSON list ONLY matching this schema:\n"
"[\n"
" {\"title\": \"Concrete task description with filename\", \"assigned_to\": \"ExactBotName\"}\n"
"]\n"
"NEVER include markdown backticks, conversational commentary, or emojis."
)
user_prompt = f"""User Goal:
{goal}
Current Workspace Files:
{ws_files}
Generate the complete JSON subtasks list for the team now:"""
coord_model = get_team_coordinator_model()
try:
resp = requests.post(
f"{OLLAMA_BASE_URL}/api/chat",
json={
"model": coord_model,
"messages": [
{"role": "system", "content": sys_prompt},
{"role": "user", "content": user_prompt}
],
"stream": False,
"options": {"temperature": 0.2}
},
timeout=100
)
raw = resp.json().get("message", {}).get("content", "").strip()
# Strip reasoning tags if present
raw_clean = re.sub(r".*?", "", raw, flags=re.DOTALL).strip()
raw_clean = re.sub(r"^```(?:json)?\s*", "", raw_clean, flags=re.MULTILINE)
raw_clean = re.sub(r"\s*```$", "", raw_clean, flags=re.MULTILINE).strip()
tasks = json.loads(raw_clean)
if isinstance(tasks, list) and tasks:
return tasks
except Exception as e:
print(f"[MagiBotTeamCoordinator] Task decomposition error with {coord_model}: {e}")
# Smart, role-aware fallback decomposition
fallback_tasks = []
for b in bots:
b_name = b.get("name", "Bot")
b_role = b.get("role", "Developer").lower()
quota = int(b.get("max_actions_per_turn", 1))
for q_idx in range(1, quota + 1):
if any(k in b_role for k in ("architect", "lead", "designer", "plan")):
t_title = f"Establish architecture & modular blueprints (Step {q_idx}) for {goal[:40]}" if q_idx == 1 else f"Define data schemas & project structure (Step {q_idx})"
elif any(k in b_role for k in ("coder", "dev", "frontend", "backend", "program")):
t_title = f"Implement core logic & main components in main.py (Step {q_idx})" if q_idx == 1 else f"Implement supporting modules & handlers (Step {q_idx})"
elif any(k in b_role for k in ("test", "qa", "verify", "review", "debug")):
t_title = f"Write verification test suite in test_suite.py (Step {q_idx})" if q_idx == 1 else f"Execute test_suite.py in .venv and verify runtime outputs (Step {q_idx})"
elif any(k in b_role for k in ("research", "web", "analyst")):
t_title = f"Research documentation & dependencies with web_search (Step {q_idx})" if q_idx == 1 else f"Fetch required references with fetch_web (Step {q_idx})"
else:
t_title = f"Implement assigned task component (Step {q_idx})"
fallback_tasks.append({"title": t_title, "assigned_to": b_name})
return fallback_tasks
class MagiBotTeamWorker(QThread):
"""
Executes collaborative rounds for up to 3 specialized bots in the same folder.
Coordinates task claiming, multi-action turns per bot, file locking, and anti-duplication.
Tasks are fixed on the board from the start.
"""
team_status_updated = pyqtSignal(str)
bot_status_updated = pyqtSignal(str, str) # (bot_name, status_text)
bot_message_posted = pyqtSignal(dict) # bubble payload
bot_chunk_received = pyqtSignal(str)
task_board_updated = pyqtSignal(dict) # board state summary
team_finished = pyqtSignal(str) # final summary
error_occurred = pyqtSignal(str)
MAX_ROUNDS = 12
def __init__(
self,
team_bots: list[dict],
goal: str,
workspace_dir: str,
gemini_key: str = "",
mistral_key: str = "",
parent=None
):
super().__init__(parent)
self.team_bots = team_bots[:3] # Max 3 bots
self.goal = goal
self.workspace_dir = workspace_dir
self.gemini_key = gemini_key
self.mistral_key = mistral_key
self._is_cancelled = False
self.board = TeamBoard(goal, workspace_dir, self.team_bots)
def cancel(self):
self._is_cancelled = True
def run(self):
try:
coord_model = get_team_coordinator_model()
self.team_status_updated.emit(f"Magi Bot TEAM: Decomposing goal with {coord_model}...")
# Step 1: Decompose tasks once at start
tasks = MagiBotTeamCoordinator.decompose_team_goal(self.goal, self.workspace_dir, self.team_bots)
self.board.set_tasks(tasks)
self.task_board_updated.emit({
"tasks": self.board.tasks,
"locked_files": self.board.locked_files,
"summary": self.board.get_summary_text()
})
# Announce team formation in chat
team_intro = (
f"**Magi Bot TEAM Activated** ({len(self.team_bots)} Specialized Bots)\n\n"
f"**Backend Coordinator:** `{coord_model}`\n"
f"**Shared Workspace:** `{self.workspace_dir}`\n\n"
f"**Team Roster & Action Quotas:**\n"
)
for b in self.team_bots:
team_intro += f"- **{b.get('name')}** — *{b.get('role', 'Developer')}* ({b.get('model')}) • **{b.get('max_actions_per_turn', 1)} actions/turn**\n"
team_intro += f"\n**Decomposed {len(tasks)} Subtasks onto the Live Board.**"
self.bot_message_posted.emit({
"sender_name": "Team Coordinator",
"text": team_intro,
"is_user": False,
"role": "Coordinator",
"color": "#6366f1",
"action": None,
"action_result": None
})
# Step 2: Collaborative execution loop
round_idx = 0
while round_idx < self.MAX_ROUNDS and not self._is_cancelled:
round_idx += 1
any_task_executed = False
for bot in self.team_bots:
if self._is_cancelled:
break
bot_name = bot.get("name", "Bot")
bot_role = bot.get("role", "Developer")
bot_color = bot.get("color", "#007aff")
bot_model = bot.get("model", "deepcoder:14b")
max_actions = int(bot.get("max_actions_per_turn", 1))
actions_done_this_turn = 0
while actions_done_this_turn < max_actions and not self._is_cancelled:
action_num = actions_done_this_turn + 1
# Find pending task on the fixed board
pending_task = self.board.get_pending_task_for_bot(bot_name)
if not pending_task:
# No more pending tasks left for this bot's role! Turn finishes cleanly.
break
task_id = pending_task["id"]
self.board.claim_task(bot_name, task_id)
turn_action_str = f"({action_num}/{max_actions})" if max_actions > 1 else ""
self.bot_status_updated.emit(bot_name, f"Working on Task #{task_id} {turn_action_str}: {pending_task['title'][:30]}...")
self.task_board_updated.emit({
"tasks": self.board.tasks,
"locked_files": self.board.locked_files,
"summary": self.board.get_summary_text()
})
# Build prompt for bot
_, ws_files = BotWorkspaceTools.list_files(self.workspace_dir)
board_summary = self.board.get_summary_text()
custom_sys = bot.get("custom_system_prompt", "").strip()
team_system_prompt = f"""You are '{bot_name}', a specialized AI engineer in the Magi Bot TEAM.
Your Assigned Role: {bot_role}
Dedicated Shared Workspace: {self.workspace_dir}
{board_summary}
Current Files in Shared Workspace:
{ws_files}
COLLABORATION & ANTI-DUPLICATION DIRECTIVES:
1. STRICT ROLE DISCIPLINE: You are acting as '{bot_role}'. Only perform actions that directly advance your role and the specific subtask you claimed.
2. PREVENT DUPLICATION & CONFLICTS:
- Carefully review the 'LIVE TEAM TASK BOARD & ARTIFACTS' above to see what other bots have ALREADY completed.
- DO NOT re-create, wipe out, or overwrite files already created by your teammates unless your subtask explicitly requires modifying them.
- If writing code that uses modules created in earlier tasks, import or read them with `read_file`.
- If your role is Tester/QA, write test scripts and run `execute_file` to verify the application.
3. SINGLE ACTION PER MESSAGE:
- Output a clear, direct explanation of your step.
- At the VERY END, output EXACTLY ONE action block in ```action ... ``` format.
4. NO EMOJIS: Never use emojis anywhere in your response.
Available action blocks:
- Create file: {{"action": "create_file", "path": "...", "content": "..."}}
- Create folder: {{"action": "create_folder", "path": "..."}}
- Edit file: {{"action": "edit_file", "path": "...", "content": "...", "mode": "overwrite"}}
- Read file: {{"action": "read_file", "path": "..."}}
- Delete file: {{"action": "delete_file", "path": "..."}}
- Move / rename file: {{"action": "move_file", "source": "...", "destination": "..."}}
- Execute file (.venv): {{"action": "execute_file", "path": "...", "args": []}}
- Install package (.venv): {{"action": "install_package", "package": "..."}}
- Web search: {{"action": "web_search", "query": "..."}}
- Fetch webpage (read URL): {{"action": "fetch_web", "url": "https://..."}}
- Get YouTube transcript: {{"action": "get_youtube_transcript", "url": "https://www.youtube.com/watch?v=..."}}
- Launch headless app (.venv): {{"action": "launch_headless_app", "path": "...", "args": []}}
- Inspect headless app: {{"action": "inspect_headless_app", "app_id": "app_1"}}
- Interact headless app: {{"action": "interact_headless_app", "app_id": "app_1", "action_type": "click", "target": "Btn"}}
- Close headless app: {{"action": "close_headless_app", "app_id": "app_1"}}
- Task done: {{"action": "done", "summary": "..."}}
"""
if custom_sys:
team_system_prompt = f"{custom_sys}\n\n{team_system_prompt}"
user_step_prompt = (
f"TEAM TASK ASSIGNMENT for '{bot_name}' ({bot_role}):\n"
f"You have claimed Task #{task_id}: '{pending_task['title']}' (Action {action_num} of {max_actions} for your turn).\n\n"
f"Instructions:\n"
f"1. Fulfill this specific task according to your role ({bot_role}).\n"
f"2. Build upon existing workspace files ({ws_files}) without duplicating existing code.\n"
f"3. Output your explanation and append your action block now."
)
messages = [
{"role": "system", "content": team_system_prompt},
{"role": "user", "content": user_step_prompt}
]
# Call model
self.team_status_updated.emit(f"Bot '{bot_name}' ({bot_role}) thinking with {bot_model} (Action {turn_action_str})...")
raw_resp = self.call_bot_model(bot_model, messages)
if self._is_cancelled:
break
clean_text, action = extract_action_from_response(raw_resp)
action_result = None
target_file = ""
# Check file locking
if action and action.get("action") in ("create_file", "edit_file", "delete_file", "move_file"):
target_file = action.get("path") or action.get("destination") or action.get("source") or ""
if target_file:
locked_ok, lock_msg = self.board.lock_file(bot_name, target_file)
if not locked_ok:
clean_text += f"\n\n*(Notice: {lock_msg})*"
action = None
action_result = lock_msg
# Execute action
if action and action.get("action") not in ("none", "", "done"):
act_name = action.get("action")
self.team_status_updated.emit(f"Bot '{bot_name}' executing {act_name}...")
succ, res_str = self.execute_team_action(action)
action_result = res_str
if target_file:
self.board.unlock_file(bot_name, target_file)
self.board.log_activity(bot_name, bot_color, act_name, res_str[:100])
elif target_file:
self.board.unlock_file(bot_name, target_file)
# Mark task complete
summary_result = action_result if action_result else (clean_text[:120] + "...")
self.board.complete_task(task_id, summary_result)
self.bot_status_updated.emit(bot_name, f"Completed Task #{task_id}")
any_task_executed = True
# Post bot message to chat
self.bot_message_posted.emit({
"sender_name": bot_name,
"text": clean_text,
"is_user": False,
"role": bot_role,
"color": bot_color,
"action": action,
"action_result": action_result
})
self.task_board_updated.emit({
"tasks": self.board.tasks,
"locked_files": self.board.locked_files,
"summary": self.board.get_summary_text()
})
actions_done_this_turn += 1
time.sleep(0.5)
if action and action.get("action") == "done":
break
if self.board.is_all_completed():
break
if self.board.is_all_completed() and not any_task_executed:
break
if self.board.is_all_completed() or not any_task_executed:
break
final_msg = f"All {len(self.board.tasks)} team tasks completed successfully in '{self.workspace_dir}'."
self.team_status_updated.emit(final_msg)
self.team_finished.emit(final_msg)
except Exception as e:
self.error_occurred.emit(f"Team execution error: {str(e)}")
def call_bot_model(self, model_name: str, messages: list) -> str:
clean_model_name = model_name.replace(" (Cloud)", "").strip()
clean_msgs = []
for m in messages:
role = str(m.get("role", "user")).strip().lower()
if role not in ("system", "user", "assistant"):
role = "user"
content = str(m.get("content", ""))
if not content.strip():
content = " "
clean_msgs.append({"role": role, "content": content})
options = {"temperature": 0.3}
if any(k in clean_model_name.lower() for k in ("qwen2.5-coder", "qwen", "coder")):
options["num_ctx"] = 32768
elif any(k in clean_model_name.lower() for k in ("deepseek-r1", "lfm")):
options["num_ctx"] = 16384
url = f"{OLLAMA_BASE_URL}/api/chat"
payload = {"model": clean_model_name, "messages": clean_msgs, "stream": True, "options": options}
full_text = []
resp = requests.post(url, json=payload, stream=True, timeout=120)
if resp.status_code != 200:
err_text = ""
try:
err_text = resp.text
err_json = resp.json()
err_text = err_json.get("error", err_text)
except Exception:
pass
raise RuntimeError(f"Ollama call ({clean_model_name}) failed [{resp.status_code}]: {err_text}")
for line in resp.iter_lines():
if self._is_cancelled:
break
if line:
chunk = json.loads(line.decode("utf-8"))
c = chunk.get("message", {}).get("content", "")
if c:
full_text.append(c)
self.bot_chunk_received.emit(c)
return "".join(full_text)
def execute_team_action(self, action: dict) -> tuple[bool, str]:
raw_act = str(action.get("action", "")).strip().lower()
ws = self.workspace_dir
# Web fetch aliases
if raw_act in ("fetch_web", "web_fetch", "fetch_url", "read_url", "browse_url", "scrape_url", "get_url", "fetch_webpage"):
target_url = action.get("url") or action.get("link") or action.get("target") or action.get("path") or ""
return BotWorkspaceTools.fetch_web(target_url)
# Web search aliases
elif raw_act in ("web_search", "search_web", "search", "google_search", "bing_search", "websearch"):
query = action.get("query") or action.get("search") or action.get("text") or action.get("prompt") or ""
return BotWorkspaceTools.web_search(query)
# YouTube transcript aliases
elif raw_act in ("get_youtube_transcript", "fetch_youtube_transcript", "youtube_transcript", "video_transcript"):
yt_url = action.get("url") or action.get("video_id") or action.get("link") or ""
return BotWorkspaceTools.get_youtube_transcript(yt_url)
# File operations
elif raw_act in ("create_file", "write_file", "new_file", "make_file"):
path = action.get("path") or action.get("file") or action.get("filename") or ""
content = action.get("content") if "content" in action else (action.get("code") or action.get("text") or "")
return BotWorkspaceTools.create_file(ws, path, content)
elif raw_act in ("create_folder", "make_folder", "mkdir", "create_dir"):
path = action.get("path") or action.get("folder") or action.get("name") or ""
return BotWorkspaceTools.create_folder(ws, path)
elif raw_act in ("edit_file", "modify_file", "update_file"):
path = action.get("path") or action.get("file") or action.get("filename") or ""
content = action.get("content") if "content" in action else (action.get("code") or action.get("text") or "")
return BotWorkspaceTools.edit_file(
ws,
path,
content,
action.get("mode", "overwrite"),
action.get("target", ""),
action.get("replacement", "")
)
elif raw_act in ("delete_file", "remove_file", "rm_file", "delete_folder"):
path = action.get("path") or action.get("file") or action.get("filename") or ""
return BotWorkspaceTools.delete_file(ws, path)
elif raw_act in ("move_file", "rename_file", "mv"):
src = action.get("source") or action.get("src") or action.get("from") or ""
dst = action.get("destination") or action.get("dst") or action.get("to") or ""
if BotWorkspaceTools.path_leaves_workspace(ws, dst):
return False, "Moving a file outside the workspace needs a user confirm dialog. Team mode skipped this move."
return BotWorkspaceTools.move_file(ws, src, dst)
elif raw_act in ("read_file", "view_file", "cat_file", "inspect_file", "get_file"):
path = action.get("path") or action.get("file") or action.get("filename") or ""
return BotWorkspaceTools.read_file(ws, path)
elif raw_act in ("execute_file", "run_file", "run_script", "exec_file", "execute_script"):
path = action.get("path") or action.get("file") or action.get("script") or ""
args = action.get("args", [])
return BotWorkspaceTools.execute_file(ws, path, args)
elif raw_act in ("install_package", "pip_install", "install"):
pkg = action.get("package") or action.get("name") or action.get("pkg") or ""
return BotWorkspaceTools.install_package(ws, pkg)
elif raw_act in ("launch_headless_app", "launch_app", "run_headless"):
path = action.get("path") or action.get("file") or ""
args = action.get("args", [])
return BotWorkspaceTools.launch_headless_app(ws, path, args)
elif raw_act in ("inspect_headless_app", "inspect_app"):
app_id = action.get("app_id") or action.get("id", "")
return BotWorkspaceTools.inspect_headless_app(app_id)
elif raw_act in ("interact_headless_app", "interact_app"):
app_id = action.get("app_id") or action.get("id", "")
return BotWorkspaceTools.interact_headless_app(
app_id,
action.get("action_type", "click"),
action.get("target", ""),
action.get("text", ""),
int(action.get("x", 0)),
int(action.get("y", 0))
)
elif raw_act in ("close_headless_app", "close_app"):
app_id = action.get("app_id") or action.get("id", "")
return BotWorkspaceTools.close_headless_app(app_id)
elif raw_act in ("list_files", "ls", "dir"):
path = action.get("path", ".")
return BotWorkspaceTools.list_files(ws, path)
elif raw_act in ("done", "finish", "complete", "task_done", "finished"):
summary = action.get("summary") or action.get("text") or "Task completed."
return True, summary
elif raw_act in ("ask_helper", "ask_bot", "consult_helper"):
q = action.get("question") or action.get("prompt") or action.get("text") or ""
target_model = action.get("model", "lfm2.5:8b")
return BotWorkspaceTools.ask_helper_bot(q, target_model)
else:
return False, f"Unknown action: {raw_act}"
# ============================================================================
# 6. USER CONFIRMATION POPUP MODAL
# ============================================================================
class BotActionConfirmDialog(QDialog):
"""
Modal confirmation dialog for moving a file outside the Magi(name) workspace.
Strictly emoji-free with Apple/VisionOS dark aesthetics.
"""
def __init__(self, bot_name: str, workspace_dir: str, action: dict, step_num: int, parent=None):
super().__init__(parent)
self.bot_name = bot_name
self.workspace_dir = workspace_dir
self.action = action
self.step_num = step_num
self.setWindowTitle("Move File Confirmation" if action.get("action") == "move_file" else "Action Confirmation Required")
self.setFixedSize(540, 360)
self.setModal(True)
self.init_ui()
def init_ui(self):
self.setStyleSheet("""
QDialog {
background-color: #12141a;
border: 1.5px solid #3b4252;
border-radius: 14px;
color: #ffffff;
}
""")
layout = QVBoxLayout(self)
layout.setContentsMargins(24, 22, 24, 22)
layout.setSpacing(14)
# Header Title
title_box = QHBoxLayout()
title_label = QLabel("Move File Permission" if self.action.get("action") == "move_file" else "Permission Request")
title_label.setStyleSheet("font-size: 17px; font-weight: 800; color: #ffffff;")
title_box.addWidget(title_label)
title_box.addStretch()
step_badge = QLabel(f"Step {self.step_num}")
step_badge.setStyleSheet("""
background-color: #1e293b;
color: #93c5fd;
border: 1px solid #3b82f6;
border-radius: 6px;
padding: 3px 8px;
font-size: 11px;
font-weight: 700;
""")
title_box.addWidget(step_badge)
layout.addLayout(title_box)
# Subtitle
if self.action.get("action") == "move_file":
desc = QLabel(f"Bot {self.bot_name} wants to move a file outside its Magi workspace:")
else:
act_type = self.action.get("action", "unknown").upper()
desc = QLabel(f"Bot {self.bot_name} is requesting permission to perform: {act_type}")
desc.setStyleSheet("font-size: 13px; color: #cbd5e1;")
desc.setWordWrap(True)
layout.addWidget(desc)
# Action Details Card
detail_card = QFrame()
detail_card.setStyleSheet("""
QFrame {
background-color: #1a1d26;
border: 1px solid #2d3345;
border-radius: 8px;
padding: 10px;
}
""")
d_layout = QVBoxLayout(detail_card)
d_layout.setContentsMargins(10, 8, 10, 8)
d_layout.setSpacing(6)
# Format details
if self.action.get("action") == "move_file":
path_lbl = QLabel(f"Source File: {self.action.get('source', '')}
Destination Target: {self.action.get('destination', '')}")
d_layout.addWidget(path_lbl)
else:
path_lbl = QLabel(json.dumps(self.action, indent=2))
d_layout.addWidget(path_lbl)
ws_lbl = QLabel(f"Workspace: {self.workspace_dir}")
d_layout.addWidget(ws_lbl)
layout.addWidget(detail_card)
layout.addStretch()
# Action Buttons
btn_box = QHBoxLayout()
btn_box.setSpacing(12)
reject_btn = QPushButton("Cancel Move" if self.action.get("action") == "move_file" else "Deny Action")
reject_btn.setCursor(QCursor(Qt.CursorShape.PointingHandCursor))
reject_btn.setStyleSheet("""
QPushButton {
background-color: #242938;
color: #94a3b8;
border: 1px solid #334155;
border-radius: 8px;
padding: 9px 20px;
font-size: 12.5px;
font-weight: 600;
}
QPushButton:hover {
background-color: #333d52;
color: #ffffff;
}
""")
reject_btn.clicked.connect(self.reject)
approve_btn = QPushButton("Confirm & Move" if self.action.get("action") == "move_file" else "Confirm & Allow")
approve_btn.setCursor(QCursor(Qt.CursorShape.PointingHandCursor))
approve_btn.setStyleSheet("""
QPushButton {
background: qlineargradient(x1:0, y1:0, x2:1, y2:0, stop:0 #007aff, stop:1 #0a84ff);
color: #ffffff;
border: none;
border-radius: 8px;
padding: 9px 24px;
font-size: 12.5px;
font-weight: 700;
}
QPushButton:hover {
background: qlineargradient(x1:0, y1:0, x2:1, y2:0, stop:0 #0062cc, stop:1 #0071e3);
}
""")
approve_btn.clicked.connect(self.accept)
btn_box.addStretch()
btn_box.addWidget(reject_btn)
btn_box.addWidget(approve_btn)
layout.addLayout(btn_box)
# ============================================================================
# 7. iMESSAGE-STYLE CHAT BUBBLE & COMPONENTS
# ============================================================================
class IMessageBubble(QFrame):
"""
Apple iMessage-styled message bubble.
- User: Right-aligned, Apple Blue (#007aff) gradient, asymmetric curve.
- Bot: Left-aligned, dark charcoal (#1f232e), subtle border, sender + role badge.
- Action Card: Displays executed tool result with status chip (if an action was run).
- Strictly emoji-free.
"""
def __init__(self, sender_name: str, text: str, is_user: bool, timestamp: str = "", action: dict = None, action_result: str = None, role: str = "", color: str = "", parent=None):
super().__init__(parent)
self.sender_name = sender_name
self.text = text
self.is_user = is_user
self.timestamp = timestamp or datetime.datetime.now().strftime("%H:%M")
self.action = action
self.action_result = action_result
self.role = role
self.color = color or ("#007aff" if is_user else "#64748b")
self.init_ui()
def init_ui(self):
main_layout = QVBoxLayout(self)
main_layout.setContentsMargins(0, 4, 0, 4)
main_layout.setSpacing(3)
# Sender & Role Header (for bots / team members)
if not self.is_user and self.sender_name:
hdr_row = QHBoxLayout()
hdr_row.setContentsMargins(6, 0, 6, 0)
hdr_row.setSpacing(6)
dot = QLabel("●")
dot.setStyleSheet(f"color: {self.color}; font-size: 10px;")
hdr_row.addWidget(dot)
name_lbl = QLabel(self.sender_name)
name_lbl.setStyleSheet("color: #cbd5e1; font-size: 11.5px; font-weight: 700;")
hdr_row.addWidget(name_lbl)
if self.role:
role_lbl = QLabel(f"[{self.role}]")
role_lbl.setStyleSheet("color: #94a3b8; font-size: 10px; font-weight: 600;")
hdr_row.addWidget(role_lbl)
hdr_row.addStretch()
main_layout.addLayout(hdr_row)
# Bubble Container
bubble_row = QHBoxLayout()
bubble_row.setContentsMargins(0, 0, 0, 0)
bubble_row.setSpacing(0)
bubble_frame = QFrame()
bubble_layout = QVBoxLayout(bubble_frame)
bubble_layout.setContentsMargins(14, 10, 14, 10)
bubble_layout.setSpacing(6)
# Message Text
if self.text:
msg_label = QLabel()
msg_label.setWordWrap(True)
msg_label.setTextInteractionFlags(Qt.TextInteractionFlag.TextSelectableByMouse)
formatted_html = format_markdown_and_latex(self.text)
msg_label.setText(formatted_html)
if self.is_user:
msg_label.setStyleSheet("color: #ffffff; font-size: 13.5px; font-family: -apple-system, 'SF Pro Text', sans-serif; line-height: 1.4;")
else:
msg_label.setStyleSheet("color: #e2e8f0; font-size: 13.5px; font-family: -apple-system, 'SF Pro Text', sans-serif; line-height: 1.4;")
bubble_layout.addWidget(msg_label)
# Action Execution Status Chip/Card (if an action took place)
if self.action:
act_card = self.create_action_card()
bubble_layout.addWidget(act_card)
# Styling
if self.is_user:
bubble_row.addStretch()
bubble_row.addWidget(bubble_frame)
bubble_frame.setStyleSheet("""
QFrame {
background: qlineargradient(x1:0, y1:0, x2:1, y2:1, stop:0 #007aff, stop:1 #0a84ff);
border-radius: 18px;
border-bottom-right-radius: 4px;
margin-left: 80px;
}
""")
else:
bubble_row.addWidget(bubble_frame)
bubble_row.addStretch()
border_col = self.color if self.role else "rgba(255, 255, 255, 0.08)"
bubble_frame.setStyleSheet(f"""
QFrame {{
background-color: #1f232e;
border: 1px solid {border_col};
border-left: 3.5px solid {self.color};
border-radius: 18px;
border-bottom-left-radius: 4px;
margin-right: 80px;
}}
""")
main_layout.addLayout(bubble_row)
# Timestamp
time_row = QHBoxLayout()
time_row.setContentsMargins(8, 0, 8, 0)
time_lbl = QLabel(self.timestamp)
time_lbl.setStyleSheet("color: #64748b; font-size: 10.5px; font-weight: 500;")
if self.is_user:
time_row.addStretch()
time_row.addWidget(time_lbl)
else:
time_row.addWidget(time_lbl)
time_row.addStretch()
main_layout.addLayout(time_row)
def create_action_card(self) -> QFrame:
card = QFrame()
card.setStyleSheet("""
QFrame {
background-color: rgba(0, 0, 0, 0.25);
border: 1px solid rgba(255, 255, 255, 0.12);
border-radius: 8px;
padding: 6px;
}
""")
c_layout = QVBoxLayout(card)
c_layout.setContentsMargins(8, 6, 8, 6)
c_layout.setSpacing(4)
act_type = self.action.get("action", "").upper()
header = QHBoxLayout()
header.setContentsMargins(0, 0, 0, 0)
tag = QLabel(f"ACTION: {act_type}")
tag.setStyleSheet("color: #38bdf8; font-size: 11px; font-weight: 700;")
header.addWidget(tag)
header.addStretch()
if self.action_result:
is_success = "rejected" not in self.action_result.lower() and "failed" not in self.action_result.lower() and "error" not in self.action_result.lower()
status_badge = QLabel("SUCCESS" if is_success else "NOTICE")
status_badge.setStyleSheet(f"""
background-color: {'#064e3b' if is_success else '#451a03'};
color: {'#34d399' if is_success else '#fbbf24'};
border: 1px solid {'#059669' if is_success else '#d97706'};
border-radius: 4px;
padding: 2px 6px;
font-size: 9.5px;
font-weight: 800;
""")
header.addWidget(status_badge)
c_layout.addLayout(header)
# Target Path or details
path_str = self.action.get("path") or (f"{self.action.get('source')} -> {self.action.get('destination')}" if self.action.get("source") else "")
if not path_str and self.action.get("package"):
path_str = f"Package: {self.action.get('package')} (.venv)"
elif not path_str and self.action.get("app_id"):
path_str = f"App ID: {self.action.get('app_id')}" + (f" | Target: {self.action.get('target')}" if self.action.get("target") else "")
elif not path_str and self.action.get("question"):
model_used = self.action.get("model", "lfm2.5:8b")
q_short = self.action.get("question", "")[:80]
path_str = f"Helper: {model_used} | Query: {q_short}"
elif not path_str and self.action.get("query"):
path_str = f"Query: {self.action.get('query')}"
elif not path_str and self.action.get("url"):
path_str = f"URL: {self.action.get('url')}"
if path_str:
p_lbl = QLabel(f"Target: {path_str}")
p_lbl.setStyleSheet("color: #94a3b8; font-size: 11px;")
c_layout.addWidget(p_lbl)
# Result summary
if self.action_result:
res_lbl = QLabel(self.action_result[:400] + ("..." if len(self.action_result) > 400 else ""))
res_lbl.setWordWrap(True)
res_lbl.setStyleSheet("color: #cbd5e1; font-size: 11px; font-family: monospace;")
c_layout.addWidget(res_lbl)
return card
# ============================================================================
# 7B. TEAM TASK BOARD DRAWER & SETUP MODAL
# ============================================================================
class TeamTaskBoardDrawer(QFrame):
"""
Collapsible glassmorphic drawer displaying the live Team Board:
- Active Subtasks table with Status & Assigned Bot
- Live Locked Files (anti-collision)
- Shared Discoveries & Notes
"""
def __init__(self, parent=None):
super().__init__(parent)
self.setVisible(False)
self.init_ui()
def init_ui(self):
self.setStyleSheet("""
TeamTaskBoardDrawer {
background-color: #12151e;
border: 1px solid #283042;
border-radius: 10px;
margin: 4px 0px 8px 0px;
}
""")
layout = QVBoxLayout(self)
layout.setContentsMargins(12, 10, 12, 10)
layout.setSpacing(8)
# Header
hdr = QHBoxLayout()
title = QLabel("Live Team Board & Anti-Duplication Registry")
title.setStyleSheet("font-size: 12.5px; font-weight: 800; color: #6366f1; letter-spacing: 0.5px;")
hdr.addWidget(title)
hdr.addStretch()
self.summary_badge = QLabel("0 Tasks")
self.summary_badge.setStyleSheet("""
background-color: #1e2538;
color: #93c5fd;
border: 1px solid #3b82f6;
border-radius: 4px;
padding: 2px 8px;
font-size: 10px;
font-weight: 700;
""")
hdr.addWidget(self.summary_badge)
layout.addLayout(hdr)
# Tasks Container Scroll
self.tasks_box = QVBoxLayout()
self.tasks_box.setSpacing(4)
layout.addLayout(self.tasks_box)
# Locked Files Row
self.locks_label = QLabel("No files currently locked.")
self.locks_label.setStyleSheet("color: #64748b; font-size: 10.5px; font-style: italic;")
layout.addWidget(self.locks_label)
def update_board(self, tasks: list[dict], locked_files: dict[str, str]):
# Clear existing items
while self.tasks_box.count() > 0:
it = self.tasks_box.takeAt(0)
if it.widget():
it.widget().deleteLater()
done_count = sum(1 for t in tasks if t.get("status") == "completed")
self.summary_badge.setText(f"{done_count}/{len(tasks)} Completed")
for t in tasks:
row = QFrame()
row.setStyleSheet("background-color: #181c28; border-radius: 6px; padding: 4px;")
r_layout = QHBoxLayout(row)
r_layout.setContentsMargins(8, 4, 8, 4)
r_layout.setSpacing(8)
tid = QLabel(f"#{t.get('id')}")
tid.setStyleSheet("color: #6366f1; font-weight: 800; font-size: 11px;")
r_layout.addWidget(tid)
desc = QLabel(t.get("title", ""))
desc.setStyleSheet("color: #e2e8f0; font-size: 11px;")
desc.setWordWrap(True)
r_layout.addWidget(desc, 1)
assignee = t.get("assigned_to", "Unassigned")
assign_lbl = QLabel(assignee)
assign_lbl.setStyleSheet("color: #94a3b8; font-size: 10px; font-weight: 600; background: #22293a; padding: 2px 6px; border-radius: 4px;")
r_layout.addWidget(assign_lbl)
status = t.get("status", "pending").upper()
status_lbl = QLabel(status)
if status == "COMPLETED":
s_style = "background-color: #064e3b; color: #34d399; border: 1px solid #059669;"
elif status == "IN_PROGRESS":
s_style = "background-color: #312e81; color: #c7d2fe; border: 1px solid #6366f1;"
else:
s_style = "background-color: #1e2538; color: #94a3b8; border: 1px solid #334155;"
status_lbl.setStyleSheet(f"{s_style} border-radius: 4px; padding: 2px 6px; font-size: 9px; font-weight: 800;")
r_layout.addWidget(status_lbl)
self.tasks_box.addWidget(row)
# Update Locks
if locked_files:
locks_text = "Locked Files: " + ", ".join(f"{f} (by {b})" for f, b in locked_files.items())
self.locks_label.setText(locks_text)
self.locks_label.setStyleSheet("color: #f59e0b; font-size: 10.5px;")
else:
self.locks_label.setText("No active file locks — all files ready for editing.")
self.locks_label.setStyleSheet("color: #64748b; font-size: 10.5px; font-style: italic;")
class TeamSetupDialog(QDialog):
"""
Configures up to 3 specialized bots for Magi Bot TEAM mode.
Full configuration: Name, Model (>=7B local Ollama), Role, Color, and Custom System Prompt for each bot.
"""
def __init__(self, current_bots: list[dict], parent=None):
super().__init__(parent)
self.all_bots = bot_manager.get_all_bots()
self.selected_bots = [dict(b) for b in current_bots[:3]]
self.eligible_models = get_eligible_bot_models()
self.setWindowTitle("Configure Magi Bot TEAM (3 Specialized Bots)")
self.setFixedSize(660, 620)
self.bot_editors = []
self.init_ui()
def init_ui(self):
self.setStyleSheet("""
QDialog {
background-color: #0f1117;
border: 1.5px solid #2d3345;
border-radius: 14px;
color: #ffffff;
}
QTabWidget::pane {
border: 1px solid #283042;
border-radius: 10px;
background-color: #141722;
padding: 8px;
}
QTabBar::tab {
background-color: #171b26;
color: #94a3b8;
border: 1px solid #283042;
border-bottom: none;
border-top-left-radius: 8px;
border-top-right-radius: 8px;
padding: 8px 18px;
font-size: 12px;
font-weight: 700;
margin-right: 4px;
}
QTabBar::tab:selected {
background-color: #141722;
color: #ffffff;
border-bottom: 2px solid #6366f1;
}
QTabBar::tab:hover {
color: #ffffff;
}
""")
layout = QVBoxLayout(self)
layout.setContentsMargins(22, 18, 22, 18)
layout.setSpacing(12)
# Title
hdr_row = QHBoxLayout()
title = QLabel("Magi Bot TEAM Configuration")
title.setStyleSheet("font-size: 17px; font-weight: 800; color: #ffffff;")
hdr_row.addWidget(title)
hdr_row.addStretch()
badge = QLabel("3 Active Bots")
badge.setStyleSheet("background-color: #312e81; color: #c7d2fe; border: 1px solid #6366f1; border-radius: 6px; padding: 3px 8px; font-size: 10.5px; font-weight: 700;")
hdr_row.addWidget(badge)
layout.addLayout(hdr_row)
sub = QLabel("Customize the name, model (>=7B local Ollama), role persona, profile color, and custom system prompt for each of your 3 collaborative bots.")
sub.setStyleSheet("font-size: 11.5px; color: #94a3b8;")
sub.setWordWrap(True)
layout.addWidget(sub)
# Tabs for Bot 1, Bot 2, Bot 3
self.tabs = QTabWidget()
default_names = ["MagiArchitect", "MagiCoder", "MagiTester"]
default_roles = ["Lead Architect", "Core Developer", "QA / Tester"]
default_models = ["deepcoder:14b", "deepcoder:14b", "lfm2.5:8b"]
default_colors = ["#007aff", "#34c759", "#ff9500"]
default_prompts = [
"You are the Lead Architect of Magi Bot TEAM. Plan modular, clean file structures and system architecture.",
"You are the Core Developer of Magi Bot TEAM. Implement functional, well-documented code files and modules.",
"You are the QA and Runtime Tester of Magi Bot TEAM. Write tests, verify headless executions, and inspect outputs."
]
for i in range(3):
if i < len(self.selected_bots):
b_data = self.selected_bots[i]
else:
b_data = {
"id": f"team_bot_{i+1}",
"name": default_names[i],
"model": default_models[i],
"color": default_colors[i],
"role": default_roles[i],
"custom_system_prompt": default_prompts[i]
}
tab_widget = self.create_bot_tab_widget(i, b_data)
self.tabs.addTab(tab_widget, f"Bot #{i+1} ({b_data.get('name', f'Bot {i+1}')})")
layout.addWidget(self.tabs, 1)
# Action Buttons Row
btn_row = QHBoxLayout()
btn_row.setSpacing(10)
reset_btn = QPushButton("Reset Defaults")
reset_btn.setCursor(QCursor(Qt.CursorShape.PointingHandCursor))
reset_btn.setStyleSheet("""
QPushButton {
background-color: #1e2433;
color: #94a3b8;
border: 1px solid #334155;
border-radius: 8px;
padding: 9px 16px;
font-size: 11.5px;
font-weight: 600;
}
QPushButton:hover {
background-color: #2b354d;
color: #ffffff;
}
""")
reset_btn.clicked.connect(self.reset_to_defaults)
cancel_btn = QPushButton("Cancel")
cancel_btn.setCursor(QCursor(Qt.CursorShape.PointingHandCursor))
cancel_btn.setStyleSheet("""
QPushButton {
background-color: #1e2230;
color: #94a3b8;
border: 1px solid #334155;
border-radius: 8px;
padding: 9px 18px;
font-size: 12px;
font-weight: 600;
}
QPushButton:hover {
background-color: #2b3247;
color: #ffffff;
}
""")
cancel_btn.clicked.connect(self.reject)
save_btn = QPushButton("Save & Apply Team")
save_btn.setCursor(QCursor(Qt.CursorShape.PointingHandCursor))
save_btn.setStyleSheet("""
QPushButton {
background: qlineargradient(x1:0, y1:0, x2:1, y2:0, stop:0 #6366f1, stop:1 #8b5cf6);
color: #ffffff;
border: none;
border-radius: 8px;
padding: 9px 24px;
font-size: 12.5px;
font-weight: 700;
}
QPushButton:hover {
background: qlineargradient(x1:0, y1:0, x2:1, y2:0, stop:0 #4f46e5, stop:1 #7c3aed);
}
""")
save_btn.clicked.connect(self.handle_save)
btn_row.addWidget(reset_btn)
btn_row.addStretch()
btn_row.addWidget(cancel_btn)
btn_row.addWidget(save_btn)
layout.addLayout(btn_row)
def create_bot_tab_widget(self, index: int, bot_data: dict) -> QWidget:
tab = QWidget()
t_layout = QVBoxLayout(tab)
t_layout.setContentsMargins(14, 12, 14, 12)
t_layout.setSpacing(10)
# Quick Clone from Solo Bot dropdown
clone_row = QHBoxLayout()
clone_lbl = QLabel("Import / Clone from Solo Bot:")
clone_lbl.setStyleSheet("color: #94a3b8; font-size: 11px; font-weight: 600;")
clone_row.addWidget(clone_lbl)
clone_combo = QComboBox()
clone_combo.setStyleSheet("""
QComboBox {
background-color: #1a1e2b;
color: #ffffff;
border: 1px solid #334155;
border-radius: 6px;
padding: 4px 8px;
font-size: 11.5px;
}
""")
clone_combo.addItem("-- Select Solo Bot to Clone --", None)
for b in self.all_bots:
clone_combo.addItem(f"{b.get('name')} ({b.get('model')})", b)
clone_row.addWidget(clone_combo, 1)
t_layout.addLayout(clone_row)
# 2 Columns: Bot Name + Role
row1 = QHBoxLayout()
row1.setSpacing(12)
name_box = QVBoxLayout()
name_box.setSpacing(3)
name_lbl = QLabel("Bot Name:")
name_lbl.setStyleSheet("color: #cbd5e1; font-size: 11.5px; font-weight: 700;")
name_input = QLineEdit(bot_data.get("name", f"Bot {index+1}"))
name_input.setStyleSheet("""
QLineEdit {
background-color: #1a1e2b;
color: #ffffff;
border: 1px solid #334155;
border-radius: 6px;
padding: 6px 10px;
font-size: 12px;
}
QLineEdit:focus { border-color: #6366f1; }
""")
name_box.addWidget(name_lbl)
name_box.addWidget(name_input)
row1.addLayout(name_box, 1)
role_box = QVBoxLayout()
role_box.setSpacing(3)
role_lbl = QLabel("Specialized Role:")
role_lbl.setStyleSheet("color: #cbd5e1; font-size: 11.5px; font-weight: 700;")
role_input = QLineEdit(bot_data.get("role", f"Role {index+1}"))
role_input.setStyleSheet("""
QLineEdit {
background-color: #1a1e2b;
color: #ffffff;
border: 1px solid #334155;
border-radius: 6px;
padding: 6px 10px;
font-size: 12px;
}
QLineEdit:focus { border-color: #6366f1; }
""")
role_box.addWidget(role_lbl)
role_box.addWidget(role_input)
row1.addLayout(role_box, 1)
t_layout.addLayout(row1)
# Model Selection (>=8B & Cloud) + Actions per Turn Row
row2 = QHBoxLayout()
row2.setSpacing(12)
model_box = QVBoxLayout()
model_box.setSpacing(3)
model_lbl = QLabel("Model (Requires >= 7B local Ollama):")
model_lbl.setStyleSheet("color: #cbd5e1; font-size: 11.5px; font-weight: 700;")
model_combo = QComboBox()
model_combo.setStyleSheet("""
QComboBox {
background-color: #1a1e2b;
color: #ffffff;
border: 1px solid #334155;
border-radius: 6px;
padding: 6px 10px;
font-size: 12px;
}
""")
model_combo.addItems(self.eligible_models)
cur_model = bot_data.get("model", "")
idx = model_combo.findText(cur_model)
if idx >= 0:
model_combo.setCurrentIndex(idx)
model_box.addWidget(model_lbl)
model_box.addWidget(model_combo)
row2.addLayout(model_box, 3)
actions_box = QVBoxLayout()
actions_box.setSpacing(3)
actions_lbl = QLabel("Actions / Turn:")
actions_lbl.setStyleSheet("color: #cbd5e1; font-size: 11.5px; font-weight: 700;")
actions_spin = QSpinBox()
actions_spin.setRange(1, 20)
actions_spin.setValue(int(bot_data.get("max_actions_per_turn", 1)))
actions_spin.setStyleSheet("""
QSpinBox {
background-color: #1a1e2b;
color: #ffffff;
border: 1px solid #334155;
border-radius: 6px;
padding: 6px 10px;
font-size: 12px;
font-weight: 700;
}
QSpinBox:focus { border-color: #6366f1; }
""")
actions_spin.setToolTip("Select how many actions this bot can execute consecutively before yielding to the next bot.")
actions_box.addWidget(actions_lbl)
actions_box.addWidget(actions_spin)
row2.addLayout(actions_box, 1)
t_layout.addLayout(row2)
# Profile Color Palette
color_lbl = QLabel("Profile Color:")
color_lbl.setStyleSheet("color: #cbd5e1; font-size: 11.5px; font-weight: 700;")
t_layout.addWidget(color_lbl)
color_row = QHBoxLayout()
color_row.setSpacing(8)
color_holder = {"selected_color": bot_data.get("color", DEFAULT_PALETTE[index % len(DEFAULT_PALETTE)])}
color_buttons = []
def make_color_selector(c_hex, buttons_list, holder_dict):
def select_fn():
holder_dict["selected_color"] = c_hex
for c_val, b_obj in buttons_list:
border = "3px solid #ffffff" if c_val == c_hex else "1px solid rgba(255,255,255,0.2)"
b_obj.setStyleSheet(f"background-color: {c_val}; border-radius: 12px; border: {border};")
return select_fn
for c in DEFAULT_PALETTE:
btn = QPushButton()
btn.setFixedSize(24, 24)
btn.setCursor(QCursor(Qt.CursorShape.PointingHandCursor))
border = "3px solid #ffffff" if c == color_holder["selected_color"] else "1px solid rgba(255,255,255,0.2)"
btn.setStyleSheet(f"background-color: {c}; border-radius: 12px; border: {border};")
color_buttons.append((c, btn))
btn.clicked.connect(make_color_selector(c, color_buttons, color_holder))
color_row.addWidget(btn)
color_row.addStretch()
t_layout.addLayout(color_row)
# Custom System Prompt
prompt_lbl = QLabel("Custom System Prompt (Persona & Specialized Instructions):")
prompt_lbl.setStyleSheet("color: #cbd5e1; font-size: 11.5px; font-weight: 700;")
t_layout.addWidget(prompt_lbl)
prompt_input = QTextEdit()
prompt_input.setPlaceholderText("Enter specialized persona instructions for this bot (e.g. 'You are the Lead Architect. Focus on code structure, clean modular patterns, and system design.').")
prompt_input.setText(bot_data.get("custom_system_prompt", ""))
prompt_input.setStyleSheet("""
QTextEdit {
background-color: #1a1e2b;
color: #ffffff;
border: 1px solid #334155;
border-radius: 6px;
padding: 6px 10px;
font-size: 11.5px;
}
QTextEdit:focus { border-color: #6366f1; }
""")
t_layout.addWidget(prompt_input, 1)
# Connect Clone Dropdown
def handle_clone():
c_bot = clone_combo.currentData()
if c_bot:
name_input.setText(c_bot.get("name", ""))
c_idx = model_combo.findText(c_bot.get("model", ""))
if c_idx >= 0:
model_combo.setCurrentIndex(c_idx)
prompt_input.setText(c_bot.get("custom_system_prompt", ""))
c_color = c_bot.get("color", DEFAULT_PALETTE[0])
make_color_selector(c_color, color_buttons, color_holder)()
clone_combo.currentIndexChanged.connect(handle_clone)
# Update tab label dynamically when name changes
def handle_name_change(new_name):
self.tabs.setTabText(index, f"Bot #{index+1} ({new_name.strip() or f'Bot {index+1}'})")
name_input.textChanged.connect(handle_name_change)
editor_dict = {
"id": bot_data.get("id", f"team_bot_{index+1}"),
"name_input": name_input,
"role_input": role_input,
"model_combo": model_combo,
"actions_spin": actions_spin,
"color_holder": color_holder,
"prompt_input": prompt_input
}
self.bot_editors.append(editor_dict)
return tab
def reset_to_defaults(self):
default_names = ["MagiArchitect", "MagiCoder", "MagiTester"]
default_roles = ["Lead Architect", "Core Developer", "QA / Tester"]
default_models = ["deepcoder:14b", "deepcoder:14b", "lfm2.5:8b"]
default_colors = ["#007aff", "#34c759", "#ff9500"]
default_prompts = [
"You are the Lead Architect of Magi Bot TEAM. Plan modular, clean file structures and system architecture.",
"You are the Core Developer of Magi Bot TEAM. Implement functional, well-documented code files and modules.",
"You are the QA and Runtime Tester of Magi Bot TEAM. Write tests, verify headless executions, and inspect outputs."
]
for i, ed in enumerate(self.bot_editors):
ed["name_input"].setText(default_names[i])
ed["role_input"].setText(default_roles[i])
m_idx = ed["model_combo"].findText(default_models[i])
if m_idx >= 0:
ed["model_combo"].setCurrentIndex(m_idx)
ed["actions_spin"].setValue(1)
ed["color_holder"]["selected_color"] = default_colors[i]
ed["prompt_input"].setText(default_prompts[i])
def handle_save(self):
team = []
for i, ed in enumerate(self.bot_editors):
name = ed["name_input"].text().strip() or f"Bot {i+1}"
role = ed["role_input"].text().strip() or f"Role {i+1}"
model = ed["model_combo"].currentText()
max_actions = ed["actions_spin"].value()
color = ed["color_holder"]["selected_color"]
prompt = ed["prompt_input"].toPlainText().strip()
team.append({
"id": ed["id"],
"name": name,
"role": role,
"model": model,
"max_actions_per_turn": max_actions,
"color": color,
"custom_system_prompt": prompt
})
self.selected_bots = team
self.accept()
# ============================================================================
# 8. BOT CONFIGURATION & CREATION MODAL
# ============================================================================
class BotConfigDialog(QDialog):
"""
Dialog to create or edit a Bot (Name, Model, Profile Color, Custom System Prompt).
Only local Ollama models >= 7B are available to select.
Strictly emoji-free.
"""
def __init__(self, bot_data: dict = None, parent=None):
super().__init__(parent)
self.bot_data = bot_data
self.is_edit = bot_data is not None
self.selected_color = bot_data.get("color", DEFAULT_PALETTE[0]) if self.is_edit else DEFAULT_PALETTE[0]
self.setWindowTitle("Edit Bot Profile" if self.is_edit else "Create New Magi Bot")
self.setFixedSize(520, 530)
self.init_ui()
def init_ui(self):
self.setStyleSheet("""
QDialog {
background-color: #111319;
border: 1.5px solid #2d3345;
border-radius: 14px;
color: #ffffff;
}
""")
layout = QVBoxLayout(self)
layout.setContentsMargins(26, 22, 26, 22)
layout.setSpacing(13)
# Title
title = QLabel("Edit Bot Profile" if self.is_edit else "Create New Magi Bot")
title.setStyleSheet("font-size: 18px; font-weight: 800; color: #ffffff;")
layout.addWidget(title)
# Subtitle
sub = QLabel("Each bot has its own workspace folder named Magi(bot_name), custom persona, and requires a local Ollama model >= 7B.")
sub.setStyleSheet("font-size: 12px; color: #94a3b8;")
sub.setWordWrap(True)
layout.addWidget(sub)
# 1. Bot Name
name_lbl = QLabel("Bot Name")
name_lbl.setStyleSheet("font-size: 12px; font-weight: 700; color: #cbd5e1;")
layout.addWidget(name_lbl)
self.name_input = QLineEdit()
self.name_input.setPlaceholderText("e.g. PythonCoder, FileMaster, Analyst...")
if self.is_edit:
self.name_input.setText(self.bot_data.get("name", ""))
self.name_input.setStyleSheet("""
QLineEdit {
background-color: #1a1c26;
color: #ffffff;
border: 1px solid #334155;
border-radius: 8px;
padding: 9px 12px;
font-size: 13px;
}
QLineEdit:focus {
border: 1.5px solid #007aff;
}
""")
layout.addWidget(self.name_input)
# 2. AI Model Selection (Filtered >= 7B local Ollama)
model_lbl = QLabel("AI Model (Minimum 7B, local Ollama)")
model_lbl.setStyleSheet("font-size: 12px; font-weight: 700; color: #cbd5e1;")
layout.addWidget(model_lbl)
self.model_combo = QComboBox()
self.model_combo.setStyleSheet("""
QComboBox {
background-color: #1a1c26;
color: #ffffff;
border: 1px solid #334155;
border-radius: 8px;
padding: 8px 12px;
font-size: 12.5px;
}
QComboBox QAbstractItemView {
background-color: #1a1c26;
color: #ffffff;
selection-background-color: #007aff;
}
""")
self.load_available_models()
layout.addWidget(self.model_combo)
model_note = QLabel("Minimum 7B local Ollama model for bot execution. Fast task planning and step decision is handled in the background by Granite 4.1:3b.")
model_note.setStyleSheet("font-size: 10.5px; color: #64748b;")
model_note.setWordWrap(True)
layout.addWidget(model_note)
# 3. Avatar Color Selection
color_lbl = QLabel("Avatar Color")
color_lbl.setStyleSheet("font-size: 12px; font-weight: 700; color: #cbd5e1;")
layout.addWidget(color_lbl)
color_row = QHBoxLayout()
color_row.setSpacing(10)
self.color_buttons = []
for c in DEFAULT_PALETTE:
btn = QPushButton()
btn.setFixedSize(28, 28)
btn.setCursor(QCursor(Qt.CursorShape.PointingHandCursor))
border_style = "3px solid #ffffff" if c == self.selected_color else "1px solid rgba(255,255,255,0.2)"
btn.setStyleSheet(f"background-color: {c}; border-radius: 14px; border: {border_style};")
btn.clicked.connect(lambda _, col=c: self.select_color(col))
self.color_buttons.append((c, btn))
color_row.addWidget(btn)
color_row.addStretch()
layout.addLayout(color_row)
# 4. Custom System Prompt (User Defined Persona)
prompt_lbl = QLabel("Custom System Prompt (Persona & Instructions)")
prompt_lbl.setStyleSheet("font-size: 12px; font-weight: 700; color: #cbd5e1;")
layout.addWidget(prompt_lbl)
self.prompt_input = QTextEdit()
self.prompt_input.setPlaceholderText("Enter custom instructions for this bot (e.g., 'You are an expert Python automation engineer. Always explain your code clearly and answer user questions directly.').")
if self.is_edit:
self.prompt_input.setText(self.bot_data.get("custom_system_prompt", ""))
self.prompt_input.setStyleSheet("""
QTextEdit {
background-color: #1a1c26;
color: #ffffff;
border: 1px solid #334155;
border-radius: 8px;
padding: 8px 12px;
font-size: 12.5px;
}
QTextEdit:focus {
border: 1.5px solid #007aff;
}
""")
layout.addWidget(self.prompt_input)
layout.addStretch()
# Bottom Buttons
btn_row = QHBoxLayout()
btn_row.setSpacing(10)
cancel_btn = QPushButton("Cancel")
cancel_btn.setCursor(QCursor(Qt.CursorShape.PointingHandCursor))
cancel_btn.setStyleSheet("""
QPushButton {
background-color: #1e2230;
color: #94a3b8;
border: 1px solid #334155;
border-radius: 8px;
padding: 9px 18px;
font-size: 12px;
font-weight: 600;
}
QPushButton:hover {
background-color: #2b3247;
color: #ffffff;
}
""")
cancel_btn.clicked.connect(self.reject)
save_btn = QPushButton("Save Bot" if self.is_edit else "Create Bot")
save_btn.setCursor(QCursor(Qt.CursorShape.PointingHandCursor))
save_btn.setStyleSheet("""
QPushButton {
background: qlineargradient(x1:0, y1:0, x2:1, y2:0, stop:0 #007aff, stop:1 #0a84ff);
color: #ffffff;
border: none;
border-radius: 8px;
padding: 9px 24px;
font-size: 12.5px;
font-weight: 700;
}
QPushButton:hover {
background: qlineargradient(x1:0, y1:0, x2:1, y2:0, stop:0 #0062cc, stop:1 #0071e3);
}
""")
save_btn.clicked.connect(self.handle_save)
btn_row.addStretch()
btn_row.addWidget(cancel_btn)
btn_row.addWidget(save_btn)
layout.addLayout(btn_row)
def select_color(self, color_hex: str):
self.selected_color = color_hex
for c, btn in self.color_buttons:
border_style = "3px solid #ffffff" if c == self.selected_color else "1px solid rgba(255,255,255,0.2)"
btn.setStyleSheet(f"background-color: {c}; border-radius: 14px; border: {border_style};")
def load_available_models(self):
"""
Loads only local Ollama models >= 7B.
"""
models = get_eligible_bot_models()
self.model_combo.clear()
self.model_combo.addItems(models)
if self.is_edit:
cur_model = self.bot_data.get("model", "")
idx = self.model_combo.findText(cur_model)
if idx >= 0:
self.model_combo.setCurrentIndex(idx)
def handle_save(self):
name = self.name_input.text().strip()
if not name:
QMessageBox.warning(self, "Invalid Name", "Please enter a valid name for the bot.")
return
model = self.model_combo.currentText()
prompt = self.prompt_input.toPlainText().strip()
color = self.selected_color
if self.is_edit:
ok, msg = bot_manager.update_bot(self.bot_data["id"], name, model, color, prompt)
else:
ok, msg, _ = bot_manager.create_bot(name, model, color, prompt)
if ok:
self.accept()
else:
QMessageBox.warning(self, "Error", msg)
# ============================================================================
# 9. MAIN MAGI BOT MESSAGING WIDGET (LEFT CONTACTS + RIGHT CHAT)
# ============================================================================
class BotContactItem(QFrame):
"""
Sidebar contact entry for a single bot.
Displays circular avatar with initial letter, name, model badge, and edit/delete actions.
"""
selected = pyqtSignal(str)
edit_requested = pyqtSignal(str)
delete_requested = pyqtSignal(str)
def __init__(self, bot_data: dict, is_active: bool = False, parent=None):
super().__init__(parent)
self.bot_data = bot_data
self.is_active = is_active
self.setCursor(QCursor(Qt.CursorShape.PointingHandCursor))
self.setFixedHeight(66)
self.init_ui()
def init_ui(self):
layout = QHBoxLayout(self)
layout.setContentsMargins(12, 8, 12, 8)
layout.setSpacing(10)
# Avatar Circle
name = self.bot_data.get("name", "Bot")
color = self.bot_data.get("color", "#007aff")
initial = name[0].upper() if name else "B"
avatar = QLabel(initial)
avatar.setFixedSize(40, 40)
avatar.setAlignment(Qt.AlignmentFlag.AlignCenter)
avatar.setStyleSheet(f"""
background-color: {color};
color: #ffffff;
font-size: 16px;
font-weight: 800;
border-radius: 20px;
""")
layout.addWidget(avatar)
# Name & Subtitle
info_box = QVBoxLayout()
info_box.setSpacing(2)
info_box.setAlignment(Qt.AlignmentFlag.AlignVCenter)
name_lbl = QLabel(name)
name_lbl.setStyleSheet("font-size: 13.5px; font-weight: 700; color: #ffffff;")
info_box.addWidget(name_lbl)
model_lbl = QLabel(self.bot_data.get("model", "deepcoder:14b"))
model_lbl.setStyleSheet("font-size: 11px; color: #94a3b8;")
info_box.addWidget(model_lbl)
layout.addLayout(info_box, 1)
# Edit button
edit_btn = QPushButton("Edit")
edit_btn.setFixedSize(36, 24)
edit_btn.setCursor(QCursor(Qt.CursorShape.PointingHandCursor))
edit_btn.setStyleSheet("""
QPushButton {
background-color: #1e2433;
color: #94a3b8;
border: 1px solid #334155;
border-radius: 6px;
font-size: 10px;
font-weight: 600;
}
QPushButton:hover {
background-color: #3b82f6;
color: #ffffff;
}
""")
edit_btn.clicked.connect(lambda: self.edit_requested.emit(self.bot_data.get("id", "")))
layout.addWidget(edit_btn)
# Delete button
del_btn = QPushButton("X")
del_btn.setFixedSize(24, 24)
del_btn.setCursor(QCursor(Qt.CursorShape.PointingHandCursor))
del_btn.setStyleSheet("""
QPushButton {
background-color: #1e2433;
color: #ef4444;
border: 1px solid #334155;
border-radius: 6px;
font-size: 11px;
font-weight: 800;
}
QPushButton:hover {
background-color: #dc2626;
color: #ffffff;
}
""")
del_btn.clicked.connect(lambda: self.delete_requested.emit(self.bot_data.get("id", "")))
layout.addWidget(del_btn)
self.update_style()
def update_style(self):
if self.is_active:
self.setStyleSheet("""
BotContactItem {
background-color: #1e2638;
border-left: 3px solid #007aff;
border-radius: 8px;
}
""")
else:
self.setStyleSheet("""
BotContactItem {
background-color: transparent;
border-left: 3px solid transparent;
border-radius: 8px;
}
BotContactItem:hover {
background-color: #161a24;
}
""")
def mousePressEvent(self, event):
if event.button() == Qt.MouseButton.LeftButton:
self.selected.emit(self.bot_data.get("id", ""))
super().mousePressEvent(event)
class MagiBotWidget(QWidget):
"""
Main Messaging App view for Magi Bots.
- Left sidebar: mode switcher (Solo Bot vs Magi Bot TEAM), contacts list (up to 5 bots), New Bot button, search bar.
- Right panel: chat view, 3-bot team live status cards, collapsible task board drawer, step execution banner, confirmation modals, iMessage bubbles.
- Strictly emoji-free.
"""
back_requested = pyqtSignal()
def __init__(self, parent=None):
super().__init__(parent)
self.active_bot_id = None
self.active_worker: MagiBotWorker | None = None
self.gemini_key = ""
self.mistral_key = ""
# Step Chaining tracking (unlimited steps until task is completed)
self.current_user_task = ""
self.step_counter = 1
# Message Queue while bot is working (up to 5 messages)
self.queued_messages: list[str] = []
self.max_queue_size = 5
# Magi Bot TEAM state
self.is_team_mode = False
self.team_worker: MagiBotTeamWorker | None = None
self.team_workspace_dir = os.path.join(get_base_bots_dir(), "Magi_Team_Workspace")
os.makedirs(self.team_workspace_dir, exist_ok=True)
self.team_history_file = os.path.join(self.team_workspace_dir, "team_history.json")
self.team_messages: list[dict] = []
self.team_bots: list[dict] = []
self.init_team_bots()
self.init_ui()
self.refresh_bot_list()
self.load_team_history()
def init_team_bots(self):
self.team_config_file = os.path.join(self.team_workspace_dir, "team_config.json")
if os.path.exists(self.team_config_file):
try:
with open(self.team_config_file, "r", encoding="utf-8") as f:
saved = json.load(f)
if isinstance(saved, list) and len(saved) == 3:
for b in saved:
if "max_actions_per_turn" not in b:
b["max_actions_per_turn"] = 1
self.team_bots = saved
return
except Exception:
pass
# Defaults for the 3 specialized bots
default_names = ["MagiArchitect", "MagiCoder", "MagiTester"]
default_roles = ["Lead Architect", "Core Developer", "QA / Tester"]
default_models = ["deepcoder:14b", "deepcoder:14b", "lfm2.5:8b"]
default_colors = ["#007aff", "#34c759", "#ff9500"]
default_prompts = [
"You are the Lead Architect of Magi Bot TEAM. Plan modular, clean file structures and system architecture.",
"You are the Core Developer of Magi Bot TEAM. Implement functional, well-documented code files and modules.",
"You are the QA and Runtime Tester of Magi Bot TEAM. Write tests, verify headless executions, and inspect outputs."
]
self.team_bots = []
for i in range(3):
self.team_bots.append({
"id": f"team_bot_{i+1}",
"name": default_names[i],
"model": default_models[i],
"color": default_colors[i],
"role": default_roles[i],
"max_actions_per_turn": 1,
"custom_system_prompt": default_prompts[i]
})
self.save_team_config()
def save_team_config(self):
try:
with open(self.team_config_file, "w", encoding="utf-8") as f:
json.dump(self.team_bots, f, indent=2, ensure_ascii=False)
except Exception as e:
print(f"[MagiBotWidget] Failed to save team config: {e}")
def load_team_history(self):
if os.path.exists(self.team_history_file):
try:
with open(self.team_history_file, "r", encoding="utf-8") as f:
self.team_messages = json.load(f)
except Exception:
self.team_messages = []
else:
self.team_messages = []
def save_team_history(self):
try:
with open(self.team_history_file, "w", encoding="utf-8") as f:
json.dump(self.team_messages, f, indent=2, ensure_ascii=False)
except Exception as e:
print(f"[MagiBotWidget] Failed to save team history: {e}")
def set_keys(self, gemini_key: str, mistral_key: str):
self.gemini_key = gemini_key
self.mistral_key = mistral_key
def init_ui(self):
main_layout = QHBoxLayout(self)
main_layout.setContentsMargins(0, 0, 0, 0)
main_layout.setSpacing(0)
# Splitter between sidebar (contacts) and chat
splitter = QSplitter(Qt.Orientation.Horizontal)
splitter.setStyleSheet("""
QSplitter::handle {
background-color: #1f2430;
width: 1.5px;
}
""")
# ------------------ LEFT SIDEBAR ------------------
sidebar = QWidget()
sidebar.setFixedWidth(290)
sidebar.setStyleSheet("background-color: #0d0f15;")
s_layout = QVBoxLayout(sidebar)
s_layout.setContentsMargins(14, 16, 14, 16)
s_layout.setSpacing(10)
# Sidebar Header
top_row = QHBoxLayout()
top_title = QLabel("Magi Bots")
top_title.setStyleSheet("font-size: 17px; font-weight: 800; color: #ffffff;")
top_row.addWidget(top_title)
self.count_badge = QLabel("0/5")
self.count_badge.setStyleSheet("font-size: 12px; font-weight: 700; color: #64748b;")
top_row.addWidget(self.count_badge)
top_row.addStretch()
self.new_bot_btn = QPushButton("+ New Bot")
self.new_bot_btn.setCursor(QCursor(Qt.CursorShape.PointingHandCursor))
self.new_bot_btn.setStyleSheet("""
QPushButton {
background: qlineargradient(x1:0, y1:0, x2:1, y2:0, stop:0 #007aff, stop:1 #0a84ff);
color: #ffffff;
border: none;
border-radius: 7px;
padding: 6px 12px;
font-size: 11.5px;
font-weight: 700;
}
QPushButton:hover {
background: qlineargradient(x1:0, y1:0, x2:1, y2:0, stop:0 #0062cc, stop:1 #0071e3);
}
QPushButton:disabled {
background-color: #1f2430;
color: #475569;
}
""")
self.new_bot_btn.clicked.connect(self.open_create_bot_dialog)
top_row.addWidget(self.new_bot_btn)
s_layout.addLayout(top_row)
# Mode Selector Bar (Solo Bot vs Magi Bot TEAM)
mode_container = QFrame()
mode_container.setStyleSheet("background-color: #161922; border-radius: 8px; padding: 2px;")
mode_layout = QHBoxLayout(mode_container)
mode_layout.setContentsMargins(3, 3, 3, 3)
mode_layout.setSpacing(4)
self.mode_solo_btn = QPushButton("Solo Bot")
self.mode_solo_btn.setCursor(QCursor(Qt.CursorShape.PointingHandCursor))
self.mode_solo_btn.clicked.connect(self.switch_to_solo_mode)
mode_layout.addWidget(self.mode_solo_btn)
self.mode_team_btn = QPushButton("TEAM (3 Bots)")
self.mode_team_btn.setCursor(QCursor(Qt.CursorShape.PointingHandCursor))
self.mode_team_btn.clicked.connect(self.switch_to_team_mode)
mode_layout.addWidget(self.mode_team_btn)
s_layout.addWidget(mode_container)
# Search Bar
self.search_input = QLineEdit()
self.search_input.setPlaceholderText("Search bots...")
self.search_input.setStyleSheet("""
QLineEdit {
background-color: #161922;
color: #ffffff;
border: 1px solid #282f40;
border-radius: 8px;
padding: 7px 12px;
font-size: 12px;
}
QLineEdit:focus {
border-color: #007aff;
}
""")
self.search_input.textChanged.connect(self.filter_bots)
s_layout.addWidget(self.search_input)
# Contacts Scroll Area
self.contact_scroll = QScrollArea()
self.contact_scroll.setWidgetResizable(True)
self.contact_scroll.setFrameShape(QFrame.Shape.NoFrame)
self.contact_scroll.setStyleSheet("background: transparent; border: none;")
self.contact_list_widget = QWidget()
self.contact_list_layout = QVBoxLayout(self.contact_list_widget)
self.contact_list_layout.setContentsMargins(0, 4, 0, 4)
self.contact_list_layout.setSpacing(6)
self.contact_list_layout.addStretch()
self.contact_scroll.setWidget(self.contact_list_widget)
s_layout.addWidget(self.contact_scroll, 1)
# Back to Home Button
back_btn = QPushButton("Back to Home")
back_btn.setCursor(QCursor(Qt.CursorShape.PointingHandCursor))
back_btn.setStyleSheet("""
QPushButton {
background-color: #161922;
color: #94a3b8;
border: 1px solid #282f40;
border-radius: 8px;
padding: 9px;
font-size: 12px;
font-weight: 600;
}
QPushButton:hover {
background-color: #222736;
color: #ffffff;
}
""")
back_btn.clicked.connect(self.back_requested.emit)
s_layout.addWidget(back_btn)
splitter.addWidget(sidebar)
# ------------------ RIGHT CHAT AREA ------------------
chat_container = QWidget()
chat_container.setStyleSheet("background-color: #0f1117;")
c_layout = QVBoxLayout(chat_container)
c_layout.setContentsMargins(20, 16, 20, 16)
c_layout.setSpacing(10)
# Chat Header Frame
self.chat_header = QFrame()
self.chat_header.setStyleSheet("""
QFrame {
background-color: #141720;
border: 1px solid #232938;
border-radius: 12px;
padding: 6px 14px;
}
""")
ch_layout = QVBoxLayout(self.chat_header)
ch_layout.setContentsMargins(8, 6, 8, 6)
ch_layout.setSpacing(8)
# Solo Header Widget
self.solo_header_widget = QWidget()
sh_layout = QHBoxLayout(self.solo_header_widget)
sh_layout.setContentsMargins(0, 0, 0, 0)
sh_layout.setSpacing(10)
self.hdr_avatar = QLabel("B")
self.hdr_avatar.setFixedSize(36, 36)
self.hdr_avatar.setAlignment(Qt.AlignmentFlag.AlignCenter)
self.hdr_avatar.setStyleSheet("""
background-color: #007aff;
color: #ffffff;
font-size: 15px;
font-weight: 800;
border-radius: 18px;
""")
sh_layout.addWidget(self.hdr_avatar)
hdr_info = QVBoxLayout()
hdr_info.setSpacing(1)
self.hdr_name = QLabel("Select a Bot")
self.hdr_name.setStyleSheet("font-size: 15px; font-weight: 800; color: #ffffff;")
self.hdr_sub = QLabel("No bot selected")
self.hdr_sub.setStyleSheet("font-size: 11px; color: #64748b;")
hdr_info.addWidget(self.hdr_name)
hdr_info.addWidget(self.hdr_sub)
sh_layout.addLayout(hdr_info)
sh_layout.addStretch()
# Step Status Badge
self.step_badge = QLabel("")
self.step_badge.setVisible(False)
self.step_badge.setStyleSheet("""
background-color: #1e293b;
color: #38bdf8;
border: 1px solid #0284c7;
border-radius: 6px;
padding: 4px 10px;
font-size: 11px;
font-weight: 700;
""")
sh_layout.addWidget(self.step_badge)
# Queue Status Badge
self.queue_badge = QLabel("")
self.queue_badge.setVisible(False)
self.queue_badge.setStyleSheet("""
background-color: #312e81;
color: #c7d2fe;
border: 1px solid #6366f1;
border-radius: 6px;
padding: 4px 10px;
font-size: 11px;
font-weight: 700;
""")
sh_layout.addWidget(self.queue_badge)
# Open Workspace Folder Button
self.open_ws_btn = QPushButton("Open Workspace")
self.open_ws_btn.setCursor(QCursor(Qt.CursorShape.PointingHandCursor))
self.open_ws_btn.setStyleSheet("""
QPushButton {
background-color: #1a202c;
color: #cbd5e1;
border: 1px solid #334155;
border-radius: 7px;
padding: 6px 12px;
font-size: 11.5px;
font-weight: 600;
}
QPushButton:hover {
background-color: #2d3748;
color: #ffffff;
}
""")
self.open_ws_btn.clicked.connect(self.open_active_workspace_in_explorer)
sh_layout.addWidget(self.open_ws_btn)
# Clear Chat Button
self.clear_chat_btn = QPushButton("Clear")
self.clear_chat_btn.setCursor(QCursor(Qt.CursorShape.PointingHandCursor))
self.clear_chat_btn.setStyleSheet("""
QPushButton {
background-color: #1a202c;
color: #94a3b8;
border: 1px solid #334155;
border-radius: 7px;
padding: 6px 12px;
font-size: 11.5px;
font-weight: 600;
}
QPushButton:hover {
background-color: #2d3748;
color: #ffffff;
}
""")
self.clear_chat_btn.clicked.connect(self.clear_active_chat)
sh_layout.addWidget(self.clear_chat_btn)
ch_layout.addWidget(self.solo_header_widget)
# Team Header Widget (3 Bot Live Cards + Controls)
self.team_header_widget = QWidget()
self.team_header_widget.setVisible(False)
th_layout = QVBoxLayout(self.team_header_widget)
th_layout.setContentsMargins(0, 0, 0, 0)
th_layout.setSpacing(8)
th_top_row = QHBoxLayout()
th_title = QLabel("Magi Bot TEAM — Collaborative Multi-Agent Swarm")
th_title.setStyleSheet("font-size: 14px; font-weight: 800; color: #6366f1;")
th_top_row.addWidget(th_title)
th_top_row.addStretch()
self.team_board_btn = QPushButton("Live Board")
self.team_board_btn.setCursor(QCursor(Qt.CursorShape.PointingHandCursor))
self.team_board_btn.setStyleSheet("""
QPushButton {
background-color: #1e2538;
color: #93c5fd;
border: 1px solid #3b82f6;
border-radius: 6px;
padding: 5px 12px;
font-size: 11px;
font-weight: 700;
}
QPushButton:hover {
background-color: #2b354d;
}
""")
self.team_board_btn.clicked.connect(self.toggle_team_task_drawer)
th_top_row.addWidget(self.team_board_btn)
self.team_setup_btn = QPushButton("Team Setup")
self.team_setup_btn.setCursor(QCursor(Qt.CursorShape.PointingHandCursor))
self.team_setup_btn.setStyleSheet("""
QPushButton {
background-color: #1e2433;
color: #cbd5e1;
border: 1px solid #334155;
border-radius: 6px;
padding: 5px 12px;
font-size: 11px;
font-weight: 600;
}
QPushButton:hover {
background-color: #2c3447;
color: #ffffff;
}
""")
self.team_setup_btn.clicked.connect(self.open_team_setup_dialog)
th_top_row.addWidget(self.team_setup_btn)
self.team_ws_btn = QPushButton("Team Folder")
self.team_ws_btn.setCursor(QCursor(Qt.CursorShape.PointingHandCursor))
self.team_ws_btn.setStyleSheet("""
QPushButton {
background-color: #1e2433;
color: #cbd5e1;
border: 1px solid #334155;
border-radius: 6px;
padding: 5px 12px;
font-size: 11px;
font-weight: 600;
}
QPushButton:hover {
background-color: #2c3447;
}
""")
self.team_ws_btn.clicked.connect(self.open_team_workspace_in_explorer)
th_top_row.addWidget(self.team_ws_btn)
self.team_clear_btn = QPushButton("Clear")
self.team_clear_btn.setCursor(QCursor(Qt.CursorShape.PointingHandCursor))
self.team_clear_btn.setStyleSheet("""
QPushButton {
background-color: #1e2433;
color: #94a3b8;
border: 1px solid #334155;
border-radius: 6px;
padding: 5px 10px;
font-size: 11px;
font-weight: 600;
}
QPushButton:hover {
background-color: #2c3447;
}
""")
self.team_clear_btn.clicked.connect(self.clear_team_chat)
th_top_row.addWidget(self.team_clear_btn)
th_layout.addLayout(th_top_row)
# 3 Bot Live Status Cards Row
self.bot_cards_layout = QHBoxLayout()
self.bot_cards_layout.setSpacing(8)
self.bot_card_widgets = []
for i in range(3):
card = QFrame()
card.setStyleSheet("background-color: #171b26; border: 1px solid #283042; border-radius: 8px; padding: 6px 10px;")
card_l = QHBoxLayout(card)
card_l.setContentsMargins(6, 4, 6, 4)
card_l.setSpacing(8)
c_av = QLabel("B")
c_av.setFixedSize(28, 28)
c_av.setAlignment(Qt.AlignmentFlag.AlignCenter)
c_av.setStyleSheet("background-color: #6366f1; color: #ffffff; border-radius: 14px; font-weight: 800; font-size: 12px;")
card_l.addWidget(c_av)
info_l = QVBoxLayout()
info_l.setSpacing(1)
c_name = QLabel(f"Bot {i+1}")
c_name.setStyleSheet("color: #ffffff; font-size: 11.5px; font-weight: 700;")
c_role = QLabel("Role")
c_role.setStyleSheet("color: #94a3b8; font-size: 9.5px;")
c_status = QLabel("Idle")
c_status.setStyleSheet("color: #34d399; font-size: 9.5px; font-weight: 600;")
info_l.addWidget(c_name)
info_l.addWidget(c_role)
info_l.addWidget(c_status)
card_l.addLayout(info_l, 1)
self.bot_card_widgets.append((card, c_av, c_name, c_role, c_status))
self.bot_cards_layout.addWidget(card)
th_layout.addLayout(self.bot_cards_layout)
ch_layout.addWidget(self.team_header_widget)
c_layout.addWidget(self.chat_header)
# Collapsible Team Task Board Drawer
self.team_task_drawer = TeamTaskBoardDrawer()
c_layout.addWidget(self.team_task_drawer)
# Chat Messages Scroll Area
self.msg_scroll = QScrollArea()
self.msg_scroll.setWidgetResizable(True)
self.msg_scroll.setFrameShape(QFrame.Shape.NoFrame)
self.msg_scroll.setStyleSheet("""
QScrollArea {
background: transparent;
border: none;
}
QScrollBar:vertical {
background: #0f1117;
width: 7px;
border-radius: 3.5px;
}
QScrollBar::handle:vertical {
background: #282f40;
min-height: 20px;
border-radius: 3.5px;
}
QScrollBar::handle:vertical:hover {
background: #3b455c;
}
QScrollBar::add-line, QScrollBar::sub-line {
background: none;
border: none;
}
""")
self.msg_container = QWidget()
self.msg_layout = QVBoxLayout(self.msg_container)
self.msg_layout.setContentsMargins(6, 6, 6, 6)
self.msg_layout.setSpacing(12)
self.msg_layout.addStretch()
self.msg_scroll.setWidget(self.msg_container)
c_layout.addWidget(self.msg_scroll, 1)
# Active Step Execution Banner
self.exec_banner = QFrame()
self.exec_banner.setVisible(False)
self.exec_banner.setStyleSheet("""
QFrame {
background-color: #171b26;
border: 1px solid #2563eb;
border-radius: 8px;
padding: 6px 12px;
}
""")
eb_layout = QHBoxLayout(self.exec_banner)
eb_layout.setContentsMargins(6, 4, 6, 4)
self.banner_text = QLabel("Step in progress...")
self.banner_text.setStyleSheet("color: #60a5fa; font-size: 12px; font-weight: 600;")
eb_layout.addWidget(self.banner_text)
eb_layout.addStretch()
self.stop_btn = QPushButton("Stop Execution")
self.stop_btn.setCursor(QCursor(Qt.CursorShape.PointingHandCursor))
self.stop_btn.setStyleSheet("""
QPushButton {
background-color: #ef4444;
color: #ffffff;
border: none;
border-radius: 5px;
padding: 4px 10px;
font-size: 11px;
font-weight: 700;
}
QPushButton:hover {
background-color: #dc2626;
}
""")
self.stop_btn.clicked.connect(self.stop_step_execution)
eb_layout.addWidget(self.stop_btn)
c_layout.addWidget(self.exec_banner)
# Message Input Bar
input_frame = QFrame()
input_frame.setStyleSheet("""
QFrame {
background-color: #141720;
border: 1.5px solid #252c3d;
border-radius: 22px;
padding: 4px 8px;
}
""")
inp_layout = QHBoxLayout(input_frame)
inp_layout.setContentsMargins(12, 4, 6, 4)
inp_layout.setSpacing(8)
self.msg_input = QLineEdit()
self.msg_input.setPlaceholderText("Message bot (chat, ask questions, or request file/script actions)...")
self.msg_input.setStyleSheet("""
QLineEdit {
background: transparent;
border: none;
color: #ffffff;
font-size: 13.5px;
font-family: -apple-system, sans-serif;
}
""")
self.msg_input.returnPressed.connect(self.send_user_message)
inp_layout.addWidget(self.msg_input, 1)
self.send_btn = QPushButton("Send")
self.send_btn.setCursor(QCursor(Qt.CursorShape.PointingHandCursor))
self.send_btn.setFixedSize(65, 34)
self.send_btn.setStyleSheet("""
QPushButton {
background: qlineargradient(x1:0, y1:0, x2:1, y2:0, stop:0 #007aff, stop:1 #0a84ff);
color: #ffffff;
border: none;
border-radius: 17px;
font-size: 12.5px;
font-weight: 700;
}
QPushButton:hover {
background: qlineargradient(x1:0, y1:0, x2:1, y2:0, stop:0 #0062cc, stop:1 #0071e3);
}
QPushButton:disabled {
background-color: #1e2433;
color: #475569;
}
""")
self.send_btn.clicked.connect(self.send_user_message)
inp_layout.addWidget(self.send_btn)
c_layout.addWidget(input_frame)
splitter.addWidget(chat_container)
splitter.setSizes([290, 730])
main_layout.addWidget(splitter)
self.update_mode_button_styles()
# ------------------ MODE SWITCHING & TEAM UI ------------------
def update_mode_button_styles(self):
if not self.is_team_mode:
self.mode_solo_btn.setStyleSheet("""
QPushButton {
background-color: #007aff;
color: #ffffff;
border: none;
border-radius: 6px;
padding: 6px 12px;
font-size: 11.5px;
font-weight: 700;
}
""")
self.mode_team_btn.setStyleSheet("""
QPushButton {
background-color: transparent;
color: #94a3b8;
border: none;
border-radius: 6px;
padding: 6px 12px;
font-size: 11.5px;
font-weight: 600;
}
QPushButton:hover {
color: #ffffff;
}
""")
else:
self.mode_solo_btn.setStyleSheet("""
QPushButton {
background-color: transparent;
color: #94a3b8;
border: none;
border-radius: 6px;
padding: 6px 12px;
font-size: 11.5px;
font-weight: 600;
}
QPushButton:hover {
color: #ffffff;
}
""")
self.mode_team_btn.setStyleSheet("""
QPushButton {
background: qlineargradient(x1:0, y1:0, x2:1, y2:0, stop:0 #6366f1, stop:1 #8b5cf6);
color: #ffffff;
border: none;
border-radius: 6px;
padding: 6px 12px;
font-size: 11.5px;
font-weight: 700;
}
""")
def switch_to_solo_mode(self):
self.is_team_mode = False
self.update_mode_button_styles()
self.solo_header_widget.setVisible(True)
self.team_header_widget.setVisible(False)
self.team_task_drawer.setVisible(False)
self.msg_input.setPlaceholderText("Message bot (chat, ask questions, or request file/script actions)...")
self.load_messages()
def switch_to_team_mode(self):
self.is_team_mode = True
self.update_mode_button_styles()
self.solo_header_widget.setVisible(False)
self.team_header_widget.setVisible(True)
self.refresh_team_header_cards()
self.msg_input.setPlaceholderText("Assign a collaborative task to Magi Bot TEAM (up to 3 bots)...")
self.load_team_messages_view()
def refresh_team_header_cards(self):
for i, (card, c_av, c_name, c_role, c_status) in enumerate(self.bot_card_widgets):
if i < len(self.team_bots):
b = self.team_bots[i]
name = b.get("name", f"Bot {i+1}")
color = b.get("color", "#6366f1")
role = b.get("role", "Developer")
model = b.get("model", "deepcoder:14b")
c_av.setText(name[0].upper() if name else "B")
c_av.setStyleSheet(f"background-color: {color}; color: #ffffff; border-radius: 14px; font-weight: 800; font-size: 12px;")
c_name.setText(name)
c_role.setText(f"{role} ({model})")
card.setVisible(True)
else:
card.setVisible(False)
def toggle_team_task_drawer(self):
self.team_task_drawer.setVisible(not self.team_task_drawer.isVisible())
self.team_board_btn.setText("Hide Board" if self.team_task_drawer.isVisible() else "Live Board")
def open_team_setup_dialog(self):
dlg = TeamSetupDialog(self.team_bots, parent=self)
if dlg.exec() == QDialog.DialogCode.Accepted:
self.team_bots = dlg.selected_bots
self.save_team_config()
self.refresh_team_header_cards()
def open_team_workspace_in_explorer(self):
try:
if sys.platform == "win32":
os.startfile(self.team_workspace_dir)
elif sys.platform == "darwin":
subprocess.Popen(["open", self.team_workspace_dir])
else:
subprocess.Popen(["xdg-open", self.team_workspace_dir])
except Exception as e:
QMessageBox.warning(self, "Workspace Error", f"Could not open team workspace directory: {e}")
def clear_team_chat(self):
reply = QMessageBox.question(
self,
"Clear Team Messages",
"Are you sure you want to clear the team conversation history?",
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No
)
if reply == QMessageBox.StandardButton.Yes:
self.team_messages = []
self.save_team_history()
self.load_team_messages_view()
def load_team_messages_view(self):
while self.msg_layout.count() > 1:
item = self.msg_layout.takeAt(0)
if item.widget():
item.widget().deleteLater()
for m in self.team_messages:
sender = m.get("sender_name", "Bot")
is_user = m.get("is_user", False)
bubble = IMessageBubble(
sender_name=sender,
text=m.get("text", ""),
is_user=is_user,
timestamp=m.get("timestamp", ""),
action=m.get("action"),
action_result=m.get("action_result"),
role=m.get("role", ""),
color=m.get("color", "")
)
self.msg_layout.insertWidget(self.msg_layout.count() - 1, bubble)
self.scroll_to_bottom()
# ------------------ SIDEBAR & BOT LIST MANAGEMENT ------------------
def refresh_bot_list(self):
bots = bot_manager.get_all_bots()
self.count_badge.setText(f"{len(bots)}/{MAX_BOTS}")
self.new_bot_btn.setEnabled(len(bots) < MAX_BOTS)
while self.contact_list_layout.count() > 1:
item = self.contact_list_layout.takeAt(0)
if item.widget():
item.widget().deleteLater()
for b in bots:
is_active = (b.get("id") == self.active_bot_id)
item = BotContactItem(b, is_active=is_active)
item.selected.connect(self.select_bot)
item.edit_requested.connect(self.open_edit_bot_dialog)
item.delete_requested.connect(self.delete_bot_confirm)
self.contact_list_layout.insertWidget(self.contact_list_layout.count() - 1, item)
if not self.active_bot_id and bots:
self.select_bot(bots[0]["id"])
def filter_bots(self, text: str):
query = text.strip().lower()
for i in range(self.contact_list_layout.count() - 1):
w = self.contact_list_layout.itemAt(i).widget()
if isinstance(w, BotContactItem):
name = w.bot_data.get("name", "").lower()
model = w.bot_data.get("model", "").lower()
w.setVisible(query in name or query in model)
def select_bot(self, bot_id: str):
self.active_bot_id = bot_id
bot = bot_manager.get_bot(bot_id)
if not bot:
return
for i in range(self.contact_list_layout.count() - 1):
w = self.contact_list_layout.itemAt(i).widget()
if isinstance(w, BotContactItem):
w.is_active = (w.bot_data.get("id") == bot_id)
w.update_style()
name = bot.get("name", "Bot")
color = bot.get("color", "#007aff")
initial = name[0].upper() if name else "B"
self.hdr_avatar.setText(initial)
self.hdr_avatar.setStyleSheet(f"""
background-color: {color};
color: #ffffff;
font-size: 15px;
font-weight: 800;
border-radius: 18px;
""")
self.hdr_name.setText(name)
ws_name = f"Magi({name})"
self.hdr_sub.setText(f"Model: {bot.get('model', 'deepcoder:14b')} (local Ollama >=7B) | Workspace: {ws_name}")
self.queued_messages.clear()
self.update_queue_ui()
if not self.is_team_mode:
self.load_messages()
def open_create_bot_dialog(self):
max_allowed = MAX_BOTS
current_count = len(bot_manager.get_all_bots())
if current_count >= max_allowed:
QMessageBox.information(
self,
"Limit Reached",
f"You have reached the maximum of {max_allowed} bots."
)
return
dlg = BotConfigDialog(parent=self)
if dlg.exec() == QDialog.DialogCode.Accepted:
self.refresh_bot_list()
bots = bot_manager.get_all_bots()
if bots:
self.select_bot(bots[-1]["id"])
def open_edit_bot_dialog(self, bot_id: str):
bot = bot_manager.get_bot(bot_id)
if not bot:
return
dlg = BotConfigDialog(bot_data=bot, parent=self)
if dlg.exec() == QDialog.DialogCode.Accepted:
self.refresh_bot_list()
self.select_bot(bot_id)
def delete_bot_confirm(self, bot_id: str):
bot = bot_manager.get_bot(bot_id)
if not bot:
return
reply = QMessageBox.question(
self,
"Delete Bot",
f"Are you sure you want to delete bot '{bot.get('name')}'?",
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No
)
if reply == QMessageBox.StandardButton.Yes:
bot_manager.delete_bot(bot_id)
if self.active_bot_id == bot_id:
self.active_bot_id = None
self.refresh_bot_list()
def open_active_workspace_in_explorer(self):
if not self.active_bot_id:
return
bot = bot_manager.get_bot(self.active_bot_id)
if not bot:
return
ws_dir = bot.get("workspace_dir", get_bot_workspace_dir(bot.get("name", "Bot")))
try:
if sys.platform == "win32":
os.startfile(ws_dir)
elif sys.platform == "darwin":
subprocess.Popen(["open", ws_dir])
else:
subprocess.Popen(["xdg-open", ws_dir])
except Exception as e:
QMessageBox.warning(self, "Workspace Error", f"Could not open workspace directory: {e}")
def clear_active_chat(self):
if not self.active_bot_id:
return
reply = QMessageBox.question(
self,
"Clear Messages",
"Are you sure you want to clear this conversation history?",
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No
)
if reply == QMessageBox.StandardButton.Yes:
bot_manager.clear_messages(self.active_bot_id)
self.load_messages()
# ------------------ CHAT RENDERING ------------------
def load_messages(self):
while self.msg_layout.count() > 1:
item = self.msg_layout.takeAt(0)
if item.widget():
item.widget().deleteLater()
if not self.active_bot_id:
return
bot = bot_manager.get_bot(self.active_bot_id)
if not bot:
return
messages = bot.get("messages", [])
for m in messages:
sender = m.get("sender", "bot")
is_user = (sender == "user")
bubble = IMessageBubble(
sender_name="You" if is_user else bot.get("name", "Bot"),
text=m.get("text", ""),
is_user=is_user,
timestamp=m.get("timestamp", ""),
action=m.get("action"),
action_result=m.get("action_result")
)
self.msg_layout.insertWidget(self.msg_layout.count() - 1, bubble)
self.scroll_to_bottom()
def scroll_to_bottom(self):
QTimer.singleShot(50, lambda: self.msg_scroll.verticalScrollBar().setValue(
self.msg_scroll.verticalScrollBar().maximum()
))
# ------------------ MESSAGE SENDING & STEP ORCHESTRATION ------------------
def send_user_message(self, forced_text: str = None):
if forced_text:
text = forced_text.strip()
else:
text = self.msg_input.text().strip()
self.msg_input.clear()
if not text:
return
# TEAM MODE HANDLING
if self.is_team_mode:
if self.team_worker and self.team_worker.isRunning():
QMessageBox.information(self, "Team Active", "Magi Bot TEAM is currently running a task. Please wait.")
return
user_msg = {
"sender_name": "You",
"text": text,
"is_user": True,
"timestamp": datetime.datetime.now().strftime("%H:%M"),
"action": None,
"action_result": None
}
self.team_messages.append(user_msg)
self.save_team_history()
user_bubble = IMessageBubble("You", text, is_user=True)
self.msg_layout.insertWidget(self.msg_layout.count() - 1, user_bubble)
self.scroll_to_bottom()
self.start_team_execution(text)
return
# SOLO BOT HANDLING
if not self.active_bot_id:
return
if self.active_worker and self.active_worker.isRunning():
if len(self.queued_messages) >= self.max_queue_size:
QMessageBox.information(
self,
"Queue Limit Reached",
f"Message queue is full ({self.max_queue_size}/{self.max_queue_size}). Please wait until the bot completes its task."
)
return
self.queued_messages.append(text)
self.update_queue_ui()
return
bot_manager.add_message(self.active_bot_id, "user", text)
user_bubble = IMessageBubble("You", text, is_user=True)
self.msg_layout.insertWidget(self.msg_layout.count() - 1, user_bubble)
self.scroll_to_bottom()
self.current_user_task = text
self.step_counter = 1
self.execute_step(prompt_for_step=text)
def start_team_execution(self, goal: str):
self.exec_banner.setVisible(True)
coord_model = get_team_coordinator_model()
self.banner_text.setText(f"Magi Bot TEAM: Decomposing goal with {coord_model}...")
self.team_task_drawer.setVisible(True)
self.team_board_btn.setText("Hide Board")
self.team_worker = MagiBotTeamWorker(
team_bots=self.team_bots,
goal=goal,
workspace_dir=self.team_workspace_dir,
gemini_key=self.gemini_key,
mistral_key=self.mistral_key,
parent=self
)
self.team_worker.team_status_updated.connect(self.handle_team_status_updated)
self.team_worker.bot_status_updated.connect(self.handle_bot_status_updated)
self.team_worker.bot_message_posted.connect(self.handle_team_message_posted)
self.team_worker.task_board_updated.connect(self.handle_task_board_updated)
self.team_worker.team_finished.connect(self.handle_team_finished)
self.team_worker.error_occurred.connect(self.handle_team_error)
self.team_worker.start()
def handle_team_status_updated(self, status_text: str):
self.banner_text.setText(status_text)
def handle_bot_status_updated(self, bot_name: str, status_text: str):
for card, c_av, c_name, c_role, c_status in self.bot_card_widgets:
if c_name.text().lower() == bot_name.lower():
c_status.setText(status_text)
break
def handle_team_message_posted(self, payload: dict):
self.team_messages.append(payload)
self.save_team_history()
bubble = IMessageBubble(
sender_name=payload.get("sender_name", "Bot"),
text=payload.get("text", ""),
is_user=payload.get("is_user", False),
action=payload.get("action"),
action_result=payload.get("action_result"),
role=payload.get("role", ""),
color=payload.get("color", "")
)
self.msg_layout.insertWidget(self.msg_layout.count() - 1, bubble)
self.scroll_to_bottom()
def handle_task_board_updated(self, board_dict: dict):
tasks = board_dict.get("tasks", [])
locks = board_dict.get("locked_files", {})
self.team_task_drawer.update_board(tasks, locks)
def handle_team_finished(self, final_summary: str):
self.exec_banner.setVisible(False)
for _, _, _, _, c_status in self.bot_card_widgets:
c_status.setText("Idle")
self.team_worker = None
def handle_team_error(self, err_msg: str):
self.exec_banner.setVisible(False)
bubble = IMessageBubble("System", f"Team Error: {err_msg}", is_user=False)
self.msg_layout.insertWidget(self.msg_layout.count() - 1, bubble)
self.scroll_to_bottom()
self.team_worker = None
def update_queue_ui(self):
q_count = len(self.queued_messages)
if q_count > 0:
self.queue_badge.setVisible(True)
self.queue_badge.setText(f"Queue: {q_count}/{self.max_queue_size}")
if self.exec_banner.isVisible():
self.banner_text.setText(f"Step {self.step_counter} in progress... [{q_count} in queue]")
else:
self.queue_badge.setVisible(False)
if self.exec_banner.isVisible():
bot = bot_manager.get_bot(self.active_bot_id)
model_name = bot.get("model", "model") if bot else "model"
self.banner_text.setText(f"Processing with {model_name}...")
def is_working(self) -> bool:
"""Returns True if a Solo Bot or Team Bot is actively executing steps in the background."""
if self.is_team_mode:
return bool(self.team_worker and self.team_worker.isRunning())
return bool(self.active_worker and self.active_worker.isRunning())
def on_view_restored(self):
"""Called whenever the user navigates back to the Magi Bot screen."""
self.refresh_bot_list()
if self.is_team_mode:
self.load_team_messages_view()
self.refresh_team_header_cards()
if self.team_worker and self.team_worker.isRunning():
self.exec_banner.setVisible(True)
self.team_task_drawer.setVisible(True)
self.team_board_btn.setText("Hide Board")
else:
self.exec_banner.setVisible(False)
else:
if self.active_bot_id:
self.load_messages()
if self.active_worker and self.active_worker.isRunning():
self.exec_banner.setVisible(True)
self.step_badge.setVisible(True)
self.update_queue_ui()
else:
self.exec_banner.setVisible(False)
self.step_badge.setVisible(False)
def execute_step(self, prompt_for_step: str):
bot = bot_manager.get_bot(self.active_bot_id)
if not bot:
return
self.working_bot_id = self.active_bot_id
self.exec_banner.setVisible(True)
self.banner_text.setText(f"Processing with {bot.get('model')}...")
self.step_badge.setVisible(True)
self.step_badge.setText(f"Step {self.step_counter}")
self.send_btn.setEnabled(True)
self.update_queue_ui()
self.active_worker = MagiBotWorker(
bot_info=bot,
user_original_prompt=self.current_user_task,
current_step_prompt=prompt_for_step,
step_number=self.step_counter,
gemini_key=self.gemini_key,
mistral_key=self.mistral_key,
parent=self
)
self.active_worker.confirmation_requested.connect(self.handle_action_confirmation)
self.active_worker.step_finished.connect(self.handle_step_finished)
self.active_worker.error_occurred.connect(self.handle_step_error)
self.active_worker.start()
def handle_action_confirmation(self, req: dict):
dlg = BotActionConfirmDialog(
bot_name=req.get("bot_name", "Bot"),
workspace_dir=req.get("workspace_dir", ""),
action=req.get("action", {}),
step_num=req.get("step_number", 1),
parent=self
)
res = dlg.exec()
approved = (res == QDialog.DialogCode.Accepted)
if self.active_worker:
self.active_worker.set_confirmation_response(approved)
def handle_step_finished(self, payload: dict):
target_bot_id = getattr(self, "working_bot_id", self.active_bot_id) or self.active_bot_id
bot = bot_manager.get_bot(target_bot_id)
bot_name = bot.get("name", "Bot") if bot else "Bot"
bot_text = payload.get("bot_text", "")
action = payload.get("action")
action_result = payload.get("action_result")
is_task_completed = payload.get("is_task_completed", True)
next_prompt = payload.get("next_prompt", "")
granite_summary = payload.get("granite_summary", "")
bot_manager.add_message(
target_bot_id,
"bot",
bot_text,
action=action,
action_result=action_result
)
# Only insert bubble into UI if the user is currently viewing this bot in solo mode
if not self.is_team_mode and self.active_bot_id == target_bot_id:
bot_bubble = IMessageBubble(
sender_name=bot_name,
text=bot_text,
is_user=False,
action=action,
action_result=action_result
)
self.msg_layout.insertWidget(self.msg_layout.count() - 1, bot_bubble)
self.scroll_to_bottom()
if action and not is_task_completed and next_prompt:
self.step_counter += 1
self.banner_text.setText(f"Granite 4.1 scheduled next step (Step {self.step_counter}). Executing in background...")
QTimer.singleShot(600, lambda: self.execute_step(prompt_for_step=next_prompt))
else:
self.finish_step_loop(granite_summary)
def handle_step_error(self, error_msg: str):
bot_bubble = IMessageBubble("System", f"Execution error: {error_msg}", is_user=False)
self.msg_layout.insertWidget(self.msg_layout.count() - 1, bot_bubble)
self.scroll_to_bottom()
self.finish_step_loop("Encountered error.")
def finish_step_loop(self, summary_msg: str = ""):
self.exec_banner.setVisible(False)
self.step_badge.setVisible(False)
self.send_btn.setEnabled(True)
self.current_user_task = ""
self.step_counter = 1
self.active_worker = None
if self.queued_messages:
selected_msg = random.choice(self.queued_messages)
self.queued_messages.clear()
self.update_queue_ui()
QTimer.singleShot(400, lambda: self.send_user_message(forced_text=selected_msg))
else:
self.update_queue_ui()
def stop_step_execution(self):
if self.is_team_mode:
if self.team_worker and self.team_worker.isRunning():
self.team_worker.cancel()
self.handle_team_finished("Stopped by user.")
return
if self.active_worker and self.active_worker.isRunning():
self.active_worker.cancel()
self.queued_messages.clear()
self.update_queue_ui()
self.finish_step_loop("Stopped by user.")