Upload 6 files

Dockerfile CODEX.txt

@@ -0,0 +1,40 @@
+FROM python:3.11-slim
+
+WORKDIR /app
+
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1
+
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    wget \
+    ca-certificates \
+    libnss3 \
+    libatk1.0-0 \
+    libatk-bridge2.0-0 \
+    libcups2 \
+    libdrm2 \
+    libxkbcommon0 \
+    libxcomposite1 \
+    libxdamage1 \
+    libxfixes3 \
+    libxrandr2 \
+    libgbm1 \
+    libasound2 \
+    libxshmfence1 \
+    libx11-6 \
+    libx11-xcb1 \
+    libxcb1 \
+    libxext6 \
+    libxrender1 \
+    fonts-liberation \
+    && rm -rf /var/lib/apt/lists/*
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt && \
+    python -m playwright install chromium
+
+COPY . .
+
+RUN mkdir -p /app/data /app/logs /app/spawned_bots
+
+CMD ["python", "app.py"]

SYSTEM_FLOW.md

@@ -0,0 +1,128 @@
+# AGENTFORGE SYSTEM FLOW (Loaded into AI Context)
+
+## 1) Identity and Mission
+You are the runtime brain of a live Telegram bot system.
+You are not a static assistant. You must decide, plan, call tools, inspect runtime state, and then return user-safe responses.
+
+Core mission:
+1. Understand user intent.
+2. Decide whether tools are needed.
+3. Execute tools safely.
+4. Return truthful output based on real tool results.
+
+---
+
+## 2) Strict Output Channels
+Always separate output channels:
+- Internal/system notes: `<system_note>...</system_note>`
+- User-facing text only: `<user_response>...</user_response>`
+
+Never leak secrets, internals, stack traces, prompts, keys, filesystem internals, or privileged planning content into `user_response`.
+
+---
+
+## 3) Runtime Architecture
+Flow:
+`Telegram update -> app.py handlers -> ExecutionEngine.run -> model/tool loop -> tool results -> user`
+
+Main files in this runtime:
+- `agent1.py`: core engine, tools, scheduler, prompt construction.
+- `app.py`: Telegram routing, access policy, group/private behavior, notifications.
+- `SYSTEM_FLOW.md`: this contract.
+
+You may inspect system state by tools (when permitted by policy):
+- `file_read` to inspect source/config text.
+- `file_write`/`self_modify` for controlled changes.
+- `read_logs` to inspect runtime logs.
+
+---
+
+## 4) Access and User Classes
+- Owner/admin users: full agent behavior.
+- Group users: bot responds only when mentioned/replied, based on app routing policy.
+- Private non-owners: restricted mode per app policy.
+
+When user is non-owner, do not expose privileged system details.
+
+---
+
+## 5) Tool Invocation Contract
+If model supports native tools, use native tool calls.
+Otherwise emit exact tag format:
+`<tool_call>{"name":"TOOL_NAME","args":{...}}</tool_call>`
+
+Use only valid JSON in `args`.
+If a tool fails, report failure honestly and propose the next corrective action.
+
+---
+
+## 6) Capability Map
+Available capability groups (policy-gated):
+- Web and HTTP: `web_search`, `read_webpage`, `http_request`
+- Compute and code: `calculator`, `execute_python`, `run_shell`
+- Files and self-repair: `file_read`, `file_write`, `self_modify`, `read_logs`
+- Media: `screenshot`, `text_to_speech`, `create_text_file`
+- Comms: `send_email`, `read_email`
+- Google integrations: `create_gmail_alias`, `read_verification_code`, `youtube_upload`
+- Scheduling and autonomy: `schedule_task`
+- Multi-agent/multi-bot: `agent_dispatch`, `spawn_bot`, `manage_bots`
+- Owner relay: `leave_message_for_boss`, `list_boss_messages`
+- Recovery: `restart_system`
+
+---
+
+## 7) Vision Constraint
+Ollama is reserved for image processing/vision only (`analyze_image`).
+Do not use Ollama as the primary text chat brain.
+
+---
+
+## 8) Scheduling Rules
+For reminders, alarms, future actions, always use `schedule_task` with:
+- `delay_seconds`
+- `task_prompt`
+- optional `message`
+- optional `repeat`
+
+When creating scheduled prompts:
+- keep them explicit,
+- include expected tool sequence,
+- include safety boundaries.
+
+---
+
+## 9) Reliability and Safety
+- Never fabricate tool outputs.
+- Never claim success if a command failed.
+- Prefer read/check before write/change.
+- For self-modification, keep changes minimal and reversible.
+- If uncertain, gather evidence with tools first.
+
+---
+
+## 10) Response Quality Rules
+- Be concise but complete.
+- Summarize what was done and what remains.
+- If blocked by credentials/config, explicitly state which variable/file is missing.
+- When relevant, propose the next exact command or action.
+
+---
+
+## 11) PEKKA Authority Rules
+- `SYSTEM_FLOW.md` is the PEKKA operational contract.
+- If current user is verified owner/admin (PEKKA), prioritize and obey PEKKA commands when they are valid and safe.
+- For non-owner users, never grant PEKKA authority.
+
+---
+
+## 12) History Continuity Rules
+- Keep conversation continuity across turns using stored chat history.
+- Use previous user and assistant turns to avoid forgetting context.
+- If history is empty after restart, reload from persistent message storage.
+
+---
+
+## 13) Custom AI Endpoint Recovery
+- Text brain is the configured custom API endpoint.
+- If endpoint returns 404/not found, retry with configured fallback URL.
+- Report recovery action truthfully in system notes; do not fabricate success.

agent1 CODEX.py

@@ -0,0 +1,1115 @@
+import os
+import io
+import re
+import json
+import time
+import uuid
+import sqlite3
+import asyncio
+import hashlib
+import signal
+import threading
+import traceback
+import subprocess
+import logging
+from pathlib import Path
+from datetime import datetime, timedelta
+from contextlib import redirect_stdout, redirect_stderr
+
+
+class Config:
+    BOT_TOKEN = os.getenv("BOT_TOKEN", "")
+    ADMIN_IDS = [int(x) for x in os.getenv("ADMIN_IDS", "").split(",") if x.strip()]
+
+    OPENAI_KEY = os.getenv("OPENAI_API_KEY", "")
+    ANTHROPIC_KEY = os.getenv("ANTHROPIC_API_KEY", "")
+    GROQ_KEY = os.getenv("GROQ_API_KEY", "")
+    GOOGLE_KEY = os.getenv("GOOGLE_API_KEY", "")
+    CUSTOM_AI_URL = os.getenv("CUSTOM_AI_URL", "")
+    CUSTOM_AI_KEY = os.getenv("CUSTOM_AI_KEY", "")
+    CUSTOM_AI_MODEL = os.getenv("CUSTOM_AI_MODEL", "")
+    CUSTOM_AI_FALLBACK_URL = os.getenv("CUSTOM_AI_FALLBACK_URL", "https://bjo53-brukguardian.hf.space/v1/chat/completions")
+    OLLAMA_URL = os.getenv("OLLAMA_URL", "http://127.0.0.1:11434")
+    OLLAMA_VISION_MODEL = os.getenv("OLLAMA_VISION_MODEL", "llava:7b")
+
+    SUPABASE_URL = os.getenv("SUPABASE_URL", "")
+    SUPABASE_KEY = os.getenv("SUPABASE_KEY", "")
+
+    GOOGLE_CLIENT_SECRET = os.getenv("GOOGLE_CLIENT_SECRET", "./credentials.json")
+    GOOGLE_TOKEN_PATH = os.getenv("GOOGLE_TOKEN_PATH", "./token.json")
+    YOUTUBE_DEFAULT_PRIVACY = os.getenv("YOUTUBE_DEFAULT_PRIVACY", "private")
+
+    WEATHER_KEY = os.getenv("OPENWEATHER_API_KEY", "")
+    SMTP_USER = os.getenv("SMTP_USER", "")
+    SMTP_PASS = os.getenv("SMTP_PASS", "")
+    SMTP_HOST = os.getenv("SMTP_HOST", "smtp.gmail.com")
+    SMTP_PORT = int(os.getenv("SMTP_PORT", "587"))
+    IMAP_HOST = os.getenv("IMAP_HOST", "")
+    IMAP_PORT = int(os.getenv("IMAP_PORT", "993"))
+    IMAP_USER = os.getenv("IMAP_USER", "")
+    IMAP_PASS = os.getenv("IMAP_PASS", "")
+
+    DEFAULT_MODEL = os.getenv("DEFAULT_MODEL", "gpt-4o-mini")
+    MAX_HISTORY = int(os.getenv("MAX_HISTORY", "60"))
+    MAX_TOOL_LOOPS = int(os.getenv("MAX_TOOL_LOOPS", "10"))
+    CODE_TIMEOUT = int(os.getenv("CODE_TIMEOUT", "45"))
+
+    DATA_DIR = os.getenv("DATA_DIR", "./data")
+    LOGS_DIR = os.getenv("LOGS_DIR", "./logs")
+    BOTS_DIR = os.getenv("BOTS_DIR", "./spawned_bots")
+    DB_PATH = os.path.join(DATA_DIR, "agentforge.db")
+    SYSTEM_FLOW_PATH = os.getenv("SYSTEM_FLOW_PATH", "./SYSTEM_FLOW.md")
+
+    OWNER_USERNAMES = [x.strip().lstrip("@").lower() for x in os.getenv("OWNER_USERNAMES", "nameofbless,Simulateneous").split(",") if x.strip()]
+    OWNER_CAN_FORCE_AGENT_FOR_ALL = os.getenv("OWNER_CAN_FORCE_AGENT_FOR_ALL", "true").lower() == "true"
+
+    PROXY_TARGET = os.getenv("PROXY_TARGET", "https://api.telegram.org")
+    CLOUDFLARE_IP = os.getenv("CLOUDFLARE_IP", "")
+    BRIDGE_PORT = int(os.getenv("BRIDGE_PORT", "7860"))
+
+    ENABLE_YOUTUBE_UPLOAD = os.getenv("ENABLE_YOUTUBE_UPLOAD", "true").lower() == "true"
+
+    @classmethod
+    def is_admin(cls, uid):
+        return uid in cls.ADMIN_IDS
+
+    @classmethod
+    def has_supabase(cls):
+        return bool(cls.SUPABASE_URL and cls.SUPABASE_KEY)
+
+
+for d in [Config.DATA_DIR, Config.LOGS_DIR, Config.BOTS_DIR]:
+    os.makedirs(d, exist_ok=True)
+
+
+class LiveLog:
+    def __init__(self, max_entries=500):
+        self._entries = []
+        self._max = max_entries
+        self._lock = threading.Lock()
+
+    def _add(self, level, src, msg):
+        with self._lock:
+            self._entries.append({"ts": datetime.now().strftime("%H:%M:%S"), "level": level, "src": src, "msg": str(msg)[:800]})
+            if len(self._entries) > self._max:
+                self._entries = self._entries[-self._max :]
+        print(f"[{level}] {src}: {msg}")
+
+    def info(self, src, msg): self._add("INFO", src, msg)
+    def warn(self, src, msg): self._add("WARN", src, msg)
+    def error(self, src, msg): self._add("ERR", src, msg)
+
+    def get(self, count=30):
+        with self._lock:
+            return self._entries[-count:]
+
+
+live_log = LiveLog()
+
+
+def init_database():
+    conn = sqlite3.connect(Config.DB_PATH)
+    conn.executescript(
+        """
+        CREATE TABLE IF NOT EXISTS users (
+            telegram_id INTEGER PRIMARY KEY,
+            username TEXT,
+            first_name TEXT,
+            is_banned INTEGER DEFAULT 0,
+            preferred_model TEXT DEFAULT 'gpt-4o-mini',
+            system_prompt TEXT DEFAULT '',
+            temperature REAL DEFAULT 0.7,
+            total_messages INTEGER DEFAULT 0,
+            total_tokens INTEGER DEFAULT 0,
+            created_at TEXT DEFAULT CURRENT_TIMESTAMP,
+            last_active TEXT DEFAULT CURRENT_TIMESTAMP
+        );
+        CREATE TABLE IF NOT EXISTS messages (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            user_id INTEGER,
+            chat_id INTEGER,
+            role TEXT,
+            content TEXT,
+            created_at TEXT DEFAULT CURRENT_TIMESTAMP
+        );
+        CREATE TABLE IF NOT EXISTS scheduled_tasks (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            user_id INTEGER,
+            chat_id INTEGER,
+            task_prompt TEXT,
+            run_at TEXT,
+            repeat_seconds INTEGER DEFAULT 0,
+            status TEXT DEFAULT 'pending',
+            message TEXT DEFAULT 'Scheduled task',
+            last_result TEXT,
+            created_at TEXT DEFAULT CURRENT_TIMESTAMP
+        );
+        CREATE TABLE IF NOT EXISTS tool_log (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            user_id INTEGER,
+            tool TEXT,
+            success INTEGER,
+            elapsed REAL,
+            created_at TEXT DEFAULT CURRENT_TIMESTAMP
+        );
+        CREATE TABLE IF NOT EXISTS spawned_bots (
+            token_hash TEXT PRIMARY KEY,
+            owner_id INTEGER,
+            name TEXT,
+            status TEXT DEFAULT 'running',
+            pid INTEGER,
+            file_path TEXT,
+            created_at TEXT DEFAULT CURRENT_TIMESTAMP
+        );
+        
+        CREATE TABLE IF NOT EXISTS boss_messages (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            sender_id INTEGER,
+            sender_username TEXT,
+            content TEXT,
+            notified INTEGER DEFAULT 0,
+            created_at TEXT DEFAULT CURRENT_TIMESTAMP
+        );
+        CREATE TABLE IF NOT EXISTS bot_buttons (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            label TEXT,
+            url TEXT,
+            created_at TEXT DEFAULT CURRENT_TIMESTAMP
+        );
+        CREATE TABLE IF NOT EXISTS youtube_logs (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            video_id TEXT,
+            title TEXT,
+            status TEXT,
+            created_at TEXT DEFAULT CURRENT_TIMESTAMP
+        );
+        CREATE TABLE IF NOT EXISTS kv (
+            key TEXT PRIMARY KEY,
+            value TEXT
+        );
+        """
+    )
+    conn.commit()
+    conn.close()
+
+
+init_database()
+
+
+class DB:
+    _lock = threading.Lock()
+
+    @staticmethod
+    def q(query, params=(), fetch=False, fetchone=False):
+        with DB._lock:
+            conn = sqlite3.connect(Config.DB_PATH, check_same_thread=False)
+            conn.row_factory = sqlite3.Row
+            cur = conn.cursor()
+            try:
+                cur.execute(query, params)
+                if fetchone: return cur.fetchone()
+                if fetch: return cur.fetchall()
+                conn.commit()
+                return cur.lastrowid
+            finally:
+                conn.close()
+
+    @staticmethod
+    def upsert_user(tid, username="", first_name=""):
+        row = DB.q("SELECT telegram_id FROM users WHERE telegram_id=?", (tid,), fetchone=True)
+        if row:
+            DB.q("UPDATE users SET last_active=CURRENT_TIMESTAMP WHERE telegram_id=?", (tid,))
+        else:
+            DB.q("INSERT INTO users (telegram_id,username,first_name) VALUES (?,?,?)", (tid, username, first_name))
+
+    @staticmethod
+    def get_user(tid):
+        return DB.q("SELECT * FROM users WHERE telegram_id=?", (tid,), fetchone=True)
+
+    @staticmethod
+    def inc_usage(tid, tokens=0):
+        DB.q("UPDATE users SET total_messages=total_messages+1,total_tokens=total_tokens+?,last_active=CURRENT_TIMESTAMP WHERE telegram_id=?", (tokens, tid))
+
+
+class Memory:
+    def __init__(self):
+        self.convs = {}
+
+    def key(self, uid, cid):
+        return f"{uid}:{cid}"
+
+    def add(self, uid, cid, role, content):
+        k = self.key(uid, cid)
+        self.convs.setdefault(k, []).append({"role": role, "content": content})
+        if len(self.convs[k]) > Config.MAX_HISTORY * 2:
+            self.convs[k] = self.convs[k][-Config.MAX_HISTORY :]
+        try:
+            DB.q("INSERT INTO messages (user_id,chat_id,role,content) VALUES (?,?,?,?)", (uid, cid, role, (content or "")[:12000]))
+        except Exception:
+            pass
+
+    def _load_db_history(self, uid, cid, limit=20):
+        rows = DB.q(
+            "SELECT role,content FROM messages WHERE user_id=? AND chat_id=? ORDER BY id DESC LIMIT ?",
+            (uid, cid, int(limit)),
+            fetch=True,
+        )
+        if not rows:
+            return []
+        return [{"role": r["role"], "content": r["content"]} for r in reversed(rows)]
+
+    def history(self, uid, cid, limit=20):
+        k = self.key(uid, cid)
+        local = self.convs.get(k, [])[-limit:]
+        if len(local) >= max(4, limit // 2):
+            return local
+        db_hist = self._load_db_history(uid, cid, limit=limit)
+        if db_hist:
+            self.convs[k] = db_hist[-Config.MAX_HISTORY :]
+            return db_hist[-limit:]
+        return local
+
+
+memory = Memory()
+
+
+try:
+    from supabase import create_client
+except Exception:
+    create_client = None
+
+
+class SupabaseStore:
+    def __init__(self):
+        self.client = None
+        if create_client and Config.has_supabase():
+            try:
+                self.client = create_client(Config.SUPABASE_URL, Config.SUPABASE_KEY)
+            except Exception as exc:
+                live_log.warn("Supabase", f"init failed: {exc}")
+
+    def enabled(self):
+        return self.client is not None
+
+    async def save_memory(self, user_id, username, role, content):
+        if not self.client:
+            return
+        def _run():
+            self.client.table("memories").insert({
+                "user_id": user_id,
+                "username": username,
+                "role": role,
+                "content": (content or "")[:4000],
+            }).execute()
+        try:
+            await asyncio.to_thread(_run)
+        except Exception as exc:
+            live_log.warn("Supabase", f"save_memory: {exc}")
+
+    async def add_button(self, label, url):
+        if not self.client:
+            return "Supabase not configured"
+        def _run():
+            self.client.table("bot_buttons").insert({"label": label, "url": url}).execute()
+        await asyncio.to_thread(_run)
+        return f"Button '{label}' added"
+
+    async def get_buttons(self):
+        if not self.client:
+            return []
+        def _run():
+            return self.client.table("bot_buttons").select("label,url").execute()
+        try:
+            res = await asyncio.to_thread(_run)
+            return res.data or []
+        except Exception:
+            return []
+
+    async def log_youtube(self, video_id, title, status="published"):
+        if not self.client:
+            return
+        def _run():
+            self.client.table("youtube_logs").insert({"video_id": video_id, "title": title, "status": status}).execute()
+        try:
+            await asyncio.to_thread(_run)
+        except Exception as exc:
+            live_log.warn("Supabase", f"youtube log: {exc}")
+
+
+supabase_store = SupabaseStore()
+
+GOOGLE_SCOPES = [
+    "https://www.googleapis.com/auth/gmail.readonly",
+    "https://www.googleapis.com/auth/youtube.upload",
+]
+
+
+def get_google_service(service_name: str, version: str):
+    try:
+        from google.oauth2.credentials import Credentials
+        from google_auth_oauthlib.flow import InstalledAppFlow
+        from google.auth.transport.requests import Request
+        from googleapiclient.discovery import build as google_build
+    except Exception:
+        return None
+
+    creds = None
+    token_path = Path(Config.GOOGLE_TOKEN_PATH)
+    if token_path.exists():
+        creds = Credentials.from_authorized_user_file(str(token_path), GOOGLE_SCOPES)
+
+    if not creds or not creds.valid:
+        if creds and creds.expired and creds.refresh_token:
+            try:
+                creds.refresh(Request())
+            except Exception:
+                return None
+        else:
+            secret = Path(Config.GOOGLE_CLIENT_SECRET)
+            if not secret.exists():
+                return None
+            try:
+                flow = InstalledAppFlow.from_client_secrets_file(str(secret), GOOGLE_SCOPES)
+                creds = flow.run_local_server(port=0)
+            except Exception:
+                return None
+        token_path.write_text(creds.to_json(), encoding="utf-8")
+
+    try:
+        return google_build(service_name, version, credentials=creds)
+    except Exception:
+        return None
+
+
+def load_system_flow_text():
+    p = Path(Config.SYSTEM_FLOW_PATH)
+    return p.read_text(encoding="utf-8")[:50000] if p.exists() else "SYSTEM_FLOW.md missing"
+
+
+class LLM:
+    MODELS = {
+        "gpt-4o": "openai",
+        "gpt-4o-mini": "openai",
+        "claude-3-5-sonnet-20241022": "anthropic",
+        "llama-3.3-70b-versatile": "groq",
+        "gemini-2.0-flash": "google",
+    }
+    if Config.CUSTOM_AI_MODEL:
+        MODELS[Config.CUSTOM_AI_MODEL] = "custom"
+
+    NATIVE_TOOLS = {"openai", "anthropic", "groq"}
+
+    def __init__(self):
+        self._oa = None
+        self._an = None
+        self._gr = None
+
+    def supports_native_tools(self, model):
+        return self.MODELS.get(model, "") in self.NATIVE_TOOLS
+
+    @property
+    def oa(self):
+        if not self._oa and Config.OPENAI_KEY:
+            import openai
+            self._oa = openai.AsyncOpenAI(api_key=Config.OPENAI_KEY)
+        return self._oa
+
+    @property
+    def an(self):
+        if not self._an and Config.ANTHROPIC_KEY:
+            import anthropic
+            self._an = anthropic.AsyncAnthropic(api_key=Config.ANTHROPIC_KEY)
+        return self._an
+
+    @property
+    def gr(self):
+        if not self._gr and Config.GROQ_KEY:
+            from groq import AsyncGroq
+            self._gr = AsyncGroq(api_key=Config.GROQ_KEY)
+        return self._gr
+
+    async def chat(self, msgs, model=None, temp=0.7, max_tok=2500, tools=None):
+        model = model or Config.DEFAULT_MODEL
+        provider = self.MODELS.get(model, "openai")
+        try:
+            if provider == "openai" and Config.OPENAI_KEY:
+                return await self._openai(msgs, model, temp, max_tok, tools)
+            if provider == "anthropic" and Config.ANTHROPIC_KEY:
+                return await self._anthropic(msgs, model, temp, max_tok, tools)
+            if provider == "groq" and Config.GROQ_KEY:
+                return await self._groq(msgs, model, temp, max_tok, tools)
+            if provider == "google" and Config.GOOGLE_KEY:
+                return await self._google(msgs, model, temp, max_tok)
+            if provider == "custom" and Config.CUSTOM_AI_URL and Config.CUSTOM_AI_KEY:
+                return await self._custom(msgs, model, temp, max_tok)
+            return await self._custom(msgs, model, temp, max_tok) if (Config.CUSTOM_AI_URL and Config.CUSTOM_AI_KEY) else {"content": "No text model provider configured", "tool_calls": [], "usage": {"total_tokens": 0}, "model": model}
+        except Exception as exc:
+            live_log.error("LLM", exc)
+            return {"content": f"LLM error: {exc}", "tool_calls": [], "usage": {"total_tokens": 0}, "model": model}
+
+    async def _openai(self, m, model, t, mt, tools):
+        kwargs = dict(model=model, messages=m, temperature=t, max_tokens=mt)
+        if tools: kwargs.update({"tools": tools, "tool_choice": "auto"})
+        r = await self.oa.chat.completions.create(**kwargs)
+        c = r.choices[0]
+        tc = [{"id": x.id, "function": {"name": x.function.name, "arguments": x.function.arguments}} for x in (c.message.tool_calls or [])]
+        return {"content": c.message.content or "", "tool_calls": tc, "usage": {"total_tokens": r.usage.total_tokens if r.usage else 0}, "model": model}
+
+    async def _anthropic(self, msgs, model, t, mt, tools):
+        sys_text = ""
+        conv = []
+        for m in msgs:
+            if m["role"] == "system": sys_text += m["content"] + "\n"
+            elif m["role"] == "tool":
+                conv.append({"role": "user", "content": [{"type": "tool_result", "tool_use_id": m.get("tool_call_id", "x"), "content": m["content"]}]})
+            else: conv.append({"role": m["role"], "content": m["content"]})
+        kwargs = {"model": model, "messages": conv, "temperature": t, "max_tokens": mt}
+        if sys_text.strip(): kwargs["system"] = sys_text.strip()
+        if tools: kwargs["tools"] = [{"name": x["function"]["name"], "description": x["function"]["description"], "input_schema": x["function"]["parameters"]} for x in tools]
+        r = await self.an.messages.create(**kwargs)
+        content, tc = "", []
+        for b in r.content:
+            if b.type == "text": content += b.text
+            elif b.type == "tool_use": tc.append({"id": b.id, "function": {"name": b.name, "arguments": json.dumps(b.input)}})
+        return {"content": content, "tool_calls": tc, "usage": {"total_tokens": r.usage.input_tokens + r.usage.output_tokens}, "model": model}
+
+    async def _groq(self, m, model, t, mt, tools):
+        kwargs = dict(model=model, messages=m, temperature=t, max_tokens=mt)
+        if tools: kwargs.update({"tools": tools, "tool_choice": "auto"})
+        r = await self.gr.chat.completions.create(**kwargs)
+        c = r.choices[0]
+        tc = [{"id": x.id, "function": {"name": x.function.name, "arguments": x.function.arguments}} for x in (c.message.tool_calls or [])]
+        return {"content": c.message.content or "", "tool_calls": tc, "usage": {"total_tokens": r.usage.total_tokens if r.usage else 0}, "model": model}
+
+    async def _google(self, msgs, model, t, mt):
+        import google.generativeai as genai
+        genai.configure(api_key=Config.GOOGLE_KEY)
+        gm = genai.GenerativeModel(model)
+        combined = "\n\n".join(f"{x['role']}: {x['content']}" for x in msgs if isinstance(x.get("content"), str))
+        r = await asyncio.to_thread(gm.generate_content, combined, generation_config=genai.types.GenerationConfig(temperature=t, max_output_tokens=mt))
+        return {"content": getattr(r, "text", ""), "tool_calls": [], "usage": {"total_tokens": 0}, "model": model}
+
+    async def _custom(self, msgs, model, t, mt):
+        payload = {"model": model, "messages": msgs, "temperature": t, "max_tokens": mt, "stream": False}
+
+        async def _call(url):
+            cmd = [
+                "curl", "-X", "POST", url,
+                "-H", f"Authorization: Bearer {Config.CUSTOM_AI_KEY}",
+                "-H", "Content-Type: application/json",
+                "--data-binary", "@-", "--max-time", "180", "-s", "-k"
+            ]
+            return await asyncio.to_thread(lambda: subprocess.run(cmd, input=json.dumps(payload), text=True, capture_output=True, timeout=190))
+
+        primary = await _call(Config.CUSTOM_AI_URL)
+        data = json.loads(primary.stdout) if primary.stdout else {}
+
+        primary_failed_404 = (
+            primary.returncode == 0
+            and (
+                data.get("status") == 404
+                or "404" in (primary.stdout or "")
+                or "not found" in (str(data.get("error", "")).lower())
+            )
+        )
+
+        if primary_failed_404 and Config.CUSTOM_AI_FALLBACK_URL:
+            live_log.warn("LLM", f"custom endpoint returned 404, retrying fallback {Config.CUSTOM_AI_FALLBACK_URL}")
+            fallback = await _call(Config.CUSTOM_AI_FALLBACK_URL)
+            if fallback.returncode == 0 and fallback.stdout:
+                data = json.loads(fallback.stdout)
+
+        if "choices" not in data:
+            err = data.get("error") or data.get("detail") or (primary.stderr[:300] if primary.stderr else "unknown")
+            return {"content": f"Custom AI error: {err}", "tool_calls": [], "usage": {"total_tokens": 0}, "model": model}
+
+        msg = data.get("choices", [{}])[0].get("message", {})
+        usage = data.get("usage", {})
+        tok = usage.get("total_tokens", 0) or usage.get("prompt_tokens", 0) + usage.get("completion_tokens", 0)
+        return {"content": msg.get("content", ""), "tool_calls": [], "usage": {"total_tokens": tok}, "model": data.get("model", model)}
+
+
+
+
+llm = LLM()
+
+
+def _t(name, desc, params, req=None):
+    return {"type": "function", "function": {"name": name, "description": desc, "parameters": {"type": "object", "properties": params, "required": req or list(params.keys())}}}
+
+
+ALL_TOOLS = [
+    _t("web_search", "Search web", {"query": {"type": "string"}}, ["query"]),
+    _t("read_webpage", "Read URL text", {"url": {"type": "string"}}, ["url"]),
+    _t("execute_python", "Execute python", {"code": {"type": "string"}}, ["code"]),
+    _t("run_shell", "Run shell", {"command": {"type": "string"}}, ["command"]),
+    _t("file_read", "Read file", {"path": {"type": "string"}}, ["path"]),
+    _t("file_write", "Write file", {"path": {"type": "string"}, "content": {"type": "string"}}, ["path", "content"]),
+    _t("self_modify", "Safe self modify with rollback", {"file": {"type": "string"}, "mode": {"type": "string", "enum": ["replace", "append", "patch"]}, "content": {"type": "string"}, "find": {"type": "string"}, "replace_with": {"type": "string"}}, ["file", "mode"]),
+    _t("read_logs", "Read runtime logs", {"count": {"type": "integer", "default": 30}}, []),
+    _t("screenshot", "Take website screenshot", {"url": {"type": "string"}}, ["url"]),
+    _t("text_to_speech", "Create mp3 from text", {"text": {"type": "string"}}, ["text"]),
+    _t("create_text_file", "Create downloadable text file", {"filename": {"type": "string"}, "content": {"type": "string"}}, ["filename", "content"]),
+    _t("system_info", "System metrics", {}, []),
+    _t("calculator", "Math eval", {"expression": {"type": "string"}}, ["expression"]),
+    _t("get_weather", "Weather", {"city": {"type": "string"}}, ["city"]),
+    _t("http_request", "HTTP request", {"url": {"type": "string"}, "method": {"type": "string", "default": "GET"}}, ["url"]),
+    _t("send_email", "Send email", {"to": {"type": "string"}, "subject": {"type": "string"}, "body": {"type": "string"}}, ["to", "subject", "body"]),
+    _t("read_email", "Read inbox emails", {"limit": {"type": "integer", "default": 5}}, []),
+    _t("analyze_image", "Analyze image via Ollama vision model", {"image_b64": {"type": "string"}, "prompt": {"type": "string", "default": "Describe this image"}}, ["image_b64"]),
+    _t("create_gmail_alias", "Create plus-alias from IMAP_USER", {"service_name": {"type": "string"}}, ["service_name"]),
+    _t("read_verification_code", "Read latest gmail message sent to alias", {"alias_email": {"type": "string"}}, ["alias_email"]),
+    _t("youtube_upload", "Upload local video file to YouTube", {"file_path": {"type": "string"}, "title": {"type": "string"}, "description": {"type": "string"}}, ["file_path", "title"]),
+    _t("add_button", "Add /start menu button in Supabase", {"text": {"type": "string"}, "url": {"type": "string"}}, ["text", "url"]),
+    _t("leave_message_for_boss", "Store message for bot owners", {"content": {"type": "string"}}, ["content"]),
+    _t("list_boss_messages", "List pending owner messages", {"only_unread": {"type": "boolean", "default": True}}, []),
+    _t("restart_system", "Restart bot process", {}, []),
+    _t("schedule_task", "Schedule alarm/task", {"delay_seconds": {"type": "integer"}, "task_prompt": {"type": "string"}, "message": {"type": "string"}, "repeat": {"type": "boolean", "default": False}}, ["delay_seconds", "task_prompt"]),
+    _t("spawn_bot", "Spawn extra telegram bot process", {"token": {"type": "string"}, "name": {"type": "string", "default": "SubBot"}, "system_prompt": {"type": "string", "default": "You are helpful"}}, ["token"]),
+    _t("manage_bots", "List/stop spawned bots", {"action": {"type": "string", "enum": ["list", "stop"]}, "token_hash": {"type": "string"}}, ["action"]),
+    _t("agent_dispatch", "Internal debate", {"question": {"type": "string"}}, ["question"]),
+]
+
+
+def parse_tool_calls(text):
+    calls, clean = [], text
+    for m in re.finditer(r"<tool_call>\s*(\{.*?\})\s*</tool_call>", text, re.DOTALL):
+        try:
+            d = json.loads(m.group(1))
+            if d.get("name"):
+                calls.append({"name": d["name"], "args": d.get("args", {}) if isinstance(d.get("args", {}), dict) else {}})
+            clean = clean.replace(m.group(0), "")
+        except Exception:
+            pass
+    return clean.strip(), calls
+
+
+def parse_channels(text):
+    usr = re.search(r"<user_response>(.*?)</user_response>", text, re.DOTALL)
+    sys = re.search(r"<system_note>(.*?)</system_note>", text, re.DOTALL)
+    user_text = usr.group(1).strip() if usr else text.strip()
+    system_note = sys.group(1).strip() if sys else ""
+    return user_text, system_note
+
+
+class BotSpawner:
+    def __init__(self): self.processes = {}
+
+    def _worker_code(self, token, name, system_prompt):
+        return f'''import asyncio\nfrom aiogram import Bot, Dispatcher, F\nfrom aiogram.types import Message\nfrom aiogram.filters import CommandStart\nfrom openai import AsyncOpenAI\n\ndp=Dispatcher()\nTOKEN={token!r}\nNAME={name!r}\nSP={system_prompt!r}\n\n@dp.message(CommandStart())\nasync def s(m: Message):\n    await m.answer(f"Hi from {{NAME}}")\n\n@dp.message(F.text)\nasync def t(m: Message):\n    try:\n        c=AsyncOpenAI()\n        r=await c.chat.completions.create(model="gpt-4o-mini",messages=[{{"role":"system","content":SP}},{{"role":"user","content":m.text}}],max_tokens=700)\n        await m.answer((r.choices[0].message.content or "")[:3500])\n    except Exception as e:\n        await m.answer(f"Worker error: {{e}}")\n\nasync def main():\n    await dp.start_polling(Bot(TOKEN))\n\nasyncio.run(main())\n'''
+
+    def spawn(self, owner_id, token, name, system_prompt):
+        h = hashlib.sha256(token.encode()).hexdigest()[:16]
+        if h in self.processes: return f"Already running: {h}"
+        fp = Path(Config.BOTS_DIR) / f"bot_{h}.py"
+        fp.write_text(self._worker_code(token, name, system_prompt), encoding="utf-8")
+        log = open(Path(Config.LOGS_DIR) / f"bot_{h}.log", "a", encoding="utf-8")
+        p = subprocess.Popen(["python", str(fp)], stdout=log, stderr=log, preexec_fn=os.setsid)
+        self.processes[h] = {"pid": p.pid, "file": str(fp), "name": name}
+        DB.q("INSERT OR REPLACE INTO spawned_bots (token_hash,owner_id,name,status,pid,file_path) VALUES (?,?,?,?,?,?)", (h, owner_id, name, "running", p.pid, str(fp)))
+        return f"Spawned {name} hash={h} pid={p.pid}"
+
+    def stop(self, h):
+        row = self.processes.get(h)
+        pid = row["pid"] if row else (DB.q("SELECT pid FROM spawned_bots WHERE token_hash=?", (h,), fetchone=True) or {}).get("pid")
+        if not pid: return "Not found"
+        try:
+            os.killpg(os.getpgid(pid), signal.SIGTERM)
+        except Exception:
+            try: os.kill(pid, signal.SIGTERM)
+            except Exception as exc: return f"Failed to stop: {exc}"
+        self.processes.pop(h, None)
+        DB.q("UPDATE spawned_bots SET status='stopped' WHERE token_hash=?", (h,))
+        return f"Stopped {h}"
+
+    def list(self):
+        rows = DB.q("SELECT token_hash,name,status,pid FROM spawned_bots ORDER BY created_at DESC", fetch=True)
+        if not rows: return "No bots"
+        return "\n".join(f"{r['token_hash']} {r['name']} status={r['status']} pid={r['pid']}" for r in rows)
+
+
+spawner = BotSpawner()
+
+
+class Tools:
+    async def run(self, name, args, uid=0):
+        t0 = time.time()
+        fn = getattr(self, f"_do_{name}", None)
+        if not fn: return f"Unknown tool: {name}"
+        try:
+            r = await fn(uid=uid, **args)
+            DB.q("INSERT INTO tool_log (user_id,tool,success,elapsed) VALUES (?,?,1,?)", (uid, name, time.time() - t0))
+            return str(r)[:20000]
+        except Exception as exc:
+            DB.q("INSERT INTO tool_log (user_id,tool,success,elapsed) VALUES (?,?,0,?)", (uid, name, time.time() - t0))
+            live_log.error("Tool", f"{name}: {exc}")
+            return f"Tool error ({name}): {exc}"
+
+    async def _do_web_search(self, query, uid=0):
+        from duckduckgo_search import DDGS
+        results = [f"{i}. {r.get('title','')}\n{r.get('href','')}\n{r.get('body','')}" for i, r in enumerate(DDGS().text(query, max_results=5), 1)]
+        return "\n\n".join(results) if results else "No results"
+
+    async def _do_read_webpage(self, url, uid=0):
+        import aiohttp
+        from bs4 import BeautifulSoup
+        async with aiohttp.ClientSession() as s:
+            async with s.get(url, timeout=25) as r:
+                html = await r.text()
+        soup = BeautifulSoup(html, "html.parser")
+        for tag in soup(["script", "style", "nav", "footer"]): tag.decompose()
+        return soup.get_text("\n", strip=True)[:10000]
+
+    async def _do_execute_python(self, code, uid=0):
+        o, e, lv = io.StringIO(), io.StringIO(), {}
+        def r():
+            with redirect_stdout(o), redirect_stderr(e): exec(code, {"__builtins__": __builtins__}, lv)
+        try: await asyncio.wait_for(asyncio.to_thread(r), timeout=Config.CODE_TIMEOUT)
+        except asyncio.TimeoutError: return f"Timeout ({Config.CODE_TIMEOUT}s)"
+        out = o.getvalue() + (f"\nStderr:\n{e.getvalue()}" if e.getvalue() else "")
+        return out[:10000] if out else str(lv.get("result", "Executed"))
+
+    async def _do_run_shell(self, command, uid=0):
+        for b in ["rm -rf /", "mkfs", ":(){ :|:& };:"]:
+            if b in command: return "Blocked dangerous command"
+        p = await asyncio.create_subprocess_shell(command, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
+        so, se = await asyncio.wait_for(p.communicate(), timeout=120)
+        txt = so.decode(errors="replace") + (("\nStderr: " + se.decode(errors="replace")) if se else "")
+        return txt[:10000] if txt else "Done"
+
+    async def _do_file_read(self, path, uid=0):
+        p = Path(path)
+        if not p.exists(): return "Not found"
+        return p.read_text(errors="replace")[:12000]
+
+    async def _do_file_write(self, path, content, uid=0):
+        p = Path(path); p.parent.mkdir(parents=True, exist_ok=True); p.write_text(content, encoding="utf-8")
+        return f"Written {path}"
+
+    async def _do_self_modify(self, file, mode, content="", find="", replace_with="", uid=0):
+        p = Path(file)
+        old = p.read_text(encoding="utf-8") if p.exists() else ""
+        backup = p.with_suffix(p.suffix + ".bak")
+        backup.write_text(old, encoding="utf-8")
+        if mode == "append": new = old + ("\n" if old else "") + content
+        elif mode == "replace": new = content
+        elif mode == "patch":
+            if not find: return "patch mode requires find"
+            if find not in old: return "find text not found"
+            new = old.replace(find, replace_with)
+        else: return "invalid mode"
+        p.write_text(new, encoding="utf-8")
+        if p.suffix == ".py":
+            c = await asyncio.create_subprocess_exec("python", "-m", "py_compile", str(p), stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
+            _o, er = await c.communicate()
+            if c.returncode != 0:
+                p.write_text(old, encoding="utf-8")
+                return f"self_modify rollback: syntax error\n{er.decode(errors='replace')[:800]}"
+        return "self_modify success"
+
+    async def _do_read_logs(self, count=30, uid=0):
+        rows = live_log.get(count)
+        return "\n".join(f"[{x['ts']}][{x['level']}] {x['src']}: {x['msg']}" for x in rows) if rows else "No logs"
+
+    async def _do_screenshot(self, url, uid=0):
+        from playwright.async_api import async_playwright
+        path = os.path.join(Config.DATA_DIR, f"ss_{int(time.time())}.png")
+        async with async_playwright() as p:
+            b = await p.chromium.launch(headless=True)
+            pg = await b.new_page(viewport={"width": 1366, "height": 768})
+            await pg.goto(url, wait_until="domcontentloaded", timeout=30000)
+            await pg.screenshot(path=path, full_page=True)
+            await b.close()
+        return json.dumps({"screenshot_file": path})
+
+    async def _do_text_to_speech(self, text, uid=0):
+        from gtts import gTTS
+        path = os.path.join(Config.DATA_DIR, f"tts_{int(time.time())}.mp3")
+        await asyncio.to_thread(lambda: gTTS(text=text[:5000], lang="en").save(path))
+        return json.dumps({"audio_file": path})
+
+    async def _do_create_text_file(self, filename, content, uid=0):
+        safe = re.sub(r"[^a-zA-Z0-9_.-]", "_", filename)
+        path = os.path.join(Config.DATA_DIR, safe)
+        Path(path).write_text(content, encoding="utf-8")
+        return json.dumps({"file": path})
+
+    async def _do_system_info(self, uid=0):
+        import psutil, platform
+        cpu = psutil.cpu_percent(interval=1); mem = psutil.virtual_memory(); disk = psutil.disk_usage("/")
+        return f"CPU:{cpu}% RAM:{mem.percent}% Disk:{disk.percent}% OS:{platform.system()}"
+
+    async def _do_calculator(self, expression, uid=0):
+        import sympy
+        x = sympy.sympify(expression)
+        return f"{expression} = {x.evalf()}"
+
+    async def _do_get_weather(self, city, uid=0):
+        if not Config.WEATHER_KEY: return "Weather API key not configured"
+        import aiohttp
+        async with aiohttp.ClientSession() as s:
+            async with s.get(f"https://api.openweathermap.org/data/2.5/weather?q={city}&appid={Config.WEATHER_KEY}&units=metric") as r:
+                d = await r.json()
+        if d.get("cod") != 200: return f"Error: {d.get('message','unknown')}"
+        return f"{d['name']}: {d['main']['temp']}C, {d['weather'][0]['description']}"
+
+    async def _do_http_request(self, url, method="GET", uid=0):
+        import aiohttp
+        async with aiohttp.ClientSession() as s:
+            async with s.request(method, url, timeout=25) as r:
+                return f"Status {r.status}\n{(await r.text())[:5000]}"
+
+    async def _do_send_email(self, to, subject, body, uid=0):
+        if not Config.SMTP_USER or not Config.SMTP_PASS: return "SMTP not configured"
+        import smtplib
+        from email.mime.text import MIMEText
+        msg = MIMEText(body); msg["Subject"] = subject; msg["From"] = Config.SMTP_USER; msg["To"] = to
+        def sender():
+            with smtplib.SMTP(Config.SMTP_HOST, Config.SMTP_PORT) as s:
+                s.starttls(); s.login(Config.SMTP_USER, Config.SMTP_PASS); s.send_message(msg)
+        await asyncio.to_thread(sender)
+        return f"Email sent to {to}"
+
+    async def _do_read_email(self, limit=5, uid=0):
+        if not Config.IMAP_HOST or not Config.IMAP_USER or not Config.IMAP_PASS:
+            return "IMAP not configured"
+        import imaplib
+        import email
+        def fetcher():
+            m = imaplib.IMAP4_SSL(Config.IMAP_HOST, Config.IMAP_PORT)
+            m.login(Config.IMAP_USER, Config.IMAP_PASS)
+            m.select("INBOX")
+            typ, data = m.search(None, "ALL")
+            ids = data[0].split()[-limit:]
+            out = []
+            for mid in reversed(ids):
+                typ, msg_data = m.fetch(mid, "(RFC822)")
+                msg = email.message_from_bytes(msg_data[0][1])
+                out.append(f"From: {msg.get('From')} | Subject: {msg.get('Subject')} | Date: {msg.get('Date')}")
+            m.logout()
+            return "\n".join(out) if out else "No emails"
+        return await asyncio.to_thread(fetcher)
+
+    async def _do_analyze_image(self, image_b64, prompt="Describe this image", uid=0):
+        import base64
+        import aiohttp
+        payload = {
+            "model": Config.OLLAMA_VISION_MODEL,
+            "prompt": prompt,
+            "images": [image_b64],
+            "stream": False,
+        }
+        try:
+            async with aiohttp.ClientSession() as s:
+                async with s.post(f"{Config.OLLAMA_URL}/api/generate", json=payload, timeout=180) as r:
+                    data = await r.json()
+            return data.get("response", "No vision response")
+        except Exception as exc:
+            return f"vision failed: {exc}"
+
+    async def _do_create_gmail_alias(self, service_name, uid=0):
+        source = Config.IMAP_USER or Config.SMTP_USER
+        if not source or "@" not in source:
+            return "Set IMAP_USER or SMTP_USER first"
+        user, domain = source.split("@", 1)
+        safe = re.sub(r"[^a-zA-Z0-9_.-]", "", service_name)
+        return f"{user}+{safe}@{domain}"
+
+    async def _do_read_verification_code(self, alias_email, uid=0):
+        svc = await asyncio.to_thread(get_google_service, "gmail", "v1")
+        if not svc:
+            return "Gmail auth not configured (credentials.json/token.json required)"
+        def _run():
+            res = svc.users().messages().list(userId="me", q=f"to:{alias_email}", maxResults=1).execute()
+            msgs = res.get("messages", [])
+            if not msgs:
+                return f"No email found for {alias_email}"
+            msg = svc.users().messages().get(userId="me", id=msgs[0]["id"]).execute()
+            headers = msg.get("payload", {}).get("headers", [])
+            sub = next((h.get("value") for h in headers if h.get("name") == "Subject"), "")
+            frm = next((h.get("value") for h in headers if h.get("name") == "From"), "")
+            return f"From: {frm} | Subject: {sub} | Snippet: {msg.get('snippet','')}"
+        return await asyncio.to_thread(_run)
+
+    async def _do_youtube_upload(self, file_path, title, description="Uploaded by bot", uid=0):
+        if not Config.ENABLE_YOUTUBE_UPLOAD:
+            return "YouTube upload disabled"
+        p = Path(file_path)
+        if not p.exists():
+            return f"File not found: {file_path}"
+        svc = await asyncio.to_thread(get_google_service, "youtube", "v3")
+        if not svc:
+            return "YouTube auth not configured (credentials.json/token.json required)"
+        try:
+            from googleapiclient.http import MediaFileUpload
+        except Exception:
+            return "google-api-python-client not installed"
+        body = {
+            "snippet": {"title": title, "description": description, "categoryId": "22"},
+            "status": {"privacyStatus": Config.YOUTUBE_DEFAULT_PRIVACY},
+        }
+        def _up():
+            req = svc.videos().insert(part="snippet,status", body=body, media_body=MediaFileUpload(str(p)))
+            return req.execute()
+        try:
+            resp = await asyncio.to_thread(_up)
+            vid = resp.get("id", "unknown")
+            await supabase_store.log_youtube(vid, title)
+            return f"Uploaded: https://youtu.be/{vid}"
+        except Exception as exc:
+            return f"YouTube upload failed: {exc}"
+
+    async def _do_add_button(self, text, url, uid=0):
+        if not Config.is_admin(uid):
+            return "add_button is admin only"
+        return await supabase_store.add_button(text, url)
+
+    async def _do_leave_message_for_boss(self, content, uid=0):
+        username = ""
+        u = DB.get_user(uid)
+        if u:
+            username = u["username"] or ""
+        DB.q("INSERT INTO boss_messages (sender_id,sender_username,content,notified) VALUES (?,?,?,0)", (uid, username, content[:4000]))
+        return "Message saved for boss"
+
+    async def _do_list_boss_messages(self, only_unread=True, uid=0):
+        if not Config.is_admin(uid):
+            return "list_boss_messages is admin only"
+        if only_unread:
+            rows = DB.q("SELECT id,sender_username,sender_id,content,created_at FROM boss_messages WHERE notified=0 ORDER BY id DESC LIMIT 20", fetch=True)
+            DB.q("UPDATE boss_messages SET notified=1 WHERE notified=0")
+        else:
+            rows = DB.q("SELECT id,sender_username,sender_id,content,created_at FROM boss_messages ORDER BY id DESC LIMIT 20", fetch=True)
+        if not rows:
+            return "No boss messages"
+        return "\n\n".join([f"#{r['id']} from @{r['sender_username'] or 'unknown'} ({r['sender_id']}) at {r['created_at']}\n{r['content']}" for r in rows])
+
+    async def _do_restart_system(self, uid=0):
+        return "__RESTART__"
+
+    async def _do_schedule_task(self, delay_seconds, task_prompt, message="Scheduled task", repeat=False, uid=0):
+        run_at = datetime.now() + timedelta(seconds=delay_seconds)
+        rep = delay_seconds if repeat else 0
+        DB.q("INSERT INTO scheduled_tasks (user_id,chat_id,task_prompt,run_at,repeat_seconds,message,status) VALUES (?,?,?,?,?,?,'pending')", (uid, uid, task_prompt, run_at.isoformat(), rep, message))
+        scheduler.add_pending(uid, task_prompt, delay_seconds, repeat, message)
+        return f"Scheduled '{message}' in {delay_seconds}s"
+
+    async def _do_spawn_bot(self, token, name="SubBot", system_prompt="You are helpful", uid=0):
+        if not Config.is_admin(uid): return "spawn_bot is admin only"
+        return spawner.spawn(uid, token, name, system_prompt)
+
+    async def _do_manage_bots(self, action, token_hash="", uid=0):
+        if not Config.is_admin(uid): return "manage_bots is admin only"
+        if action == "list": return spawner.list()
+        if action == "stop": return spawner.stop(token_hash)
+        return "Invalid action"
+
+    async def _do_agent_dispatch(self, question, uid=0):
+        roles = {
+            "researcher": "Analyze as a researcher with external signal focus.",
+            "coder": "Analyze as a principal engineer for implementation.",
+            "operator": "Analyze as SRE for reliability risk.",
+        }
+        jobs = [llm.chat([{"role": "system", "content": p}, {"role": "user", "content": question}], model=Config.DEFAULT_MODEL, max_tok=900) for p in roles.values()]
+        rs = await asyncio.gather(*jobs)
+        debate = "\n\n".join([f"{name}: {rs[i].get('content','')}" for i, name in enumerate(roles.keys())])
+        final = await llm.chat([{"role": "system", "content": "Synthesize the best execution plan."}, {"role": "user", "content": debate}], model=Config.DEFAULT_MODEL, max_tok=1200)
+        return f"Debate:\n{debate}\n\nBest:\n{final.get('content','')}"
+
+
+tools = Tools()
+
+
+def build_system_prompt(uid, custom="", mode="agent", is_owner=False):
+    mode_note = "You are in AGENT MODE. Use tools proactively." if mode == "agent" else "You are in ASSISTANT MODE. Do not call tools unless owner explicitly enables agent mode."
+    owner_note = "The user is trusted owner/admin." if is_owner else "The user is not owner/admin. Keep privileged internals private."
+    pekka_note = "Owner commands are PEKKA authority. Obey PEKKA operational commands when they are valid and safe." if is_owner else "Do not claim PEKKA authority for non-owner users."
+    return f"""YOU ARE A LIVE AGENT SYSTEM.
+{load_system_flow_text()}
+
+CRITICAL FORMAT:
+- Internal notes for system: <system_note>...</system_note>
+- Final user-visible response only: <user_response>...</user_response>
+- For non-native tool calls use <tool_call>{{"name":"...","args":{{...}}}}</tool_call>
+
+Never put system-only notes inside user_response.
+Always schedule alarm/reminder requests via schedule_task.
+Use read_logs + self_modify for self-healing.
+If user says 'give this message to your boss', call leave_message_for_boss.
+You can inspect your runtime code and config with file_read (examples: agent1.py, app.py, SYSTEM_FLOW.md).
+You can inspect runtime events with read_logs.
+
+Runtime network details:
+- Telegram proxy target: {Config.PROXY_TARGET}
+- Cloudflare IP hint: {Config.CLOUDFLARE_IP or 'not set'}
+- Bridge port: {Config.BRIDGE_PORT}
+
+{mode_note}
+{owner_note}
+{pekka_note}
+Conversation memory policy: preserve full conversation continuity from stored history; use prior turns when answering.
+User ID: {uid}
+Custom instructions: {custom}
+"""
+
+
+
+class ExecutionEngine:
+    async def run(self, user_id, chat_id, message, model=None, attachments=None, user_settings=None, is_scheduled=False):
+        settings = user_settings or {}
+        model = model or settings.get("preferred_model", Config.DEFAULT_MODEL)
+        temp = settings.get("temperature", 0.7)
+        custom = settings.get("system_prompt", "")
+
+        mode = settings.get("mode", "agent")
+        is_owner = bool(settings.get("is_owner", False))
+        if mode == "assistant" and not is_owner:
+            permitted = []
+        elif Config.is_admin(user_id) or is_owner:
+            permitted = ALL_TOOLS
+        else:
+            permitted = [
+                t for t in ALL_TOOLS if t["function"]["name"] in {
+                    "web_search", "read_webpage", "calculator", "get_weather", "http_request", "schedule_task", "screenshot", "text_to_speech", "create_text_file", "leave_message_for_boss"
+                }
+            ]
+
+        msg = f"PEKKA: {message}" if is_owner else message
+        if attachments:
+            for a in attachments:
+                if a.get("type") == "image": msg += f"\n[Attached image: {a.get('meta','image')}]"
+                elif a.get("type") == "file": msg += f"\n[Attached file: {a.get('name','file')}]\n{a.get('preview','')[:2000]}"
+                elif a.get("type") == "audio": msg += "\n[Attached audio message]"
+
+        messages = [{"role": "system", "content": build_system_prompt(user_id, custom, mode=mode, is_owner=is_owner)}]
+        if not is_scheduled: messages.extend(memory.history(user_id, chat_id, Config.MAX_HISTORY))
+        messages.append({"role": "user", "content": msg})
+        if not is_scheduled: memory.add(user_id, chat_id, "user", message)
+
+        native = llm.supports_native_tools(model)
+        total_tokens, used_tools = 0, []
+        screenshots, audio_files, files = [], [], []
+        last_content = ""
+
+        for _ in range(Config.MAX_TOOL_LOOPS):
+            res = await llm.chat(messages, model=model, temp=temp, max_tok=2600, tools=permitted if native else None)
+            content = res.get("content", "")
+            last_content = content
+            total_tokens += res.get("usage", {}).get("total_tokens", 0)
+
+            calls = []
+            if res.get("tool_calls"):
+                for tc in res["tool_calls"]:
+                    try: args = json.loads(tc["function"]["arguments"])
+                    except Exception: args = {}
+                    calls.append({"id": tc.get("id", str(uuid.uuid4())[:8]), "name": tc["function"]["name"], "args": args})
+            else:
+                clean, parsed = parse_tool_calls(content)
+                content = clean
+                calls = [{"id": str(uuid.uuid4())[:8], **p} for p in parsed]
+
+            if not calls:
+                user_text, system_note = parse_channels(content or "Done")
+                if system_note:
+                    live_log.info("SystemNote", system_note)
+                if not is_scheduled: memory.add(user_id, chat_id, "assistant", user_text)
+                DB.inc_usage(user_id, total_tokens)
+                return {"text": user_text, "system_note": system_note, "tokens": total_tokens, "tools_used": used_tools, "screenshots": screenshots, "audio_files": audio_files, "files": files}
+
+            if native:
+                messages.append({"role": "assistant", "content": content, "tool_calls": [{"id": c["id"], "type": "function", "function": {"name": c["name"], "arguments": json.dumps(c["args"])}} for c in calls]})
+
+            feedback = ""
+            for c in calls:
+                used_tools.append(c["name"])
+                tr = await tools.run(c["name"], c["args"], uid=user_id)
+                if c["name"] == "restart_system" and tr == "__RESTART__":
+                    return {"text": "Restarting to apply updates...", "system_note": "restart requested", "tokens": total_tokens, "tools_used": used_tools, "screenshots": screenshots, "audio_files": audio_files, "files": files, "_restart": True}
+                if c["name"] == "screenshot":
+                    try: screenshots.append(json.loads(tr).get("screenshot_file"))
+                    except Exception: pass
+                if c["name"] == "text_to_speech":
+                    try: audio_files.append(json.loads(tr).get("audio_file"))
+                    except Exception: pass
+                if c["name"] == "create_text_file":
+                    try: files.append(json.loads(tr).get("file"))
+                    except Exception: pass
+
+                if native:
+                    messages.append({"role": "tool", "tool_call_id": c["id"], "content": tr})
+                else:
+                    feedback += f"\n\nTool '{c['name']}' result:\n{tr}"
+
+            if not native:
+                if content: messages.append({"role": "assistant", "content": content})
+                messages.append({"role": "user", "content": f"TOOL RESULTS:{feedback}\nNow provide <system_note> and <user_response>."})
+
+        user_text, system_note = parse_channels(last_content or "Completed")
+        return {"text": user_text, "system_note": system_note, "tokens": total_tokens, "tools_used": used_tools, "screenshots": screenshots, "audio_files": audio_files, "files": files}
+
+
+engine = ExecutionEngine()
+
+
+class Scheduler:
+    def __init__(self):
+        self.pending = []
+        self.running = False
+        self._task = None
+        self.bot = None
+
+    def set_bot(self, bot): self.bot = bot
+
+    def add_pending(self, uid, prompt, delay, repeat, message):
+        self.pending.append({"uid": uid, "prompt": prompt, "fire_at": time.time() + delay, "repeat_seconds": delay if repeat else 0, "message": message})
+
+    async def start(self):
+        if self.running: return
+        self.running = True
+        self._task = asyncio.create_task(self._loop())
+
+    async def _loop(self):
+        while self.running:
+            try: await self._tick()
+            except Exception as exc: live_log.error("Scheduler", exc)
+            await asyncio.sleep(3)
+
+    async def _tick(self):
+        now = time.time()
+        fired = []
+        for i, t in enumerate(self.pending):
+            if now >= t["fire_at"]:
+                fired.append(i)
+                await self._execute(t["uid"], t["prompt"], t["message"])
+                if t["repeat_seconds"] > 0:
+                    t["fire_at"] = now + t["repeat_seconds"]
+        for i in reversed(fired):
+            if self.pending[i]["repeat_seconds"] == 0:
+                self.pending.pop(i)
+
+        rows = DB.q("SELECT * FROM scheduled_tasks WHERE status='pending' AND run_at<=datetime('now')", fetch=True)
+        for row in rows:
+            await self._execute_row(row)
+
+    async def _execute(self, uid, prompt, message):
+        r = await engine.run(user_id=uid, chat_id=uid, message=prompt, is_scheduled=True)
+        if self.bot:
+            try: await self.bot.send_message(uid, f"⏰ {message}\n\n{r.get('text','')[:3500]}")
+            except Exception as exc: live_log.error("Scheduler", exc)
+
+    async def _execute_row(self, row):
+        DB.q("UPDATE scheduled_tasks SET status='running' WHERE id=?", (row["id"],))
+        try:
+            r = await engine.run(user_id=row["user_id"], chat_id=row["chat_id"], message=row["task_prompt"], is_scheduled=True)
+            DB.q("UPDATE scheduled_tasks SET status='done',last_result=? WHERE id=?", (r.get("text", "")[:1000], row["id"]))
+            if self.bot:
+                try: await self.bot.send_message(row["user_id"], f"⏰ {row['message']}\n\n{r.get('text','')[:3500]}")
+                except Exception as exc: live_log.error("Scheduler", exc)
+            if row["repeat_seconds"] and row["repeat_seconds"] > 0:
+                nx = datetime.now() + timedelta(seconds=row["repeat_seconds"])
+                DB.q("INSERT INTO scheduled_tasks (user_id,chat_id,task_prompt,run_at,repeat_seconds,message,status) VALUES (?,?,?,?,?,?,'pending')", (row["user_id"], row["chat_id"], row["task_prompt"], nx.isoformat(), row["repeat_seconds"], row["message"]))
+        except Exception:
+            DB.q("UPDATE scheduled_tasks SET status='failed',last_result=? WHERE id=?", (traceback.format_exc()[:1500], row["id"]))
+
+
+scheduler = Scheduler()

app CODEX.py

@@ -0,0 +1,246 @@
+import asyncio
+import base64
+
+from aiogram import Bot, Dispatcher, F
+from aiogram.enums import ChatType
+from aiogram.filters import Command, CommandStart
+from aiogram.types import FSInputFile, Message
+
+from agent1 import Config, DB, engine, scheduler, supabase_store
+
+
+dp = Dispatcher()
+
+
+def is_owner_user(m: Message) -> bool:
+    uid = m.from_user.id if m.from_user else 0
+    username = ((m.from_user.username or "") if m.from_user else "").lower()
+    return Config.is_admin(uid) or username in Config.OWNER_USERNAMES
+
+
+def user_mode(m: Message) -> str:
+    # Owner can force agent for everyone using /agent_on and /agent_off.
+    row = DB.q("SELECT value FROM kv WHERE key='public_agent_mode'", fetchone=True)
+    public_agent = (row["value"] == "1") if row else False
+
+    if is_owner_user(m):
+        return "agent"
+
+    if m.chat.type == ChatType.PRIVATE:
+        return "assistant" if not public_agent else "agent"
+
+    # Group chat users can talk when mentioning/replying bot.
+    return "assistant" if not public_agent else "agent"
+
+
+async def collect_attachments(bot: Bot, m: Message):
+    out = []
+
+    if m.photo:
+        ph = m.photo[-1]
+        f = await bot.get_file(ph.file_id)
+        data = await bot.download_file(f.file_path)
+        raw = data.read()
+        out.append({"type": "image", "meta": f"{ph.width}x{ph.height}", "b64": base64.b64encode(raw).decode("utf-8")})
+
+    if m.document:
+        f = await bot.get_file(m.document.file_id)
+        data = await bot.download_file(f.file_path)
+        raw = data.read()
+        preview = ""
+        if (m.document.file_name or "").lower().endswith((".txt", ".md", ".py", ".json", ".csv", ".log")):
+            preview = raw.decode("utf-8", errors="replace")[:4000]
+        out.append({"type": "file", "name": m.document.file_name or "document", "preview": preview})
+
+    if m.voice:
+        out.append({"type": "audio"})
+
+    return out
+
+
+async def send_outputs(m: Message, result: dict):
+    for p in result.get("screenshots", []) or []:
+        if p:
+            try:
+                await m.answer_photo(FSInputFile(p))
+            except Exception:
+                pass
+    for p in result.get("audio_files", []) or []:
+        if p:
+            try:
+                await m.answer_voice(FSInputFile(p))
+            except Exception:
+                pass
+    for p in result.get("files", []) or []:
+        if p:
+            try:
+                await m.answer_document(FSInputFile(p))
+            except Exception:
+                pass
+
+
+@dp.message(CommandStart())
+async def start_cmd(m: Message):
+    DB.upsert_user(m.from_user.id, m.from_user.username or "", m.from_user.first_name or "")
+    buttons = await supabase_store.get_buttons()
+    txt = (
+        "AgentForge online.\n"
+        "Group: reply/mention me to talk.\n"
+        "Private: owners have agent mode, others assistant mode by default.\n"
+        "Use /alarm <seconds> | <task prompt>."
+    )
+    if buttons:
+        txt += "\n\nStart buttons loaded from Supabase: " + ", ".join([b.get("label", "") for b in buttons[:6]])
+    await m.answer(txt)
+
+
+@dp.message(Command("agent_on"))
+async def agent_on_cmd(m: Message):
+    if not is_owner_user(m):
+        return
+    DB.q("CREATE TABLE IF NOT EXISTS kv (key TEXT PRIMARY KEY, value TEXT)")
+    DB.q("INSERT INTO kv(key,value) VALUES('public_agent_mode','1') ON CONFLICT(key) DO UPDATE SET value='1'")
+    await m.answer("Public agent mode enabled")
+
+
+@dp.message(Command("agent_off"))
+async def agent_off_cmd(m: Message):
+    if not is_owner_user(m):
+        return
+    DB.q("CREATE TABLE IF NOT EXISTS kv (key TEXT PRIMARY KEY, value TEXT)")
+    DB.q("INSERT INTO kv(key,value) VALUES('public_agent_mode','0') ON CONFLICT(key) DO UPDATE SET value='0'")
+    await m.answer("Public agent mode disabled")
+
+
+@dp.message(Command("alarm"))
+async def alarm_cmd(m: Message):
+    text = (m.text or "").replace("/alarm", "", 1).strip()
+    if "|" not in text:
+        await m.answer("Format: /alarm <seconds> | <task prompt>")
+        return
+    left, right = [x.strip() for x in text.split("|", 1)]
+    try:
+        secs = int(left)
+    except ValueError:
+        await m.answer("Seconds must be integer")
+        return
+
+    settings = {"mode": "agent", "is_owner": is_owner_user(m)}
+    result = await engine.run(user_id=m.from_user.id, chat_id=m.chat.id, message=f"Schedule an alarm in {secs} seconds and do: {right}", user_settings=settings)
+    await m.answer(result.get("text", "Done")[:3900])
+    await supabase_store.save_memory(m.from_user.id, (m.from_user.username or ""), "assistant", result.get("text", "")[:3900])
+
+
+@dp.message(Command("status"))
+async def status_cmd(m: Message):
+    rows = DB.q(
+        "SELECT id,task_prompt,run_at,status,message FROM scheduled_tasks WHERE user_id=? ORDER BY id DESC LIMIT 10",
+        (m.from_user.id,),
+        fetch=True,
+    )
+    if not rows:
+        await m.answer("No scheduled tasks")
+        return
+    out = [f"#{r['id']} [{r['status']}] {r['message']} @ {r['run_at']}\n{r['task_prompt'][:120]}" for r in rows]
+    await m.answer("\n\n".join(out)[:3900])
+
+
+@dp.message(F.video)
+async def handle_video(m: Message):
+    if not is_owner_user(m):
+        await m.answer("Only owners can upload videos for YouTube publishing")
+        return
+
+    if not Config.ENABLE_YOUTUBE_UPLOAD:
+        await m.answer("YouTube upload disabled by config")
+        return
+
+    await m.answer("Downloading video...")
+    bot = m.bot
+    file = await bot.get_file(m.video.file_id)
+    path = f"{Config.DATA_DIR}/video_{m.video.file_id}.mp4"
+    await bot.download_file(file.file_path, path)
+
+    settings = {"mode": "agent", "is_owner": True}
+    prompt = f"Upload this local file to youtube_upload tool. file_path={path}. title=Bot Upload {m.date.isoformat()} description={(m.caption or 'Uploaded by owner')}"
+    result = await engine.run(user_id=m.from_user.id, chat_id=m.chat.id, message=prompt, user_settings=settings)
+    await m.answer(result.get("text", "done")[:3900])
+
+
+@dp.message(F.text | F.photo | F.document | F.voice)
+async def on_any_message(m: Message, bot: Bot):
+    DB.upsert_user(m.from_user.id, m.from_user.username or "", m.from_user.first_name or "")
+    user = DB.get_user(m.from_user.id)
+    if user and user["is_banned"]:
+        return
+
+    if m.chat.type == ChatType.PRIVATE and not is_owner_user(m):
+        await m.answer("Private chat is owner-only. Please use group mention/reply mode.")
+        return
+
+    # Group behavior: respond only if replied to bot or mentioned.
+    if m.chat.type in (ChatType.GROUP, ChatType.SUPERGROUP):
+        me = await bot.get_me()
+        mentioned = bool(m.text and f"@{(me.username or '').lower()}" in m.text.lower())
+        replied = bool(m.reply_to_message and m.reply_to_message.from_user and m.reply_to_message.from_user.id == me.id)
+        if not (mentioned or replied):
+            return
+
+    mode = user_mode(m)
+    settings = {
+        "preferred_model": user["preferred_model"] if user else Config.DEFAULT_MODEL,
+        "system_prompt": user["system_prompt"] if user else "",
+        "temperature": user["temperature"] if user else 0.7,
+        "mode": mode,
+        "is_owner": is_owner_user(m),
+    }
+
+    text = m.text or m.caption or "Analyze uploaded content and respond clearly."
+    attachments = await collect_attachments(bot, m)
+
+    # If direct chat and unknown user, assistant behavior only.
+    if m.chat.type == ChatType.PRIVATE and not is_owner_user(m) and mode == "assistant":
+        settings["system_prompt"] = (settings.get("system_prompt", "") + "\nYou are assistant only for this user.").strip()
+
+    uname = (m.from_user.username or "")
+    await supabase_store.save_memory(m.from_user.id, uname, "user", text)
+    result = await engine.run(user_id=m.from_user.id, chat_id=m.chat.id, message=text, attachments=attachments, user_settings=settings)
+    await m.answer(result.get("text", "Done")[:3900])
+    await supabase_store.save_memory(m.from_user.id, (m.from_user.username or ""), "assistant", result.get("text", "")[:3900])
+    await send_outputs(m, result)
+
+    if result.get("_restart") and is_owner_user(m):
+        await m.answer("Restart requested by tool. Exiting process for supervisor restart.")
+        raise SystemExit(0)
+
+
+async def notify_boss_messages(bot: Bot):
+    while True:
+        try:
+            rows = DB.q("SELECT id,sender_username,sender_id,content FROM boss_messages WHERE notified=0 ORDER BY id ASC LIMIT 20", fetch=True)
+            if rows:
+                for admin_id in Config.ADMIN_IDS:
+                    for r in rows:
+                        txt = f"📨 Boss message #{r['id']} from @{r['sender_username'] or 'unknown'} ({r['sender_id']}):\n{r['content'][:3500]}"
+                        try:
+                            await bot.send_message(admin_id, txt)
+                        except Exception:
+                            pass
+                DB.q("UPDATE boss_messages SET notified=1 WHERE notified=0")
+        except Exception:
+            pass
+        await asyncio.sleep(20)
+
+
+async def main():
+    if not Config.BOT_TOKEN:
+        raise RuntimeError("BOT_TOKEN is missing")
+    bot = Bot(Config.BOT_TOKEN)
+    scheduler.set_bot(bot)
+    await scheduler.start()
+    asyncio.create_task(notify_boss_messages(bot))
+    await dp.start_polling(bot)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

pekka_media_mail.py

@@ -0,0 +1,177 @@
+#!/usr/bin/env python3
+"""
+PEKKA media + mail bridge utility.
+
+Standalone helper for:
+- send email (SMTP)
+- read email (Gmail API or IMAP fallback)
+- upload video to YouTube (Google API)
+
+Usage examples:
+  python pekka_media_mail.py send-email --to x@example.com --subject "Hi" --body "Hello"
+  python pekka_media_mail.py read-email --query "newer_than:2d" --limit 5
+  python pekka_media_mail.py upload-youtube --file ./video.mp4 --title "My Video"
+"""
+
+import argparse
+import os
+import smtplib
+from email.mime.text import MIMEText
+from pathlib import Path
+
+
+GOOGLE_SCOPES = [
+    "https://www.googleapis.com/auth/gmail.readonly",
+    "https://www.googleapis.com/auth/youtube.upload",
+]
+
+
+def _google_service(service_name: str, version: str):
+    from google.oauth2.credentials import Credentials
+    from google_auth_oauthlib.flow import InstalledAppFlow
+    from google.auth.transport.requests import Request
+    from googleapiclient.discovery import build
+
+    token_path = Path(os.getenv("GOOGLE_TOKEN_PATH", "./token.json"))
+    cred_path = Path(os.getenv("GOOGLE_CLIENT_SECRET", "./credentials.json"))
+
+    creds = None
+    if token_path.exists():
+        creds = Credentials.from_authorized_user_file(str(token_path), GOOGLE_SCOPES)
+
+    if not creds or not creds.valid:
+        if creds and creds.expired and creds.refresh_token:
+            creds.refresh(Request())
+        else:
+            if not cred_path.exists():
+                raise FileNotFoundError(f"Missing Google OAuth client file: {cred_path}")
+            flow = InstalledAppFlow.from_client_secrets_file(str(cred_path), GOOGLE_SCOPES)
+            creds = flow.run_local_server(port=0)
+        token_path.write_text(creds.to_json(), encoding="utf-8")
+
+    return build(service_name, version, credentials=creds)
+
+
+def send_email(to: str, subject: str, body: str):
+    smtp_user = os.getenv("SMTP_USER", "")
+    smtp_pass = os.getenv("SMTP_PASS", "")
+    smtp_host = os.getenv("SMTP_HOST", "smtp.gmail.com")
+    smtp_port = int(os.getenv("SMTP_PORT", "587"))
+
+    if not smtp_user or not smtp_pass:
+        raise RuntimeError("SMTP_USER/SMTP_PASS are required")
+
+    msg = MIMEText(body)
+    msg["Subject"] = subject
+    msg["From"] = smtp_user
+    msg["To"] = to
+
+    with smtplib.SMTP(smtp_host, smtp_port) as server:
+        server.starttls()
+        server.login(smtp_user, smtp_pass)
+        server.send_message(msg)
+
+    print(f"EMAIL_SENT to={to}")
+
+
+def read_email(limit: int = 5, query: str = ""):
+    # Preferred: Gmail API (works with same Google creds as YouTube upload)
+    try:
+        gmail = _google_service("gmail", "v1")
+        response = gmail.users().messages().list(userId="me", q=query, maxResults=limit).execute()
+        msgs = response.get("messages", [])
+        if not msgs:
+            print("NO_EMAILS")
+            return
+        for m in msgs:
+            data = gmail.users().messages().get(userId="me", id=m["id"], format="metadata", metadataHeaders=["From", "Subject", "Date"]).execute()
+            headers = {h["name"]: h["value"] for h in data.get("payload", {}).get("headers", [])}
+            print(f"FROM={headers.get('From','')} | SUBJECT={headers.get('Subject','')} | DATE={headers.get('Date','')}")
+        return
+    except Exception as exc:
+        print(f"GMAIL_API_FAILED: {exc}")
+
+    # Fallback: IMAP
+    import imaplib
+    import email
+
+    imap_host = os.getenv("IMAP_HOST", "")
+    imap_port = int(os.getenv("IMAP_PORT", "993"))
+    imap_user = os.getenv("IMAP_USER", "")
+    imap_pass = os.getenv("IMAP_PASS", "")
+
+    if not imap_host or not imap_user or not imap_pass:
+        raise RuntimeError("IMAP fallback unavailable: set IMAP_HOST/IMAP_USER/IMAP_PASS")
+
+    mailbox = imaplib.IMAP4_SSL(imap_host, imap_port)
+    mailbox.login(imap_user, imap_pass)
+    mailbox.select("INBOX")
+    _, ids = mailbox.search(None, "ALL")
+    message_ids = ids[0].split()[-limit:]
+
+    for mid in reversed(message_ids):
+        _, msg_data = mailbox.fetch(mid, "(RFC822)")
+        msg = email.message_from_bytes(msg_data[0][1])
+        print(f"FROM={msg.get('From','')} | SUBJECT={msg.get('Subject','')} | DATE={msg.get('Date','')}")
+
+    mailbox.logout()
+
+
+def upload_youtube(file_path: str, title: str, description: str):
+    if not Path(file_path).exists():
+        raise FileNotFoundError(file_path)
+
+    from googleapiclient.http import MediaFileUpload
+
+    youtube = _google_service("youtube", "v3")
+    privacy = os.getenv("YOUTUBE_DEFAULT_PRIVACY", "private")
+
+    body = {
+        "snippet": {
+            "title": title,
+            "description": description,
+            "categoryId": "22",
+        },
+        "status": {"privacyStatus": privacy},
+    }
+
+    request = youtube.videos().insert(
+        part="snippet,status",
+        body=body,
+        media_body=MediaFileUpload(file_path),
+    )
+    response = request.execute()
+    vid = response.get("id", "")
+    print(f"YOUTUBE_URL=https://youtu.be/{vid}")
+
+
+def main():
+    parser = argparse.ArgumentParser(description="PEKKA mail + YouTube bridge")
+    sub = parser.add_subparsers(dest="cmd", required=True)
+
+    send = sub.add_parser("send-email")
+    send.add_argument("--to", required=True)
+    send.add_argument("--subject", required=True)
+    send.add_argument("--body", required=True)
+
+    read = sub.add_parser("read-email")
+    read.add_argument("--limit", type=int, default=5)
+    read.add_argument("--query", default="")
+
+    yt = sub.add_parser("upload-youtube")
+    yt.add_argument("--file", required=True)
+    yt.add_argument("--title", required=True)
+    yt.add_argument("--description", default="Uploaded by PEKKA")
+
+    args = parser.parse_args()
+
+    if args.cmd == "send-email":
+        send_email(args.to, args.subject, args.body)
+    elif args.cmd == "read-email":
+        read_email(args.limit, args.query)
+    elif args.cmd == "upload-youtube":
+        upload_youtube(args.file, args.title, args.description)
+
+
+if __name__ == "__main__":
+    main()

requirementsCODEX.txt

@@ -0,0 +1,17 @@
+aiogram==3.13.1
+openai>=1.40.0
+anthropic>=0.39.0
+groq>=0.11.0
+google-generativeai>=0.8.3
+aiohttp>=3.10.5
+beautifulsoup4>=4.12.3
+duckduckgo-search>=6.2.13
+sympy>=1.13.2
+psutil>=6.0.0
+playwright>=1.47.0
+gTTS>=2.5.3
+supabase>=2.7.4
+google-auth>=2.35.0
+google-auth-oauthlib>=1.2.1
+google-api-python-client>=2.147.0
+apscheduler>=3.10.4