1661 lines
72 KiB
Python
1661 lines
72 KiB
Python
import os
|
|
import sys
|
|
import json
|
|
import time
|
|
import re
|
|
import shutil
|
|
import asyncio
|
|
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__)
|
|
|
|
# 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 = ""
|
|
|
|
@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",
|
|
)
|
|
}
|
|
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"])
|
|
|
|
# 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,
|
|
}
|
|
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}")
|
|
|
|
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)
|
|
|
|
if clean_target in accessible:
|
|
self.cancel_active_task(chat_id)
|
|
session.active_project = clean_target
|
|
self.save()
|
|
return accessible[clean_target]
|
|
|
|
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()
|
|
return 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"❌ پروژه <code>{proj_name}</code> یافت نشد."
|
|
|
|
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"⚙️ سرویس سیستمی <code>{matched_key}.service</code> متوقف و حذف شد.")
|
|
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 (<code>{matched_key}.msa.artacloud.ir</code>) آزاد و حذف شد.")
|
|
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"🗑️ پوشه فایلها (<code>{proj_obj.workspace}</code>) پاکسازی شد.")
|
|
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"✅ <b>پروژه <code>{matched_key}</code> با موفقیت به صورت کامل و هوشمند حذف گردید:</b>\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"❌ پروژه <code>{old_name}</code> یافت نشد."
|
|
|
|
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"❌ پروژهای با نام <code>{clean_new}</code> قبلاً وجود دارد."
|
|
|
|
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"✅ نام پروژه با موفقیت به <code>{clean_new}</code> تغییر یافت."
|
|
|
|
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"ℹ️ این پروژه قبلاً با کاربر <code>{target_user_id}</code> به اشتراک گذاشته شده بود.", proj
|
|
|
|
proj.shared_with.append(target_user_id)
|
|
self.save()
|
|
return True, f"🎉 پروژه <b>{proj.name}</b> با موفقیت با کاربر <code>{target_user_id}</code> به اشتراک گذاشته شد.", 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"❌ کاربر <code>{target_user_id}</code> دسترسی به این پروژه نداشت.", 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"✅ دسترسی کاربر <code>{target_user_id}</code> به پروژه <b>{proj.name}</b> لغو گردید.", 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"✅ گفتگوی <code>{target_conv_id[:8]}...</code> با موفقیت باز و فعال شد.", 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)
|
|
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}» یافت نشد.", {}
|
|
|
|
meta = get_conversation_metadata(matched_cid)
|
|
|
|
if curr.conversation_id and curr.conversation_id != matched_cid and curr.conversation_id not in curr.conversation_history:
|
|
curr.conversation_history.append(curr.conversation_id)
|
|
|
|
curr.conversation_id = matched_cid
|
|
if matched_cid not in curr.conversation_history:
|
|
curr.conversation_history.append(matched_cid)
|
|
|
|
output_data = get_conversation_last_output(matched_cid)
|
|
curr.last_response = output_data.get("text") or meta.get("last_response")
|
|
self.save()
|
|
return True, f"✅ با موفقیت به گفتگوی <code>{matched_cid[:8]}...</code> سوییچ کردید.", 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}")
|
|
|
|
self.save()
|
|
return True, f"✅ گفتگوی <code>{matched_cid[:8]}...</code> با موفقیت حذف شد."
|
|
|
|
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
|
|
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 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)
|
|
|
|
# For admin or default project, also discover existing brain directories if list is small
|
|
is_admin_user = settings.is_admin(chat_id)
|
|
if is_admin_user or curr.name == "default" or len(conv_ids) < 3:
|
|
try:
|
|
brain_p = Path("/root/.gemini/antigravity-cli/brain")
|
|
if brain_p.exists():
|
|
for d in brain_p.iterdir():
|
|
if d.is_dir() and len(d.name) > 30 and (d / ".system_generated/logs/transcript.jsonl").exists():
|
|
if d.name not in conv_ids:
|
|
conv_ids.append(d.name)
|
|
except Exception as e:
|
|
logger.debug(f"Error scanning brain dir: {e}")
|
|
|
|
result = []
|
|
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)
|
|
result.append({
|
|
"id": cid,
|
|
"is_current": is_current,
|
|
"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:
|
|
if not raw_text:
|
|
return ""
|
|
text = raw_text
|
|
req_match = re.search(r"<USER_REQUEST>(.*?)</USER_REQUEST>", text, re.DOTALL)
|
|
if req_match:
|
|
text = req_match.group(1)
|
|
text = re.sub(r"\[SYSTEM INSTRUCTION:.*?\]", "", text, flags=re.DOTALL)
|
|
text = re.sub(r"<ADDITIONAL_METADATA>.*?</ADDITIONAL_METADATA>", "", text, flags=re.DOTALL)
|
|
text = re.sub(r"<USER_SETTINGS_CHANGE>.*?</USER_SETTINGS_CHANGE>", "", text, flags=re.DOTALL)
|
|
text = re.sub(r"<[^>]+>", "", text)
|
|
return text.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", "")
|
|
res["first_prompt"] = clean_user_prompt(first_raw)
|
|
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/<conversation_id>/.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=\"<brief reason in 1 sentence>\"]]\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=\"<brief reason in 1 sentence>\"]]\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"
|
|
" • نام پروژه: <project_name>\n"
|
|
" • سابدامین فعال: https://<subdomain>.msa.artacloud.ir (ONLY include this line if a subdomain is configured/active)\n"
|
|
" • آدرس مخزن: https://git.msa.artacloud.ir/root/<project_name>"
|
|
)
|
|
|
|
# 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 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)
|
|
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".join(system_instructions) + "\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"• <code>{t}</code>" 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,
|
|
) -> 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:
|
|
# 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})")
|
|
|
|
is_fa = (session.language or "").lower() in ("fa", "farsi", "persian", "🇮🇷 persian / farsi (فارسی)")
|
|
if on_thought:
|
|
if target_eff == "low":
|
|
thought_msg = (
|
|
f"⚡ <b>سادگی تسک تشخیص داده شد:</b> انتقال به <code>Low Effort</code> برای سرعت و صرفهجویی"
|
|
+ (f"\n<i>دلیل: {esc_reason}</i>" if esc_reason else "")
|
|
if is_fa else
|
|
f"⚡ <b>Simple Task Detected:</b> Switching to <code>Low Effort</code> for speed and quota efficiency"
|
|
+ (f"\n<i>Reason: {esc_reason}</i>" if esc_reason else "")
|
|
)
|
|
else:
|
|
thought_msg = (
|
|
f"🧠 <b>تشخیص پیچیدگی تسک:</b> ارتقای خودکار استدلال به <code>{target_eff.upper()} Effort</code>"
|
|
+ (f"\n<i>دلیل: {esc_reason}</i>" if esc_reason else "")
|
|
if is_fa else
|
|
f"🧠 <b>Task Complexity Detected:</b> Auto-escalating reasoning to <code>{target_eff.upper()} Effort</code>"
|
|
+ (f"\n<i>Reason: {esc_reason}</i>" 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
|
|
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)
|