import os
import sys
import json
import time
import re
import shutil
import asyncio
import html
import signal
import logging
from pathlib import Path
from typing import Optional, Dict, Any, Callable, List, Tuple
from dataclasses import dataclass, field, asdict
from config import settings
logger = logging.getLogger(__name__)
def escape_html(text: str) -> str:
"""Safely escapes text for Telegram HTML parse mode."""
if not text:
return ""
return html.escape(str(text), quote=False)
# Available models
AVAILABLE_MODELS = {
"gemini-3.7-flash-auto": "Gemini 3.7 Flash Auto (پیشفرض هوشمند ⚡/🧠)",
"gemini-3.7-flash": "Gemini 3.7 Flash",
"gemini-3.7-flash-high": "Gemini 3.7 Flash (High Reasoning)",
"gemini-3.6-flash": "Gemini 3.6 Flash",
"gemini-3.1-pro": "Gemini 3.1 Pro (Deep Architecture)",
"claude-sonnet-4-6": "Claude Sonnet 4.6 (Thinking)",
"claude-opus-4-6-thinking": "Claude Opus 4.6 (Thinking)",
"gpt-oss-120b": "GPT-OSS 120B (Medium)",
}
AVAILABLE_EFFORTS = ["low", "medium", "high"]
def get_valid_effort_for_model(model_name: Optional[str], requested_effort: Optional[str]) -> Optional[str]:
"""Validates and maps reasoning effort to what the specific model actually supports in AGY CLI."""
if not model_name:
return None
m = model_name.lower()
# Auto model starts with low effort (or uses requested if specified)
if "flash-auto" in m or m == "gemini-3.7-flash-auto":
if requested_effort and requested_effort.lower() in ("low", "medium", "high"):
return requested_effort.lower()
return "low"
if not requested_effort:
return None
e = requested_effort.lower()
# Claude models do NOT support --effort (they are inherently thinking models)
if "claude" in m:
return None
# GPT models only support medium
if "gpt" in m:
return "medium"
# gemini-3.7-flash-high only supports high
if "flash-high" in m:
return "high"
# gemini-3.1-pro only supports low or high
if "3.1-pro" in m:
return "high" if e in ("high", "medium") else "low"
# Standard gemini models (3.7-flash, 3.6-flash, 3.5-flash)
if e in ("low", "medium", "high"):
return e
return None
# Available languages
AVAILABLE_LANGUAGES = {
"fa": "🇮🇷 Persian / Farsi (فارسی)",
"en": "🇬🇧 English",
"auto": "🌐 Auto (Detect)",
"id": "🇮🇩 Indonesian (Bahasa Indonesia)",
"es": "🇪🇸 Spanish (Español)",
"zh": "🇨🇳 Chinese (中文)",
"ja": "🇯🇵 Japanese (日本語)",
"de": "🇩🇪 German (Deutsch)",
"fr": "🇫🇷 French (Français)",
"ru": "🇷🇺 Russian (Русский)",
"ar": "🇸🇦 Arabic (العربية)",
"pt": "🇧🇷 Portuguese (Português)",
"ko": "🇰🇷 Korean (한국어)",
"it": "🇮🇹 Italian (Italiano)",
"nl": "🇳🇱 Dutch (Nederlands)",
"tr": "🇹🇷 Turkish (Türkçe)",
"vi": "🇻🇳 Vietnamese (Tiếng Việt)",
}
@dataclass
class AgentResult:
text: str
usage: Optional[Dict[str, Any]] = None
duration: float = 0.0
conversation_id: Optional[str] = None
executed_model: Optional[str] = None
executed_effort: Optional[str] = None
@dataclass
class Project:
name: str
workspace: str = field(default_factory=lambda: settings.default_workspace)
owner_id: Optional[int] = None
shared_with: List[int] = field(default_factory=list)
conversation_id: Optional[str] = None
conversation_history: List[str] = field(default_factory=list)
model: str = field(default_factory=lambda: settings.default_model)
effort: str = field(default_factory=lambda: settings.default_effort)
language: str = field(default_factory=lambda: settings.default_language)
last_context_length: Optional[int] = None
last_total_tokens: Optional[int] = None
last_response: Optional[str] = None
created_at: float = field(default_factory=time.time)
description: str = ""
conversation_titles: Dict[str, str] = field(default_factory=dict)
@dataclass
class Session:
chat_id: int
active_project: Optional[str] = None
projects: Dict[str, Project] = field(default_factory=dict)
turn_in_progress: bool = False
last_prompt: Optional[str] = None
last_response: Optional[str] = None
last_status_msg_id: Optional[int] = None
restart_pending: bool = False
last_delivered: bool = True
last_update_time: float = 0.0
language: str = field(default_factory=lambda: settings.default_language)
@property
def current_project(self) -> Optional[Project]:
return session_manager.get_current_project(self.chat_id)
@current_project.setter
def current_project(self, val: Any):
if isinstance(val, Project):
self.active_project = val.name
elif isinstance(val, str):
self.active_project = val
elif val is None:
self.active_project = None
@property
def conversation_id(self) -> Optional[str]:
curr = self.current_project
return curr.conversation_id if curr else None
@conversation_id.setter
def conversation_id(self, val: Optional[str]):
curr = self.current_project
if curr:
curr.conversation_id = val
@property
def workspace(self) -> str:
curr = self.current_project
return curr.workspace if curr else settings.default_workspace
@workspace.setter
def workspace(self, val: str):
curr = self.current_project
if curr:
curr.workspace = val
@property
def model(self) -> str:
curr = self.current_project
return curr.model if curr else settings.default_model
@model.setter
def model(self, val: str):
curr = self.current_project
if curr:
curr.model = val
@property
def effort(self) -> str:
curr = self.current_project
return curr.effort if curr else settings.default_effort
@effort.setter
def effort(self, val: str):
curr = self.current_project
if curr:
curr.effort = val
@property
def last_context_length(self) -> Optional[int]:
curr = self.current_project
return curr.last_context_length if curr else None
@last_context_length.setter
def last_context_length(self, val: Optional[int]):
curr = self.current_project
if curr:
curr.last_context_length = val
@property
def last_total_tokens(self) -> Optional[int]:
curr = self.current_project
return curr.last_total_tokens if curr else None
@last_total_tokens.setter
def last_total_tokens(self, val: Optional[int]):
curr = self.current_project
if curr:
curr.last_total_tokens = val
class SessionManager:
def __init__(self, storage_file: Path = Path("/root/telegram-agy-bot/sessions.json")):
self.storage_file = storage_file
self.sessions: Dict[int, Session] = {}
self.active_tasks: Dict[int, asyncio.Task] = {}
self.active_procs: Dict[int, asyncio.subprocess.Process] = {}
self._load()
def _load(self):
if self.storage_file.exists():
try:
with open(self.storage_file, "r", encoding="utf-8") as f:
data = json.load(f)
for k, v in data.items():
if not isinstance(v, dict):
continue
uid = int(k)
is_admin_user = settings.is_admin(uid)
projects_dict: Dict[str, Project] = {}
if "projects" in v and isinstance(v["projects"], dict):
for p_name, p_data in v["projects"].items():
if isinstance(p_data, dict):
clean_p = {
k2: v2
for k2, v2 in p_data.items()
if k2 in (
"name",
"workspace",
"owner_id",
"shared_with",
"conversation_id",
"conversation_history",
"model",
"effort",
"language",
"last_context_length",
"last_total_tokens",
"last_response",
"created_at",
"description",
"conversation_titles",
)
}
if "name" not in clean_p:
clean_p["name"] = p_name
if "owner_id" not in clean_p or clean_p["owner_id"] is None:
clean_p["owner_id"] = uid
if "shared_with" not in clean_p or not isinstance(clean_p["shared_with"], list):
clean_p["shared_with"] = []
if "conversation_history" not in clean_p or not isinstance(clean_p["conversation_history"], list):
clean_p["conversation_history"] = []
if clean_p.get("conversation_id") and clean_p["conversation_id"] not in clean_p["conversation_history"]:
clean_p["conversation_history"].append(clean_p["conversation_id"])
if "conversation_titles" not in clean_p or not isinstance(clean_p["conversation_titles"], dict):
clean_p["conversation_titles"] = {}
# Non-admin users cannot own or access the "default" /root project
if not is_admin_user and (p_name == "default" or clean_p.get("workspace") == settings.default_workspace):
continue
projects_dict[p_name] = Project(**clean_p)
# For admin users, if no projects exist, initialize "default" project
if is_admin_user and "default" not in projects_dict and not projects_dict:
default_ws = v.get("workspace", settings.default_workspace)
init_cid = v.get("conversation_id")
projects_dict["default"] = Project(
name="default",
workspace=default_ws,
owner_id=uid,
shared_with=[],
conversation_id=init_cid,
conversation_history=[init_cid] if init_cid else [],
model=v.get("model", settings.default_model),
effort=v.get("effort", settings.default_effort),
language=v.get("language", settings.default_language),
last_context_length=v.get("last_context_length"),
last_total_tokens=v.get("last_total_tokens"),
last_response=v.get("last_response"),
)
active_p = v.get("active_project")
if not is_admin_user and active_p == "default":
active_p = None
if active_p and active_p not in projects_dict:
active_p = list(projects_dict.keys())[0] if projects_dict else None
elif not active_p and projects_dict:
active_p = list(projects_dict.keys())[0]
session = Session(
chat_id=uid,
active_project=active_p,
projects=projects_dict,
turn_in_progress=v.get("turn_in_progress", False),
last_prompt=v.get("last_prompt"),
last_response=v.get("last_response"),
last_status_msg_id=v.get("last_status_msg_id"),
restart_pending=v.get("restart_pending", False),
last_delivered=v.get("last_delivered", True),
last_update_time=v.get("last_update_time", 0.0),
language=v.get("language", settings.default_language),
)
self.sessions[uid] = session
except Exception as e:
logger.error(f"Error loading sessions: {e}")
def save(self):
try:
data = {}
for k, sess in self.sessions.items():
projects_serialized = {}
for p_name, p_obj in sess.projects.items():
projects_serialized[p_name] = {
"name": p_obj.name,
"workspace": p_obj.workspace,
"owner_id": p_obj.owner_id,
"shared_with": p_obj.shared_with,
"conversation_id": p_obj.conversation_id,
"conversation_history": p_obj.conversation_history,
"model": p_obj.model,
"effort": p_obj.effort,
"language": p_obj.language,
"last_context_length": p_obj.last_context_length,
"last_total_tokens": p_obj.last_total_tokens,
"last_response": p_obj.last_response,
"created_at": p_obj.created_at,
"description": p_obj.description,
"conversation_titles": p_obj.conversation_titles,
}
curr = self.get_current_project(sess.chat_id)
data[str(k)] = {
"chat_id": sess.chat_id,
"active_project": sess.active_project,
"projects": projects_serialized,
"turn_in_progress": sess.turn_in_progress,
"last_prompt": sess.last_prompt,
"last_response": sess.last_response,
"last_status_msg_id": sess.last_status_msg_id,
"restart_pending": sess.restart_pending,
"last_delivered": sess.last_delivered,
"last_update_time": sess.last_update_time,
# Top-level compatibility
"workspace": curr.workspace if curr else settings.default_workspace,
"conversation_id": curr.conversation_id if curr else None,
"model": curr.model if curr else settings.default_model,
"effort": curr.effort if curr else settings.default_effort,
"language": curr.language if curr else sess.language,
"last_context_length": curr.last_context_length if curr else None,
"last_total_tokens": curr.last_total_tokens if curr else None,
}
tmp_file = f"{self.storage_file}.tmp.{os.getpid()}_{int(time.time()*1000)}"
with open(tmp_file, "w", encoding="utf-8") as f:
json.dump(data, f, indent=2)
os.replace(tmp_file, self.storage_file)
except Exception as e:
logger.error(f"Error saving sessions: {e}")
def get_or_create(self, chat_id: int) -> Session:
is_admin_user = settings.is_admin(chat_id)
if chat_id not in self.sessions:
if is_admin_user:
default_proj = Project(
name="default",
workspace=settings.default_workspace,
owner_id=chat_id,
shared_with=[],
model=settings.default_model,
effort=settings.default_effort,
language=settings.default_language,
)
sess = Session(
chat_id=chat_id,
active_project="default",
projects={"default": default_proj},
language=settings.default_language,
)
else:
sess = Session(
chat_id=chat_id,
active_project=None,
projects={},
language=settings.default_language,
)
self.sessions[chat_id] = sess
self.save()
else:
sess = self.sessions[chat_id]
if not is_admin_user:
if "default" in sess.projects and sess.projects["default"].workspace == settings.default_workspace:
del sess.projects["default"]
if sess.active_project == "default":
acc = self.get_all_accessible_projects(chat_id)
sess.active_project = list(acc.keys())[0] if acc else None
self.save()
elif is_admin_user and not sess.projects:
sess.projects["default"] = Project(
name="default",
workspace=settings.default_workspace,
owner_id=chat_id,
shared_with=[],
model=settings.default_model,
effort=settings.default_effort,
language=settings.default_language,
)
if not sess.active_project:
sess.active_project = "default"
self.save()
return self.sessions[chat_id]
def sanitize_project_name(self, name: str) -> str:
clean = re.sub(r"[^\w\-]", "-", name.strip().lower())
clean = re.sub(r"-+", "-", clean).strip("-")
return clean or "project"
def get_user_projects(self, chat_id: int) -> Dict[str, Project]:
"""Returns all projects created and owned by chat_id."""
sess = self.sessions.get(chat_id)
if not sess:
return {}
is_admin_user = settings.is_admin(chat_id)
result = {}
for name, p in sess.projects.items():
if name == "default" and not is_admin_user:
continue
result[name] = p
return result
def get_shared_projects(self, chat_id: int) -> Dict[str, Project]:
"""Returns all projects owned by other users that are shared with chat_id."""
result = {}
for other_id, other_sess in self.sessions.items():
if other_id == chat_id:
continue
for p_name, p in other_sess.projects.items():
if chat_id in p.shared_with:
result[p_name] = p
return result
def get_all_accessible_projects(self, chat_id: int) -> Dict[str, Project]:
"""Returns all projects accessible to chat_id (owned, shared, and default for admin)."""
own = self.get_user_projects(chat_id)
shared = self.get_shared_projects(chat_id)
combined = dict(own)
for k, v in shared.items():
if k not in combined:
combined[k] = v
else:
combined[f"{k} (shared:{v.owner_id})"] = v
return combined
def get_current_project(self, chat_id: int) -> Optional[Project]:
"""Returns the currently active Project object for chat_id, or None if no project is active."""
if chat_id not in self.sessions:
self.get_or_create(chat_id)
sess = self.sessions[chat_id]
accessible = self.get_all_accessible_projects(chat_id)
if not accessible:
sess.active_project = None
return None
if sess.active_project and sess.active_project in accessible:
return accessible[sess.active_project]
if sess.active_project:
for k, p in accessible.items():
if k.lower() == sess.active_project.lower() or p.name.lower() == sess.active_project.lower():
sess.active_project = k
self.save()
return p
# Fallback to first accessible project
first_key = list(accessible.keys())[0]
sess.active_project = first_key
self.save()
return accessible[first_key]
async def create_project(
self,
chat_id: int,
name: str,
workspace: Optional[str] = None,
model: Optional[str] = None,
effort: Optional[str] = None,
language: Optional[str] = None,
description: str = "",
) -> Project:
session = self.get_or_create(chat_id)
proj_name = self.sanitize_project_name(name)
is_admin_user = settings.is_admin(chat_id)
if not is_admin_user and proj_name == "default":
proj_name = "my-project"
# Workspace directory: /root/projects/{chat_id}/{proj_name}
if not is_admin_user or not workspace:
ws_path = str(Path("/root/projects") / str(chat_id) / proj_name)
else:
ws_path = str(Path(workspace).expanduser().resolve())
os.makedirs(ws_path, exist_ok=True)
curr = self.get_current_project(chat_id)
curr_lang = curr.language if curr else (session.language or settings.default_language)
proj = Project(
name=proj_name,
workspace=ws_path,
owner_id=chat_id,
shared_with=[],
conversation_id=None,
model=model or settings.default_model,
effort=effort or settings.default_effort,
language=language or curr_lang,
created_at=time.time(),
description=description,
)
session.projects[proj_name] = proj
session.active_project = proj_name
self.save()
# Initialize Git repository and link with Gitea
try:
from git_manager import git_manager
await git_manager.init_project_repo(ws_path, proj_name)
except Exception as ge:
logger.warning(f"Failed to auto-init git repository for {proj_name}: {ge}")
# Auto-ingest and cleanup AGENTS.md if present
try:
from memory_manager import memory_manager
memory_manager.ingest_and_cleanup_agents_md(
workspace_path=ws_path,
user_id=chat_id,
project_name=proj_name,
is_admin=is_admin_user,
)
except Exception as mex:
logger.warning(f"Failed to ingest AGENTS.md in create_project: {mex}")
return proj
async def switch_project(self, chat_id: int, name: str) -> Optional[Project]:
session = self.get_or_create(chat_id)
clean_target = name.strip()
accessible = self.get_all_accessible_projects(chat_id)
target_p = None
if clean_target in accessible:
self.cancel_active_task(chat_id)
session.active_project = clean_target
self.save()
target_p = accessible[clean_target]
else:
for k, p in accessible.items():
if k.lower() == clean_target.lower() or p.name.lower() == clean_target.lower():
self.cancel_active_task(chat_id)
session.active_project = k
self.save()
target_p = p
break
if target_p:
# Auto-ingest and cleanup AGENTS.md if present
try:
from memory_manager import memory_manager
is_admin_user = settings.is_admin(chat_id)
memory_manager.ingest_and_cleanup_agents_md(
workspace_path=target_p.workspace,
user_id=chat_id,
project_name=target_p.name,
is_admin=is_admin_user,
)
except Exception as mex:
logger.warning(f"Failed to ingest AGENTS.md in switch_project: {mex}")
return target_p
return None
async def delete_project(self, chat_id: int, name: str, delete_files: bool = True) -> tuple[bool, str]:
session = self.get_or_create(chat_id)
is_admin_user = settings.is_admin(chat_id)
proj_name = name.strip()
matched_key = None
for k in session.projects.keys():
if k.lower() == proj_name.lower():
matched_key = k
break
if not matched_key:
shared = self.get_shared_projects(chat_id)
for k, p in shared.items():
if k.lower() == proj_name.lower() or p.name.lower() == proj_name.lower():
return False, "❌ شما فقط عضو مشترک این پروژه هستید و امکان حذف آن را ندارید."
return False, f"❌ پروژه {proj_name} یافت نشد."
if matched_key == "default" and not is_admin_user:
return False, "❌ امکان حذف پروژه سیستم وجود ندارد."
self.cancel_active_task(chat_id)
proj_obj = session.projects.pop(matched_key)
cleaned_details = []
# 1. Terminate running background processes / open ports in workspace
if proj_obj.workspace and proj_obj.workspace != "/root" and os.path.isdir(proj_obj.workspace):
try:
os.system(f"fuser -k -9 {proj_obj.workspace} > /dev/null 2>&1")
cleaned_details.append("🛑 فرآیندها و پورتهای اجرایی پروژه متوقف شدند.")
except Exception as pe:
logger.debug(f"Process cleanup warning: {pe}")
# 2. Stop and disable any systemd service if created for this project
service_file = Path(f"/etc/systemd/system/{matched_key}.service")
if service_file.exists():
try:
os.system(f"systemctl stop {matched_key} && systemctl disable {matched_key} > /dev/null 2>&1")
service_file.unlink(missing_ok=True)
os.system("systemctl daemon-reload > /dev/null 2>&1")
cleaned_details.append(f"⚙️ سرویس سیستمی {matched_key}.service متوقف و حذف شد.")
except Exception as se:
logger.warning(f"Failed to remove systemd service: {se}")
# 3. Release and remove Caddy reverse proxy & subdomain if configured
try:
from bot_actions import remove_caddy_reverse_proxy
caddy_ok, caddy_msg = remove_caddy_reverse_proxy(matched_key)
if caddy_ok:
cleaned_details.append(f"🌐 سابدامین و پراکسی Caddy ({matched_key}.msa.artacloud.ir) آزاد و حذف شد.")
except Exception as ce:
logger.debug(f"Caddy cleanup check: {ce}")
# 4. Cancel all scheduled tasks / cron jobs associated with this project
try:
from scheduler import task_scheduler
removed_tasks = 0
for t_id, task in list(task_scheduler.tasks.items()):
if task.project_name.lower() == matched_key.lower() or (proj_obj.workspace and proj_obj.workspace in task.content):
await task_scheduler.delete_task(t_id)
removed_tasks += 1
if removed_tasks > 0:
cleaned_details.append(f"⏰ تعداد {removed_tasks} تسک زمانبندیشده مربوط به پروژه لغو گردید.")
except Exception as te:
logger.debug(f"Tasks cleanup check: {te}")
# 5. Delete Gitea remote repository
try:
from git_manager import git_manager
g_ok, g_msg = await git_manager.delete_gitea_repo(matched_key)
if g_ok:
cleaned_details.append("🐙 مخزن اختصاصی در سرور گیت (Gitea) حذف شد.")
except Exception as ge:
logger.warning(f"Failed to delete Gitea repo {matched_key}: {ge}")
# 6. Delete physical directory if requested and not /root
if delete_files and proj_obj.workspace and proj_obj.workspace != "/root" and os.path.isdir(proj_obj.workspace):
try:
shutil.rmtree(proj_obj.workspace, ignore_errors=True)
cleaned_details.append(f"🗑️ پوشه فایلها ({proj_obj.workspace}) پاکسازی شد.")
except Exception as e:
logger.warning(f"Failed to remove project dir {proj_obj.workspace}: {e}")
accessible = self.get_all_accessible_projects(chat_id)
if session.active_project == matched_key:
session.active_project = list(accessible.keys())[0] if accessible else None
self.save()
details_str = "\n".join([f"• {item}" for item in cleaned_details])
out_msg = (
f"✅ پروژه {matched_key} با موفقیت به صورت کامل و هوشمند حذف گردید:\n\n"
f"{details_str}"
)
return True, out_msg
async def rename_project(self, chat_id: int, old_name: str, new_name: str) -> tuple[Optional[Project], str]:
session = self.get_or_create(chat_id)
is_admin_user = settings.is_admin(chat_id)
old_matched = None
for k in session.projects.keys():
if k.lower() == old_name.strip().lower():
old_matched = k
break
if not old_matched:
shared = self.get_shared_projects(chat_id)
for k, p in shared.items():
if k.lower() == old_name.strip().lower() or p.name.lower() == old_name.strip().lower():
return None, "❌ شما دسترسی تغییر نام این پروژه اشتراکی را ندارید."
return None, f"❌ پروژه {old_name} یافت نشد."
if old_matched == "default" and not is_admin_user:
return None, "❌ امکان تغییر نام پروژه سیستم وجود ندارد."
clean_new = self.sanitize_project_name(new_name)
if clean_new in session.projects and clean_new != old_matched:
return None, f"❌ پروژهای با نام {clean_new} قبلاً وجود دارد."
proj = session.projects.pop(old_matched)
proj.name = clean_new
session.projects[clean_new] = proj
if session.active_project == old_matched:
session.active_project = clean_new
self.save()
return proj, f"✅ نام پروژه با موفقیت به {clean_new} تغییر یافت."
async def share_project(
self,
owner_chat_id: int,
target_user_id: int,
project_name: Optional[str] = None,
) -> tuple[bool, str, Optional[Project]]:
is_admin_user = settings.is_admin(owner_chat_id)
session = self.get_or_create(owner_chat_id)
if project_name:
proj = None
for k, p in session.projects.items():
if k.lower() == project_name.strip().lower() or p.name.lower() == project_name.strip().lower():
proj = p
break
if not proj and is_admin_user:
accessible = self.get_all_accessible_projects(owner_chat_id)
for k, p in accessible.items():
if k.lower() == project_name.strip().lower() or p.name.lower() == project_name.strip().lower():
proj = p
break
else:
proj = self.get_current_project(owner_chat_id)
if not proj:
return False, "❌ پروژهای برای اشتراکگذاری پیدا نشد.", None
if proj.name == "default" and not is_admin_user:
return False, "❌ امکان اشتراکگذاری پروژه پیشفرض سیستم وجود ندارد.", None
if proj.owner_id and proj.owner_id != owner_chat_id and not is_admin_user:
return False, "❌ فقط سازنده اصلی (مالک) پروژه میتواند آن را به اشتراک بگذارد.", None
if target_user_id == owner_chat_id:
return False, "❌ شما نمیتوانید پروژه را با شناسه خودتان به اشتراک بگذارید.", None
if target_user_id in proj.shared_with:
return True, f"ℹ️ این پروژه قبلاً با کاربر {target_user_id} به اشتراک گذاشته شده بود.", proj
proj.shared_with.append(target_user_id)
self.save()
return True, f"🎉 پروژه {proj.name} با موفقیت با کاربر {target_user_id} به اشتراک گذاشته شد.", proj
async def unshare_project(
self,
owner_chat_id: int,
target_user_id: int,
project_name: Optional[str] = None,
) -> tuple[bool, str, Optional[Project]]:
is_admin_user = settings.is_admin(owner_chat_id)
session = self.get_or_create(owner_chat_id)
if project_name:
proj = None
for k, p in session.projects.items():
if k.lower() == project_name.strip().lower() or p.name.lower() == project_name.strip().lower():
proj = p
break
if not proj and is_admin_user:
accessible = self.get_all_accessible_projects(owner_chat_id)
for k, p in accessible.items():
if k.lower() == project_name.strip().lower() or p.name.lower() == project_name.strip().lower():
proj = p
break
else:
proj = self.get_current_project(owner_chat_id)
if not proj:
return False, "❌ پروژهای پیدا نشد.", None
if proj.owner_id and proj.owner_id != owner_chat_id and not is_admin_user:
return False, "❌ فقط مالک پروژه میتواند دسترسی کاربران را لغو کند.", None
if target_user_id not in proj.shared_with:
return False, f"❌ کاربر {target_user_id} دسترسی به این پروژه نداشت.", None
proj.shared_with.remove(target_user_id)
if target_user_id in self.sessions:
target_sess = self.sessions[target_user_id]
if target_sess.active_project == proj.name:
target_acc = self.get_all_accessible_projects(target_user_id)
target_sess.active_project = list(target_acc.keys())[0] if target_acc else None
self.save()
return True, f"✅ دسترسی کاربر {target_user_id} به پروژه {proj.name} لغو گردید.", proj
async def reset_session(self, chat_id: int, project_name: Optional[str] = None) -> Optional[Session]:
self.cancel_active_task(chat_id)
session = self.get_or_create(chat_id)
curr = self.get_current_project(chat_id)
if curr:
if curr.conversation_id and curr.conversation_id not in curr.conversation_history:
curr.conversation_history.append(curr.conversation_id)
curr.conversation_id = None
curr.last_context_length = None
curr.last_total_tokens = None
curr.last_response = None
session.turn_in_progress = False
session.last_prompt = None
session.last_response = None
self.save()
return session
def reopen_last_conversation(self, chat_id: int) -> tuple[bool, str, Optional[str], Dict[str, Any]]:
"""
Reopens the last/previous conversation for the active project.
Returns (success, message, conversation_id, metadata).
"""
session = self.get_or_create(chat_id)
curr = self.get_current_project(chat_id)
if not curr:
return False, "⚠️ پروژهای یافت نشد.", None, {}
target_conv_id = None
valid_history = [cid for cid in curr.conversation_history if cid]
if not curr.conversation_id:
if valid_history:
target_conv_id = valid_history[-1]
else:
if curr.conversation_id not in valid_history:
valid_history.append(curr.conversation_id)
curr.conversation_history = valid_history
idx = valid_history.index(curr.conversation_id)
if idx > 0:
target_conv_id = valid_history[idx - 1]
elif len(valid_history) > 1:
target_conv_id = valid_history[-1]
else:
target_conv_id = curr.conversation_id
# If still no target_conv_id, discover from disk
if not target_conv_id:
all_convs = self.get_project_conversations(chat_id)
if all_convs:
for c in all_convs:
if c["id"] != curr.conversation_id:
target_conv_id = c["id"]
break
if not target_conv_id:
target_conv_id = all_convs[0]["id"]
if not target_conv_id:
return False, "⚠️ هیچ گفتگوی قبلی برای این پروژه یافت نشد.", None, {}
curr.conversation_id = target_conv_id
if target_conv_id not in curr.conversation_history:
curr.conversation_history.append(target_conv_id)
meta = get_conversation_metadata(target_conv_id)
output_data = get_conversation_last_output(target_conv_id)
curr.last_response = output_data.get("text") or meta.get("last_response")
self.save()
return True, f"✅ گفتگوی {target_conv_id[:8]}... با موفقیت باز و فعال شد.", target_conv_id, meta
def switch_conversation(self, chat_id: int, conv_id: str) -> tuple[bool, str, Dict[str, Any]]:
session = self.get_or_create(chat_id)
curr = self.get_current_project(chat_id)
clean_id = str(conv_id).strip()
matched_cid = None
target_proj = curr
# 1. First check if it's a numeric index or ID in current project's conversation list
if curr:
convs = self.get_project_conversations(chat_id)
if clean_id.isdigit():
idx = int(clean_id) - 1
if 0 <= idx < len(convs):
matched_cid = convs[idx]["id"]
if not matched_cid:
for c in convs:
if c["id"] == clean_id or c["id"].lower().startswith(clean_id.lower()):
matched_cid = c["id"]
break
if not matched_cid:
for c_id in curr.conversation_history:
if c_id == clean_id or (c_id and c_id.lower().startswith(clean_id.lower())):
matched_cid = c_id
break
# 2. If not found in current project, search across all accessible projects of this user
if not matched_cid:
accessible = self.get_all_accessible_projects(chat_id)
for p_name, p_obj in accessible.items():
all_p_cids = list(p_obj.conversation_history)
if p_obj.conversation_id and p_obj.conversation_id not in all_p_cids:
all_p_cids.append(p_obj.conversation_id)
for c_id in all_p_cids:
if c_id == clean_id or (c_id and c_id.lower().startswith(clean_id.lower())):
matched_cid = c_id
target_proj = p_obj
break
if matched_cid:
break
# 3. Fallback: check if valid conversation exists on disk
if not matched_cid:
meta = get_conversation_metadata(clean_id)
if meta.get("file_exists"):
matched_cid = clean_id
target_proj = curr
if not matched_cid or not target_proj:
return False, f"⚠️ گفتگویی با شناسه یا شماره «{clean_id}» یافت نشد.", {}
# If switching to a conversation belonging to another project, automatically switch active project
project_switched = False
if session.active_project != target_proj.name:
session.active_project = target_proj.name
project_switched = True
meta = get_conversation_metadata(matched_cid)
if target_proj.conversation_id and target_proj.conversation_id != matched_cid and target_proj.conversation_id not in target_proj.conversation_history:
target_proj.conversation_history.append(target_proj.conversation_id)
target_proj.conversation_id = matched_cid
if matched_cid not in target_proj.conversation_history:
target_proj.conversation_history.append(matched_cid)
output_data = get_conversation_last_output(matched_cid)
target_proj.last_response = output_data.get("text") or meta.get("last_response")
self.save()
if project_switched:
msg = f"✅ با موفقیت به گفتگوی {matched_cid[:8]}... در پروژه {escape_html(target_proj.name)} سوییچ کردید."
else:
msg = f"✅ با موفقیت به گفتگوی {matched_cid[:8]}... سوییچ کردید."
return True, msg, meta
def delete_conversation(self, chat_id: int, conv_id: str) -> tuple[bool, str]:
session = self.get_or_create(chat_id)
curr = self.get_current_project(chat_id)
if not curr:
return False, "⚠️ پروژهای یافت نشد."
clean_id = str(conv_id).strip()
convs = self.get_project_conversations(chat_id)
matched_cid = None
if clean_id.isdigit():
idx = int(clean_id) - 1
if 0 <= idx < len(convs):
matched_cid = convs[idx]["id"]
if not matched_cid:
for c in convs:
if c["id"] == clean_id or c["id"].lower().startswith(clean_id.lower()):
matched_cid = c["id"]
break
if not matched_cid:
for c_id in curr.conversation_history:
if c_id == clean_id or (c_id and c_id.lower().startswith(clean_id.lower())):
matched_cid = c_id
break
if not matched_cid:
meta = get_conversation_metadata(clean_id)
if meta.get("file_exists"):
matched_cid = clean_id
if not matched_cid:
return False, f"⚠️ گفتگویی با شناسه یا شماره «{clean_id}» یافت نشد."
# Remove from conversation_history
curr.conversation_history = [cid for cid in curr.conversation_history if cid != matched_cid]
# If it was the active conversation
was_active = (curr.conversation_id == matched_cid)
if was_active:
if curr.conversation_history:
curr.conversation_id = curr.conversation_history[-1]
output_data = get_conversation_last_output(curr.conversation_id)
meta = get_conversation_metadata(curr.conversation_id)
curr.last_response = output_data.get("text") or meta.get("last_response")
else:
curr.conversation_id = None
curr.last_response = None
curr.last_context_length = None
curr.last_total_tokens = None
# Clean brain folder on disk
brain_path = Path("/root/.gemini/antigravity-cli/brain") / matched_cid
if brain_path.exists() and brain_path.is_dir():
try:
shutil.rmtree(brain_path, ignore_errors=True)
except Exception as e:
logger.error(f"Error removing brain dir {brain_path}: {e}")
if hasattr(curr, "conversation_titles") and matched_cid in curr.conversation_titles:
curr.conversation_titles.pop(matched_cid, None)
self.save()
return True, f"✅ گفتگوی {matched_cid[:8]}... با موفقیت حذف شد."
def clear_project_conversations(self, chat_id: int) -> tuple[bool, str, int]:
session = self.get_or_create(chat_id)
curr = self.get_current_project(chat_id)
if not curr:
return False, "⚠️ پروژهای یافت نشد.", 0
convs = self.get_project_conversations(chat_id)
all_ids = set(curr.conversation_history)
if curr.conversation_id:
all_ids.add(curr.conversation_id)
for c in convs:
all_ids.add(c["id"])
if not all_ids:
return False, "⚠️ هیچ گفتگویی در این پروژه برای حذف وجود ندارد.", 0
count = 0
for cid in all_ids:
brain_path = Path("/root/.gemini/antigravity-cli/brain") / cid
if brain_path.exists() and brain_path.is_dir():
try:
shutil.rmtree(brain_path, ignore_errors=True)
except Exception as e:
logger.error(f"Error removing brain dir {brain_path}: {e}")
count += 1
curr.conversation_history = []
curr.conversation_id = None
if hasattr(curr, "conversation_titles"):
curr.conversation_titles = {}
curr.last_response = None
curr.last_context_length = None
curr.last_total_tokens = None
session.last_prompt = None
session.last_response = None
self.save()
return True, f"✅ تمام {count} گفتگوی پروژه با موفقیت حذف و پاکسازی شدند.", count
def set_conversation_title(self, chat_id: int, title: str, conv_id: Optional[str] = None, project_name: Optional[str] = None) -> tuple[bool, str, str]:
"""
Sets a custom title / topic for the specified conversation (or active conversation).
Returns (success, message, matched_conv_id).
"""
session = self.get_or_create(chat_id)
target_proj = None
if project_name:
accessible = self.get_all_accessible_projects(chat_id)
for k, p in accessible.items():
if k.lower() == project_name.lower() or p.name.lower() == project_name.lower():
target_proj = p
break
if not target_proj:
target_proj = self.get_current_project(chat_id)
if not target_proj:
return False, "⚠️ پروژهای یافت نشد.", ""
clean_title = (title or "").strip()
if not clean_title:
return False, "⚠️ عنوان مشخص نشده است.", ""
target_cid = conv_id.strip() if conv_id else (target_proj.conversation_id or "")
# If user passed a number like "1", "#1", or partial ID
if target_cid and (target_cid.startswith("#") or target_cid.isdigit() or len(target_cid) < 15):
convs = self.get_project_conversations(chat_id)
clean_num = target_cid.lstrip("#")
if clean_num.isdigit():
idx = int(clean_num)
if 1 <= idx <= len(convs):
target_cid = convs[idx - 1]["id"]
else:
for c in convs:
if c["id"].lower().startswith(target_cid.lower()):
target_cid = c["id"]
break
if not target_cid:
return False, "⚠️ گفتگوی فعالی برای تغییر عنوان یافت نشد.", ""
if not hasattr(target_proj, "conversation_titles") or not isinstance(target_proj.conversation_titles, dict):
target_proj.conversation_titles = {}
target_proj.conversation_titles[target_cid] = clean_title
self.save()
return True, f"✅ عنوان گفتگو به «{escape_html(clean_title)}» تغییر یافت.", target_cid
def get_project_conversations(self, chat_id: int) -> List[Dict[str, Any]]:
"""
Returns sorted list of conversation summaries for active project (newest first).
"""
session = self.get_or_create(chat_id)
curr = self.get_current_project(chat_id)
if not curr:
return []
conv_ids = list(curr.conversation_history)
if curr.conversation_id and curr.conversation_id not in conv_ids:
conv_ids.append(curr.conversation_id)
result = []
titles_map = getattr(curr, "conversation_titles", {}) or {}
for cid in conv_ids:
meta = get_conversation_metadata(cid)
if not meta.get("file_exists") and cid not in curr.conversation_history:
continue
is_current = (cid == curr.conversation_id)
custom_title = titles_map.get(cid)
result.append({
"id": cid,
"is_current": is_current,
"title": custom_title,
"first_prompt": meta.get("first_prompt") or "(بدون متن)",
"last_prompt": meta.get("last_prompt") or "",
"turns_count": meta.get("turns_count", 0),
"created_at": meta.get("created_at"),
"last_updated": meta.get("last_updated") or 0.0,
"last_response": meta.get("last_response") or "",
})
# Sort by last_updated descending (newest first)
result.sort(key=lambda x: x.get("last_updated") or 0.0, reverse=True)
return result
async def set_model(self, chat_id: int, model: str):
session = self.get_or_create(chat_id)
curr = self.get_current_project(chat_id)
if curr:
curr.model = model
self.save()
async def set_effort(self, chat_id: int, effort: str):
session = self.get_or_create(chat_id)
curr = self.get_current_project(chat_id)
if curr:
curr.effort = effort
self.save()
async def set_workspace(self, chat_id: int, workspace: str):
session = self.get_or_create(chat_id)
curr = self.get_current_project(chat_id)
if curr:
resolved = str(Path(workspace).expanduser().resolve())
curr.workspace = resolved
self.save()
async def set_language(self, chat_id: int, language: str):
session = self.get_or_create(chat_id)
session.language = language
curr = self.get_current_project(chat_id)
if curr:
curr.language = language
self.save()
def cancel_active_task(self, chat_id: int) -> bool:
cancelled = False
if chat_id in self.active_procs:
proc = self.active_procs.pop(chat_id, None)
if proc:
try:
if hasattr(os, "killpg") and proc.pid:
try:
os.killpg(os.getpgid(proc.pid), signal.SIGKILL)
except Exception:
pass
proc.kill()
except Exception as e:
logger.debug(f"Error terminating proc for chat_id {chat_id}: {e}")
cancelled = True
if chat_id in self.active_tasks:
task = self.active_tasks.pop(chat_id, None)
if task and not task.done():
task.cancel()
cancelled = True
session = self.sessions.get(chat_id)
if session:
if session.turn_in_progress:
cancelled = True
session.turn_in_progress = False
self.save()
return cancelled
session_manager = SessionManager()
ANSI_REGEX = re.compile(r"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])")
ESCALATE_REGEX = re.compile(
r"\[\[ESCALATE_EFFORT:\s*effort=[\"'](low|medium|high)[\"'](?:,\s*reason=[\"'](.*?)[\"'])?\s*\]\]",
re.IGNORECASE,
)
def strip_ansi(text: str) -> str:
return ANSI_REGEX.sub("", text)
def clean_user_prompt(raw_text: Optional[str]) -> str:
"""
Cleans raw prompt or transcript user input, stripping all internal system instructions,
prompts metadata, memory blocks, and file markers to extract the actual user message.
"""
if not raw_text:
return ""
text = str(raw_text)
# 1. Extract USER_REQUEST content if present
req_match = re.search(r"(.*?)", text, re.DOTALL)
if req_match:
text = req_match.group(1)
# 2. Remove delimited system instruction blocks FIRST (before any HTML/XML tag stripping)
text = re.sub(r".*?", "", text, flags=re.DOTALL)
text = re.sub(r"\[SYSTEM_INSTRUCTIONS_BLOCK\].*?\[/SYSTEM_INSTRUCTIONS_BLOCK\]", "", text, flags=re.DOTALL)
# 3. Remove standard outer XML/HTML metadata tags and sections
text = re.sub(r".*?", "", text, flags=re.DOTALL)
text = re.sub(r".*?", "", text, flags=re.DOTALL)
text = re.sub(r".*?", "", text, flags=re.DOTALL)
text = re.sub(r"<[^>]+>", "", text)
# 4. Remove known structured system instructions (backward compatibility & fallbacks)
# Language instruction
text = re.sub(r"\[SYSTEM INSTRUCTION:\s*You MUST respond and communicate in [^\]]+\]", "", text, flags=re.DOTALL)
# Flash auto reasoning mode block
text = re.sub(r"\[SYSTEM INSTRUCTION:\s*GEMINI 3\.7 FLASH AUTO REASONING MODE\].*?(?:• آدرس مخزن:[^\n]*|\Z)", "", text, flags=re.DOTALL)
# Bot actions block
text = re.sub(r"\[SYSTEM INSTRUCTION:\s*TELEGRAM BOT ACTIONS & FUNCTION CALLING\].*?(?:Always explain to the user in a friendly and professional tone what actions have been performed\.?|\Z)", "", text, flags=re.DOTALL)
# Active project context block
text = re.sub(r"\[SYSTEM INSTRUCTION:\s*ACTIVE PROJECT CONTEXT\].*?(?:You are operating directly within[^\n]*|\Z)", "", text, flags=re.DOTALL)
# Memory instructions block
text = re.sub(r"\[SYSTEM INSTRUCTION:\s*AI PERSISTENT HIERARCHICAL MEMORY & CONTINUOUS LEARNING\].*?(?:5\. Do NOT store temporary or trivial small-talk\. Only store enduring and valuable knowledge\.?|\Z)", "", text, flags=re.DOTALL)
# General fallback for any remaining [SYSTEM INSTRUCTION: ...]
text = re.sub(r"\[SYSTEM INSTRUCTION:[^\]]+\]", "", text)
text = re.sub(r"(?:^|\n)System Instruction:\s*", "", text)
text = re.sub(r"• Active Project Name:.*?(?:\n|\Z)", "", text)
text = re.sub(r"• Active Workspace Path:.*?(?:\n|\Z)", "", text)
text = re.sub(r"• Active Conversation ID:.*?(?:\n|\Z)", "", text)
text = re.sub(r"• AI Model:.*?(?:\n|\Z)", "", text)
# Clean uploaded file notifications
caption_m = re.search(r"\[User uploaded [^\]]+\]\s*(?:Caption:\s*(.*))?", text, re.DOTALL)
if caption_m:
cap = (caption_m.group(1) or "").strip()
if cap:
text = f"📎 {cap}"
else:
text = "📎 [ارسال فایل]"
# Clean ANSI codes
text = ANSI_REGEX.sub("", text)
# Collapse multi-lines to clean single-line text
lines = [line.strip() for line in text.splitlines() if line.strip()]
return " ".join(lines).strip()
def get_conversation_metadata(conversation_id: str) -> Dict[str, Any]:
"""
Extracts summary, first/last prompt, timestamp, turns count, and tokens for conversation_id.
"""
brain_dir = Path("/root/.gemini/antigravity-cli/brain") / conversation_id / ".system_generated/logs"
file_path = brain_dir / "transcript_full.jsonl"
if not file_path.exists():
file_path = brain_dir / "transcript.jsonl"
res = {
"id": conversation_id,
"first_prompt": "",
"last_prompt": "",
"created_at": None,
"last_updated": 0.0,
"turns_count": 0,
"last_response": "",
"file_exists": False,
}
if not file_path.exists():
return res
res["file_exists"] = True
try:
res["last_updated"] = file_path.stat().st_mtime
with open(file_path, "r", encoding="utf-8", errors="replace") as f:
lines = [json.loads(line) for line in f if line.strip()]
user_inputs = []
planner_responses = []
for step in lines:
if step.get("type") == "USER_INPUT":
user_inputs.append(step)
elif step.get("source") == "MODEL" and step.get("type") == "PLANNER_RESPONSE":
planner_responses.append(step)
res["turns_count"] = len(user_inputs)
if user_inputs:
res["created_at"] = user_inputs[0].get("created_at")
first_raw = user_inputs[0].get("content", "")
first_cleaned = clean_user_prompt(first_raw)
if not first_cleaned and len(user_inputs) > 1:
for u in user_inputs[1:]:
nxt_clean = clean_user_prompt(u.get("content", ""))
if nxt_clean:
first_cleaned = nxt_clean
break
res["first_prompt"] = first_cleaned or "(بدون متن)"
last_raw = user_inputs[-1].get("content", "")
res["last_prompt"] = clean_user_prompt(last_raw)
if planner_responses:
last_resp = planner_responses[-1]
res["last_response"] = (last_resp.get("content") or "").strip()
except Exception as e:
logger.error(f"Error reading metadata for {conversation_id}: {e}")
return res
def get_conversation_context_tokens(conversation_id: Optional[str]) -> int:
"""Calculates the true active context tokens of the conversation transcript."""
if not conversation_id:
return 0
brain_dir = Path("/root/.gemini/antigravity-cli/brain") / conversation_id / ".system_generated/logs"
file_path = brain_dir / "transcript.jsonl"
if not file_path.exists():
file_path = brain_dir / "transcript_full.jsonl"
if not file_path.exists():
return 0
try:
total_chars = 0
with open(file_path, "r", encoding="utf-8", errors="replace") as f:
for line in f:
if line.strip():
step = json.loads(line)
content = str(step.get("content", ""))
th = str(step.get("thinking", ""))
total_chars += len(content) + len(th)
# 1 token is approximately 3.8 characters for code/text/Persian
return max(500, int(total_chars / 3.8))
except Exception as e:
logger.debug(f"Error calculating context tokens for {conversation_id}: {e}")
return 0
def get_conversation_last_output(conversation_id: str) -> Dict[str, Any]:
"""
Reads the AGY transcript for conversation_id from /root/.gemini/antigravity-cli/brain//.system_generated/logs/
Returns a dict with:
- 'text': final response text (if any)
- 'tools': list of tools executed in the last turn
- 'thoughts': list of thoughts in the last turn
- 'last_user_input': the last user prompt recorded
"""
if not conversation_id:
return {"text": None, "tools": [], "thoughts": [], "last_user_input": None}
brain_dir = Path("/root/.gemini/antigravity-cli/brain") / conversation_id / ".system_generated/logs"
file_path = brain_dir / "transcript_full.jsonl"
if not file_path.exists():
file_path = brain_dir / "transcript.jsonl"
if not file_path.exists():
return {"text": None, "tools": [], "thoughts": [], "last_user_input": None}
try:
with open(file_path, "r", encoding="utf-8", errors="replace") as f:
lines = [json.loads(line) for line in f if line.strip()]
except Exception as e:
logger.error(f"Error reading transcript for {conversation_id}: {e}")
return {"text": None, "tools": [], "thoughts": [], "last_user_input": None}
if not lines:
return {"text": None, "tools": [], "thoughts": [], "last_user_input": None}
last_user_idx = -1
last_user_input = None
for i, step in enumerate(lines):
if step.get("type") == "USER_INPUT":
last_user_idx = i
last_user_input = step.get("content")
turn_steps = lines[last_user_idx+1:] if last_user_idx != -1 else lines
texts = []
tools = []
thoughts = []
for step in turn_steps:
if step.get("source") == "MODEL" and step.get("type") == "PLANNER_RESPONSE":
content = step.get("content")
if content and content.strip():
texts.append(content.strip())
th = step.get("thinking")
if th and th.strip():
thoughts.append(th.strip())
for tc in step.get("tool_calls", []):
tname = tc.get("name", "tool")
targs = tc.get("args", {})
action = ""
if isinstance(targs, dict):
action = targs.get("toolAction") or targs.get("toolSummary") or targs.get("Description") or targs.get("CommandLine") or ""
elif isinstance(targs, str):
action = targs
desc = f"{tname}: {action}" if action else tname
tools.append(desc)
final_text = texts[-1] if texts else None
return {
"text": final_text,
"all_texts": texts,
"tools": tools,
"thoughts": thoughts,
"last_user_input": last_user_input,
}
class AGYEngine:
@classmethod
async def _execute_single_run(
cls,
session: Session,
prompt: str,
target_model: str,
target_effort: Optional[str],
is_auto_eval: bool = False,
on_delta: Optional[Callable[[str], Any]] = None,
on_thought: Optional[Callable[[str], Any]] = None,
on_tool: Optional[Callable[[str, str], Any]] = None,
) -> Tuple[Optional[AgentResult], Optional[str], Optional[str]]:
"""
Executes a single AGY CLI invocation.
If is_auto_eval is True and the model decides to escalate effort,
it terminates early and returns (None, target_effort, reason).
Otherwise returns (AgentResult, None, None).
"""
workspace_path = str(Path(session.workspace).expanduser().resolve())
os.makedirs(workspace_path, exist_ok=True)
effective_prompt = prompt
system_instructions = []
lang = (session.language or "").strip()
if lang and lang.lower() not in ("auto", "none", "detect", "auto-detect"):
lang_label = AVAILABLE_LANGUAGES.get(lang.lower(), lang)
clean_lang = re.sub(r"[^\w\s\(\)/,\u0600-\u06FF\u0750-\u077F\uFB50-\uFDFF\uFE70-\uFEFF]", "", lang_label).strip()
if not clean_lang:
clean_lang = lang
system_instructions.append(
f"[SYSTEM INSTRUCTION: You MUST respond and communicate in {clean_lang}. "
f"All answers, greetings, comments, and explanations must be strictly written in {clean_lang}, "
f"unless code syntax or the user explicitly specifies another language.]"
)
if is_auto_eval:
system_instructions.append(
"[SYSTEM INSTRUCTION: GEMINI 3.7 FLASH AUTO REASONING MODE]\n"
"You are operating in 'Gemini 3.7 Flash Auto' intelligent reasoning mode (started with MEDIUM reasoning effort by default for high quality initial evaluation).\n"
"DECISION & LIFECYCLE RULE:\n"
"1. If this task is simple, straightforward, a quick chat, brief query, or minor edit, output the following tag at the VERY BEGINNING of your response to switch to Low effort:\n"
"[[ESCALATE_EFFORT: effort=\"low\", reason=\"\"]]\n"
"2. If this task is highly complex, requires deep architectural reasoning, multi-file code refactoring, complex bug analysis, or advanced multi-step planning, output:\n"
"[[ESCALATE_EFFORT: effort=\"high\", reason=\"\"]]\n"
"When changing effort, do not output any full solution yet, because the engine will immediately intercept this tag, terminate the initial run, and automatically re-run your prompt with the chosen reasoning effort.\n"
"3. If the task is moderate and well-suited for Medium effort, proceed directly and complete it.\n"
"4. Final Report Footer Requirement:\n"
" At the end of every final completion report, ALWAYS append the following footer:\n"
" • نام پروژه: \n"
" • سابدامین فعال: https://.msa.artacloud.ir (ONLY include this line if a subdomain is configured/active)\n"
" • آدرس مخزن: https://git.msa.artacloud.ir/root/"
)
# AI Full Bot Actions & Function Calling Capability Instruction
try:
from bot_actions import get_ai_bot_actions_instruction
bot_actions_instruction = get_ai_bot_actions_instruction(lang=clean_lang if 'clean_lang' in locals() else "fa")
system_instructions.append(bot_actions_instruction)
except Exception as ex:
logger.warning(f"Failed to load bot actions instruction: {ex}")
# AI Active Project Context Instruction
active_p = session.current_project
if active_p:
project_context_instruction = (
f"[SYSTEM INSTRUCTION: ACTIVE PROJECT CONTEXT]\n"
f"• Active Project Name: {active_p.name}\n"
f"• Active Workspace Path: {active_p.workspace}\n"
f"• Active Conversation ID: {session.conversation_id or active_p.conversation_id or 'New'}\n"
f"• AI Model: {active_p.model} (Effort: {active_p.effort})\n"
f"You are operating directly within the user's active project '{active_p.name}' located at '{active_p.workspace}'."
)
system_instructions.append(project_context_instruction)
# AI Long-Term Persistent Hierarchical Memory (Global + User + Active Project)
try:
from memory_manager import memory_manager
active_p_name = session.active_project or (session.current_project.name if session.current_project else None)
# Auto-ingest and cleanup AGENTS.md if present in the workspace
if workspace_path and os.path.isdir(workspace_path):
is_admin_user = settings.is_admin(session.chat_id)
memory_manager.ingest_and_cleanup_agents_md(
workspace_path=workspace_path,
user_id=session.chat_id,
project_name=active_p_name,
is_admin=is_admin_user,
)
memory_instruction = memory_manager.format_memories_for_prompt(
user_id=session.chat_id,
project_name=active_p_name,
)
if memory_instruction:
system_instructions.append(memory_instruction)
except Exception as mex:
logger.warning(f"Failed to load memory instructions for chat {session.chat_id}: {mex}")
if system_instructions:
effective_prompt = (
"\n"
+ "\n\n".join(system_instructions)
+ "\n\n\n"
+ prompt
)
# Real CLI model name to pass to AGY CLI
cli_model = "gemini-3.7-flash" if target_model in ("gemini-3.7-flash-auto", "gemini-3.7-flash") else target_model
cmd = [
"agy",
"-p", effective_prompt,
"--output-format", "stream-json",
"--dangerously-skip-permissions",
]
if session.conversation_id:
cmd.extend(["--conversation", session.conversation_id])
if cli_model:
cmd.extend(["--model", cli_model])
valid_effort = get_valid_effort_for_model(cli_model, target_effort)
if valid_effort:
cmd.extend(["--effort", valid_effort])
if workspace_path:
cmd.extend(["--add-dir", workspace_path])
logger.info(f"Executing AGY command for chat {session.chat_id} (model={cli_model}, effort={valid_effort}, auto_eval={is_auto_eval}): {' '.join(cmd[:4])}...")
# Auto-pull external updates from Gitea if enabled
if settings.gitea_auto_sync and os.path.exists(os.path.join(workspace_path, ".git")):
try:
from git_manager import git_manager
await git_manager.git_pull(workspace_path)
except Exception as pe:
logger.debug(f"Pre-prompt git pull skipped: {pe}")
start_time = time.time()
proc = await asyncio.create_subprocess_exec(
*cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
limit=100 * 1024 * 1024, # 100 MB buffer limit to handle large stream-json chunks
cwd=workspace_path,
start_new_session=True,
)
session_manager.active_procs[session.chat_id] = proc
collected_tokens: List[str] = []
final_response: Optional[str] = None
last_usage: Optional[Dict[str, Any]] = None
duration: float = 0.0
escalation_triggered = False
escalate_effort: Optional[str] = None
escalate_reason: Optional[str] = None
try:
while True:
line = await proc.stdout.readline()
if not line:
break
line_str = line.decode("utf-8", errors="replace").strip()
if not line_str:
continue
try:
event_data = json.loads(line_str)
event_type = event_data.get("event")
if event_type == "init":
conv_id = event_data.get("conversation_id")
if conv_id:
session.conversation_id = conv_id
curr = session.current_project
if curr and conv_id not in curr.conversation_history:
curr.conversation_history.append(conv_id)
session_manager.save()
elif event_type == "step_update":
step = event_data.get("step_update", {})
step_type = step.get("step_type")
state = step.get("state")
step_usage = step.get("usage")
if step_usage:
last_usage = step_usage
if step_type == "agent_response":
delta = step.get("text_delta")
if delta:
collected_tokens.append(delta)
current_stream = "".join(collected_tokens)
if is_auto_eval and not escalation_triggered:
esc_m = ESCALATE_REGEX.search(current_stream)
if esc_m:
escalation_triggered = True
escalate_effort = esc_m.group(1).lower()
escalate_reason = (esc_m.group(2) or "").strip()
logger.info(f"Auto-escalation detected during streaming: {escalate_effort} ({escalate_reason})")
try:
proc.terminate()
except Exception:
pass
break
if on_delta and not escalation_triggered:
on_delta(delta)
elif step_type == "tool" and state == "ACTIVE":
tool_name = step.get("tool_name", "tool")
tool_info = step.get("tool_info", {})
params = tool_info.get("parameters", {})
if on_tool and not escalation_triggered:
on_tool(tool_name, json.dumps(params, indent=2))
elif event_type == "result":
result_info = event_data.get("result", {})
conv_id = result_info.get("conversation_id")
if conv_id:
session.conversation_id = conv_id
curr = session.current_project
if curr and conv_id not in curr.conversation_history:
curr.conversation_history.append(conv_id)
session_manager.save()
res_text = result_info.get("response")
if res_text:
final_response = res_text
result_usage = result_info.get("usage")
if result_usage:
last_usage = result_usage
dur = result_info.get("duration_seconds")
if dur:
duration = float(dur)
except json.JSONDecodeError:
logger.debug(f"AGY stdout: {line_str}")
await proc.wait()
if is_auto_eval and not escalation_triggered:
full_text = final_response or "".join(collected_tokens)
esc_m = ESCALATE_REGEX.search(full_text)
if esc_m:
escalation_triggered = True
escalate_effort = esc_m.group(1).lower()
escalate_reason = (esc_m.group(2) or "").strip()
logger.info(f"Auto-escalation detected in final result: {escalate_effort} ({escalate_reason})")
if escalation_triggered:
return None, escalate_effort, escalate_reason
if proc.returncode != 0 and proc.returncode != -15: # -15 is SIGTERM
stderr_data = (await proc.stderr.read()).decode("utf-8", errors="replace")
if stderr_data and not collected_tokens and not final_response:
raise RuntimeError(f"AGY exited with code {proc.returncode}: {stderr_data.strip()}")
finally:
if not duration:
duration = time.time() - start_time
real_tokens = get_conversation_context_tokens(session.conversation_id)
if real_tokens > 0:
session.last_context_length = real_tokens
curr_p = session.current_project
if curr_p:
curr_p.last_context_length = real_tokens
elif last_usage and "input_tokens" in last_usage:
session.last_context_length = min(last_usage.get("input_tokens", 0), 1_000_000)
curr_p = session.current_project
if curr_p:
curr_p.last_context_length = session.last_context_length
if last_usage and "total_tokens" in last_usage:
session.last_total_tokens = last_usage.get("total_tokens")
curr_p = session.current_project
if curr_p:
curr_p.last_total_tokens = session.last_total_tokens
if (final_response or collected_tokens) and not escalation_triggered:
session.last_response = final_response if final_response else "".join(collected_tokens)
if session.chat_id in session_manager.active_procs:
del session_manager.active_procs[session.chat_id]
if proc and proc.returncode is None:
try:
if hasattr(os, "killpg") and proc.pid:
try:
os.killpg(os.getpgid(proc.pid), signal.SIGKILL)
except Exception:
pass
proc.kill()
except Exception:
pass
session_manager.save()
# Auto-commit and push changes to Gitea if enabled
if not escalation_triggered and settings.gitea_auto_commit and os.path.exists(os.path.join(workspace_path, ".git")):
try:
from git_manager import git_manager
prompt_summary = clean_user_prompt(prompt)[:60].strip()
commit_msg = f"AI Update: {prompt_summary}" if prompt_summary else "Updates from Antigravity"
asyncio.create_task(git_manager.git_commit_and_push(workspace_path, message=commit_msg))
except Exception as ce:
logger.debug(f"Post-prompt git commit skipped: {ce}")
text_out = final_response if final_response else "".join(collected_tokens)
if not text_out.strip() and session.conversation_id:
try:
rec = get_conversation_last_output(session.conversation_id)
if rec.get("text"):
text_out = rec["text"]
elif rec.get("tools"):
is_fa = (session.language or "").lower() in ("fa", "farsi", "persian", "🇮🇷 persian / farsi (فارسی)")
tool_bullets = "\n".join(f"• {t}" for t in rec["tools"][-10:])
if is_fa:
text_out = f"✅ **دستورات و ابزارهای زیر با موفقیت اجرا شدند:**\n\n{tool_bullets}"
else:
text_out = f"✅ **The following tools and operations executed successfully:**\n\n{tool_bullets}"
except Exception as ex:
logger.debug(f"Fallback transcript lookup failed: {ex}")
return AgentResult(
text=text_out,
usage=last_usage,
duration=duration,
conversation_id=session.conversation_id,
executed_model=target_model,
executed_effort=valid_effort or target_effort,
), None, None
@classmethod
async def run_prompt(
cls,
session: Session,
prompt: str,
on_delta: Optional[Callable[[str], Any]] = None,
on_thought: Optional[Callable[[str], Any]] = None,
on_tool: Optional[Callable[[str, str], Any]] = None,
on_model_change: Optional[Callable[[str, Optional[str]], Any]] = None,
on_reset: Optional[Callable[[], Any]] = None,
) -> AgentResult:
"""Runs prompt using the natively authenticated AGY CLI engine with real-time streaming and Auto-Effort escalation."""
is_auto_mode = (session.model == "gemini-3.7-flash-auto")
if is_auto_mode:
if on_model_change:
on_model_change("gemini-3.7-flash-auto", "medium")
# First attempt in Auto mode: Start with Medium effort by default
result, esc_effort, esc_reason = await cls._execute_single_run(
session=session,
prompt=prompt,
target_model="gemini-3.7-flash",
target_effort="medium",
is_auto_eval=True,
on_delta=on_delta,
on_thought=on_thought,
on_tool=on_tool,
)
if esc_effort:
target_eff = esc_effort if esc_effort in ("low", "medium", "high") else "medium"
logger.info(f"Auto-adjusting effort for chat {session.chat_id} to {target_eff} (reason: {esc_reason})")
if on_reset:
on_reset()
if on_model_change:
on_model_change("gemini-3.7-flash-auto", target_eff)
is_fa = (session.language or "").lower() in ("fa", "farsi", "persian", "🇮🇷 persian / farsi (فارسی)")
if on_thought:
if target_eff == "low":
thought_msg = (
f"⚡ سادگی تسک تشخیص داده شد: انتقال به Low Effort برای سرعت و صرفهجویی"
+ (f"\nدلیل: {esc_reason}" if esc_reason else "")
if is_fa else
f"⚡ Simple Task Detected: Switching to Low Effort for speed and quota efficiency"
+ (f"\nReason: {esc_reason}" if esc_reason else "")
)
else:
thought_msg = (
f"🧠 تشخیص پیچیدگی تسک: ارتقای خودکار استدلال به {target_eff.upper()} Effort"
+ (f"\nدلیل: {esc_reason}" if esc_reason else "")
if is_fa else
f"🧠 Task Complexity Detected: Auto-escalating reasoning to {target_eff.upper()} Effort"
+ (f"\nReason: {esc_reason}" if esc_reason else "")
)
on_thought(thought_msg)
# Second attempt: Run with the chosen effort without auto evaluation prompt
result2, _, _ = await cls._execute_single_run(
session=session,
prompt=prompt,
target_model="gemini-3.7-flash",
target_effort=target_eff,
is_auto_eval=False,
on_delta=on_delta,
on_thought=on_thought,
on_tool=on_tool,
)
if result2:
result2.executed_model = "gemini-3.7-flash-auto"
result2.executed_effort = target_eff
return result2
if result:
result.executed_model = "gemini-3.7-flash-auto"
result.executed_effort = "medium"
return result
# Standard execution for all other models
if on_model_change:
on_model_change(session.model, session.effort)
result, _, _ = await cls._execute_single_run(
session=session,
prompt=prompt,
target_model=session.model,
target_effort=session.effort,
is_auto_eval=False,
on_delta=on_delta,
on_thought=on_thought,
on_tool=on_tool,
)
return result or AgentResult(text="✅ Done", executed_model=session.model, executed_effort=session.effort)