#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ FlynCode 0.1.0 — AI-агент в терминале с доступом к файлам, shell и интернету. Один файл, только стандартная библиотека. Python 3.8+ (Windows 10+, macOS, Linux). Запуск: python flyncode.py Команды: /help /model /think /provider /perm /cd /settings /usage /models /key /skills /clear /exit """ import atexit import base64 import copy import difflib import fnmatch import glob import hashlib import io import platform import zipfile import html import json import os import re import shlex import shutil import signal import socket import ssl import subprocess import sys import textwrap import threading import time import unicodedata import urllib.error import urllib.parse import urllib.request import pathlib import random import tempfile from datetime import datetime from html.parser import HTMLParser VERSION = "0.1.0" IS_WIN = os.name == "nt" if IS_WIN: import msvcrt else: import select as _sel import termios CFG_DIR = os.path.join(os.path.expanduser("~"), ".flyncode") CFG_FILE = os.path.join(CFG_DIR, "config.json") MAX_STEPS = 30 UA = {"User-Agent": "Mozilla/5.0 FlynCode/" + VERSION} # ════════════════════════════════════════════════════════════════════════════ # ТЕМА И ВЫВОД (палитра OpenCode Dark: синий + янтарный) # ════════════════════════════════════════════════════════════════════════════ THEMES = { "flyn": ("OpenCode Dark (синий + янтарный)", (97, 175, 239), (229, 192, 123)), "orange": ("Оранжево-белый", (255, 138, 36), (245, 245, 245)), "ocean": ("Ocean (голубой)", (86, 182, 255), (120, 130, 255)), "matrix": ("Matrix (зелёный)", (127, 216, 143), (60, 170, 100)), "rose": ("Rose (розовый)", (255, 121, 198), (189, 147, 249)), "mono": ("Mono (серый)", (230, 230, 230), (150, 150, 150)), } PAL = { "text": (220, 223, 228), "muted": (142, 149, 163), "dim": (98, 105, 122), "err": (224, 108, 117), "ok": (152, 195, 121), "warn": (229, 192, 123), "info": (86, 182, 194), "bg_box": (24, 26, 31), "bg_top": (33, 37, 43), "accent": THEMES["flyn"][1], "accent2": THEMES["flyn"][2], } RESET = "\x1b[0m" MODES = { "build": ("Build", "полный доступ: файлы, bash, интернет"), "plan": ("Plan", "только читаю и планирую, ничего не меняю"), "chat": ("Chat", "просто общаемся, инструменты выключены"), } MODE_ORDER = ["build", "plan", "chat"] MODE_COLOR = {"build": (102, 194, 255), "plan": (235, 150, 60), "chat": (130, 137, 151)} def rgbtxt(rgb, s, bold=False): return ("\x1b[1m" if bold else "") + fg(rgb) + str(s) + RESET PLAN_BLOCKED = {"ask_agent", "write_file", "append_file", "edit_file", "multi_edit", "apply_patch", "move_path", "delete_path", "bash", "run_script", "git", "download_file"} MODE_RULES = { "build": ("Режим Build: делай задачу сам — читай, пиши, запускай. Если в чате сохранён план, " "выполняй его по шагам и отмечай выполненные шаги через todo_write."), "plan": ("Режим Plan: разбери задачу и составь план по шагам через todo_write — он сохранится в чате " "и будет виден пользователю. Файлы не меняй и команды не запускай. Когда план готов, кратко скажи, " "что дальше можно переключиться в Build (tab) и выполнить его. Если план уже есть — обнови его, а не делай новый."), "chat": ("Режим Chat: обычный разговор. Отвечай живо и по-человечески, коротко, " "без отчётов и списков, если их не просят. Инструменты не используй."), } def set_theme(name): t = THEMES.get(name, THEMES["flyn"]) PAL["accent"], PAL["accent2"] = t[1], t[2] def fg(rgb): return "\x1b[38;2;%d;%d;%dm" % rgb def bg(rgb): return "\x1b[48;2;%d;%d;%dm" % rgb def c(name, s, bold=False, italic=False): pre = ("\x1b[1m" if bold else "") + ("\x1b[3m" if italic else "") return pre + fg(PAL[name]) + str(s) + RESET def link(url, text=None): """Кликабельная ссылка (OSC 8): в поддерживающих терминалах открывается по Ctrl/Cmd+клику.""" text = text or url return "\x1b[4m" + fg(PAL["info"]) + "\x1b]8;;%s\x1b\\%s\x1b]8;;\x1b\\" % (url, text) + RESET def cw(ch): if unicodedata.combining(ch): return 0 return 2 if unicodedata.east_asian_width(ch) in ("W", "F") else 1 def vlen(s): return sum(cw(ch) for ch in s) ANSI_RE = re.compile(r"\x1b\[[0-9;?]*[A-Za-z]|\x1b\][^\x07\x1b]*(?:\x07|\x1b\\)") def strip_ansi(s): return ANSI_RE.sub("", s) def vlen_ansi(s): return vlen(strip_ansi(s)) def padv_ansi(s, w): """Дополняет строку пробелами до видимой ширины w (ANSI-коды не считаются).""" return s + " " * max(0, w - vlen_ansi(s)) def box_line(s, w=None): """Строка «карточки» OpenCode: тёмная подложка на всю ширину. RESET внутри строки не должен сбрасывать подложку, поэтому после каждого RESET её включаем заново.""" w = (w or term_w()) - 1 B = bg(PAL["bg_box"]) s = s.replace(RESET, RESET + B) return B + padv_ansi(s, w) + RESET def clip(s, w): if w <= 0: return "" if vlen(s) <= w: return s res, n = "", 0 for ch in s: if n + cw(ch) > w - 1: break res += ch n += cw(ch) return res + "…" def padv(s, w): return s + " " * max(0, w - vlen(s)) SB_W = 30 # ширина правой панели: компактная SIDEBAR = {"on": True} # переключается /sidebar и настройкой sidebar APP_REF = {"app": None} # текущее приложение — для правой панели def full_cols(): return max(40, shutil.get_terminal_size((100, 30)).columns) def sidebar_active(): return SIDEBAR["on"] and APP_REF["app"] is not None and full_cols() >= 100 def term_w(): """Ширина основной колонки: если справа панель, то всё, что левее неё.""" cols = full_cols() if sidebar_active(): return cols - SB_W - 1 return cols def term_h(): return max(12, shutil.get_terminal_size((100, 30)).lines) # ════════════════════════════════════════════════════════════════════════════ # ЭКРАН: лента вывода + активный виджет внизу + правая панель. Всё рисуется кадром # в альтернативном буфере, поэтому панель не дублируется и не уезжает при прокрутке. # ════════════════════════════════════════════════════════════════════════════ _CSI_RE = re.compile(r"\x1b\[[0-9;?]*([A-Za-z])") _ESC78_RE = re.compile(r"\x1b[78]") _CTRL_RE = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1a\x1c-\x1f\x7f]") _SGR_RE = re.compile(r"\x1b\[[0-9;]*m|\x1b\][^\x07\x1b]*(?:\x07|\x1b\\)") SB_CACHE = {"key": None, "used": 0} def _keep_sgr(m): return m.group(0) if m.group(1) == "m" else "" class Screen: """Вся картинка терминала. Вывод — в ленту; виджет (строка ввода, меню, спиннер) кладёт свои строки в слот «внизу»; кадр пересобирается целиком, без курсорных трюков.""" MAX_LINES = 6000 def __init__(self): self.lines = [] # логические строки ленты (с SGR-цветами) self.cur = "" # незавершённая строка self.dyn = [] # строки активного виджета self.dyn_cur = None # курсор внутри виджета: (строка, колонка) или None self.scroll = 0 # на сколько логических строк прокручена лента вверх self.lock = threading.RLock() self.dirty = True self.enabled = False self.last = 0.0 self.center = False # стартовый экран: блок по центру def start(self): if not sys.stdout.isatty(): return self.enabled = True sys.stdout.write("\x1b[?1049h\x1b[2J") sys.stdout.flush() threading.Thread(target=self._loop, daemon=True).start() atexit.register(self.stop) def stop(self): if not self.enabled: return self.enabled = False sys.stdout.write("\x1b[0m\x1b[?25h\x1b[?1049l") sys.stdout.flush() def _loop(self): while self.enabled: time.sleep(0.03) with self.lock: if self.dirty or time.time() - self.last > 0.5: try: self._draw() except Exception: pass # ---- вывод в ленту def write(self, s): if not s: return with self.lock: for k, seg in enumerate(s.replace("\r\n", "\n").split("\r")): if k: self.cur = "" # возврат каретки: строка переписывается self._put(seg) self.dirty = True def _put(self, t): t = _ESC78_RE.sub("", _CSI_RE.sub(_keep_sgr, t)) for i, p in enumerate(t.split("\n")): if i: self.lines.append(self.cur) self.cur = "" self.cur += _CTRL_RE.sub("", p).replace("\t", " ") if len(self.lines) > self.MAX_LINES: del self.lines[: len(self.lines) - self.MAX_LINES] # ---- виджеты def set_dyn(self, lines, cursor=None): with self.lock: self.dyn = list(lines) self.dyn_cur = cursor self.dirty = True def clear(self): with self.lock: self.lines = [] self.cur = "" self.scroll = 0 self.dirty = True def scroll_by(self, n): with self.lock: self.scroll = max(0, min(self.scroll + n, len(self.lines))) self.dirty = True # ---- сборка кадра def _wrap(self, s, W): """Режет строку с SGR-кодами на видимые строки шириной W; цвет переходит на следующую строку.""" if W <= 0: return [""] rows, row, w, active, i = [], "", 0, "", 0 while i < len(s): m = _SGR_RE.match(s, i) if m: code = m.group(0) row += code if code.startswith("\x1b["): active = "" if code == RESET else active + code i = m.end() continue ch = s[i] cwid = cw(ch) if w + cwid > W: rows.append(row + RESET) row, w = active, 0 row += ch w += cwid i += 1 rows.append(row) return rows def _body(self, W, n): if n <= 0: return [] if self.center and self.scroll == 0: allrows = [] for line in self.lines + [self.cur]: allrows += self._wrap(line, W) if len(allrows) >= n: return allrows[-n:] top = (n - len(allrows)) // 2 return [""] * top + allrows + [""] * (n - top - len(allrows)) logical = self.lines + [self.cur] i = len(logical) - self.scroll - 1 rows = [] while i >= 0 and len(rows) < n: rows = self._wrap(logical[i], W) + rows i -= 1 return rows[-n:] def _draw(self): R = term_h() C = full_cols() sb = sidebar_active() MW = (C - SB_W - 1) if sb else (C - 1) # ширина основной колонки dyn_rows = [self._wrap(d, MW)[0] for d in self.dyn][-R:] body_n = R - len(dyn_rows) body = self._body(MW, body_n) rows = [""] * (body_n - len(body)) + body + dyn_rows side = sidebar_rows(R) if sb else [] buf = ["\x1b[?25l"] for r in range(R): left = rows[r] if r < len(rows) else "" text = RESET + padv_ansi(left, MW) + RESET if sb: right = side[r] if r < len(side) else "" text += fg(PAL["dim"]) + "│" + RESET + padv_ansi(right, SB_W - 1) + RESET buf.append("\x1b[%d;1H" % (r + 1) + text) if self.dyn_cur is not None: cr, cc = self.dyn_cur row = min(R - 1, body_n + cr) buf.append("\x1b[%d;%dH\x1b[?25h" % (row + 1, min(cc, MW - 1) + 1)) sys.stdout.write("".join(buf)) sys.stdout.flush() self.dirty = False self.last = time.time() SCREEN = Screen() def out(s=""): SCREEN.write(s) def _home_short(p): home = os.path.expanduser("~") if p.lower().startswith(home.lower()): p = "~" + p[len(home):] return p.replace("\\", "/") def sidebar_rows(H): """Содержимое правой панели: заголовок чата, Context, План, доступ, путь и версия.""" app = APP_REF["app"] W = SB_W - 2 rows = [""] title = "Новый чат" for m in app.messages: if m.get("role") == "user" and not m.get("summary"): title = " ".join(str(m.get("content", "")).split()) or title break rows.append(c("text", clip(title, W), bold=True)) rows.append("") rows.append(c("text", "Context", bold=True)) key = (len(app.messages), app.model) if SB_CACHE["key"] != key: SB_CACHE["key"] = key SB_CACHE["used"] = msgs_tokens(app.messages) if app.messages else 0 used = SB_CACHE["used"] lim = 0 for m in app.models: if m.get("id") == app.model: lim = m.get("ctx") or 0 lim = lim or (app.cfg["settings"].get("compact_at", 60000) or 60000) rows.append(c("muted", "{:,} tokens".format(used))) rows.append(c("muted", "%d%% used" % min(999, used * 100 // lim))) rows.append(c("muted", "$%.2f spent" % app.tot["cost"])) if app.plan: done = sum(1 for t in app.plan if t["done"]) rows.append("") rows.append(c("text", "Plan", bold=True) + c("muted", " %d/%d" % (done, len(app.plan)))) for t in app.plan[:8]: mark = c("ok", "[x]") if t["done"] else c("muted", "[ ]") rows.append(mark + " " + c("muted" if t["done"] else "text", clip(t["text"], W - 2))) rows.append("") rows.append(c("text", "Access", bold=True)) st = app.cfg["settings"] if not st["tools"]: rows.append(c("muted", "инструменты выкл")) else: n_ask = sum(1 for k in ASK if st["perms"].get(k) == "ask") if n_ask and not app.allow_all: rows.append(c("warn", "спрашиваю: %d" % n_ask)) else: rows.append(c("ok", "без вопросов")) rows.append(rgbtxt(MODE_COLOR[app.mode], "режим: " + MODES[app.mode][0])) bottom = [c("muted", clip(_home_short(app.cwd), W)), fg(PAL["ok"]) + "•" + RESET + " " + c("text", "FlynCode " + VERSION, bold=True)] top = rows[:max(0, H - len(bottom) - 1)] pad = [""] * max(0, H - len(top) - len(bottom)) return top + pad + bottom def println(s=""): out(s + "\n") def clear_screen(): SCREEN.clear() def fmt_tok(n): n = int(n or 0) if n >= 1_000_000: return "%.2fM" % (n / 1e6) if n >= 10_000: return "%.1fk" % (n / 1e3) if n >= 1000: return "%.2fk" % (n / 1e3) return str(n) def fmt_dur(s): if s < 60: return "%.1fс" % s return "%dм %02dс" % (int(s // 60), int(s % 60)) def fmt_bytes(n): v = float(n) for u in ("Б", "КБ", "МБ"): if v < 1024: return "%.0f %s" % (v, u) v /= 1024 return "%.1f ГБ" % v def est_tokens(text): return max(1, int(len(text) / 3.2)) def lerp(a, b, t): return tuple(int(a[i] + (b[i] - a[i]) * t) for i in range(3)) # Высота символа в пикселях для классической консоли Windows (cmd, PowerShell, «Командная строка»). # Больше — крупнее весь текст. Меняется переменной окружения FLYNCODE_FONT, по умолчанию 22. CONSOLE_FONT_PX = int(os.environ.get("FLYNCODE_FONT") or 22) def set_console_font(px): """Крупный шрифт консоли Windows через SetCurrentConsoleFontEx. Windows Terminal и VS Code это игнорируют.""" try: import ctypes class COORD(ctypes.Structure): _fields_ = [("X", ctypes.c_short), ("Y", ctypes.c_short)] class CONSOLE_FONT_INFOEX(ctypes.Structure): _fields_ = [("cbSize", ctypes.c_ulong), ("nFont", ctypes.c_ulong), ("dwFontSize", COORD), ("FontFamily", ctypes.c_uint), ("FontWeight", ctypes.c_uint), ("FaceName", ctypes.c_wchar * 32)] k = ctypes.windll.kernel32 info = CONSOLE_FONT_INFOEX() info.cbSize = ctypes.sizeof(CONSOLE_FONT_INFOEX) info.dwFontSize.X = 0 # 0 — ширину символа Windows подберёт сам info.dwFontSize.Y = px info.FontFamily = 54 # FF_MODERN | FIXED_PITCH info.FontWeight = 400 info.FaceName = "Consolas" k.SetCurrentConsoleFontEx(k.GetStdHandle(-11), False, ctypes.byref(info)) except Exception: pass def setup_console(): try: sys.stdout.reconfigure(encoding="utf-8", errors="replace") sys.stdin.reconfigure(encoding="utf-8", errors="replace") except Exception: pass if IS_WIN: try: import ctypes k = ctypes.windll.kernel32 h = k.GetStdHandle(-11) mode = ctypes.c_ulong() k.GetConsoleMode(h, ctypes.byref(mode)) k.SetConsoleMode(h, mode.value | 0x0004) k.SetConsoleOutputCP(65001) except Exception: os.system("") set_console_font(CONSOLE_FONT_PX) # ════════════════════════════════════════════════════════════════════════════ # КЛАВИАТУРА (Windows / Linux / macOS) # ════════════════════════════════════════════════════════════════════════════ class RawMode: def __enter__(self): if not IS_WIN: self.fd = sys.stdin.fileno() self.old = termios.tcgetattr(self.fd) new = termios.tcgetattr(self.fd) new[3] &= ~(termios.ECHO | termios.ICANON | termios.ISIG | termios.IEXTEN) new[6][termios.VMIN] = 1 new[6][termios.VTIME] = 0 termios.tcsetattr(self.fd, termios.TCSADRAIN, new) return self def __exit__(self, *a): if not IS_WIN: termios.tcsetattr(self.fd, termios.TCSADRAIN, self.old) _CSI = {"A": "up", "B": "down", "C": "right", "D": "left", "H": "home", "F": "end"} _TILDE = {"1": "home", "3": "delete", "4": "end", "5": "pgup", "6": "pgdn", "7": "home", "8": "end"} _SIMPLE = {13: "enter", 10: "enter", 9: "tab", 127: "backspace", 8: "backspace", 3: "ctrl-c", 4: "ctrl-d", 1: "ctrl-a", 5: "ctrl-e", 16: "ctrl-p", 21: "ctrl-u", 23: "ctrl-w", 18: "ctrl-r", 14: "ctrl-n", 22: "ctrl-v", 7: "ctrl-g"} def key_pending(): if IS_WIN: return msvcrt.kbhit() return bool(_sel.select([sys.stdin.fileno()], [], [], 0)[0]) def read_key(): if IS_WIN: ch = msvcrt.getwch() if ch in ("\x00", "\xe0"): code = msvcrt.getwch() return {"H": "up", "P": "down", "K": "left", "M": "right", "S": "delete", "G": "home", "O": "end", "I": "pgup", "Q": "pgdn"}.get(code, "noop") m = {"\r": "enter", "\n": "enter", "\t": "tab", "\x08": "backspace", "\x7f": "backspace", "\x1b": "esc", "\x03": "ctrl-c", "\x04": "ctrl-d", "\x01": "ctrl-a", "\x05": "ctrl-e", "\x10": "ctrl-p", "\x15": "ctrl-u", "\x17": "ctrl-w", "\x12": "ctrl-r", "\x0e": "ctrl-n", "\x16": "ctrl-v", "\x07": "ctrl-g"} return m.get(ch, ch if ord(ch) >= 32 else "noop") fd = sys.stdin.fileno() b = os.read(fd, 1) if not b: return "ctrl-d" n = b[0] if n == 27: if not _sel.select([fd], [], [], 0.04)[0]: return "esc" if os.read(fd, 1) not in (b"[", b"O"): return "noop" seq = b"" while _sel.select([fd], [], [], 0.04)[0]: ch = os.read(fd, 1) if not ch: break seq += ch if 0x40 <= ch[0] <= 0x7E: break s = seq.decode("ascii", "ignore") if not s: return "noop" if s[-1] == "~": return _TILDE.get(s[:-1].split(";")[0], "noop") return _CSI.get(s[-1], "noop") if n in _SIMPLE: return _SIMPLE[n] if n < 32: return "noop" extra = 3 if n >= 0xF0 else 2 if n >= 0xE0 else 1 if n >= 0xC0 else 0 data = b + b"".join(os.read(fd, 1) for _ in range(extra)) try: return data.decode("utf-8") except UnicodeDecodeError: return "noop" # ════════════════════════════════════════════════════════════════════════════ # ЛОГО И СПИННЕР # ════════════════════════════════════════════════════════════════════════════ LETTERS = { "F": ["███████╗", "██╔════╝", "█████╗ ", "██╔══╝ ", "██║ ", "╚═╝ "], "L": ["██╗ ", "██║ ", "██║ ", "██║ ", "███████╗", "╚══════╝"], "Y": ["██╗ ██╗", "╚██╗ ██╔╝", " ╚████╔╝ ", " ╚██╔╝ ", " ██║ ", " ╚═╝ "], "N": ["███╗ ██╗", "████╗ ██║", "██╔██╗ ██║", "██║╚██╗██║", "██║ ╚████║", "╚═╝ ╚═══╝"], "C": [" ██████╗", "██╔════╝", "██║ ", "██║ ", "╚██████╗", " ╚═════╝"], "O": [" ██████╗ ", "██╔═══██╗", "██║ ██║", "██║ ██║", "╚██████╔╝", " ╚═════╝ "], "D": ["██████╗ ", "██╔══██╗", "██║ ██║", "██║ ██║", "██████╔╝", "╚═════╝ "], "E": ["███████╗", "██╔════╝", "█████╗ ", "██╔══╝ ", "███████╗", "╚══════╝"], } # Компактный логотип для узких окон: 6 пикселей в высоту, половинные блоки → 3 строки текста, 4 колонки на букву. PIXEL = { "F": ["1111", "1000", "1110", "1000", "1000", "1000"], "L": ["1000", "1000", "1000", "1000", "1000", "1111"], "Y": ["1001", "1001", "0110", "0110", "0100", "0100"], "N": ["1001", "1101", "1011", "1001", "1001", "1001"], "C": ["0111", "1000", "1000", "1000", "1000", "0111"], "O": ["0110", "1001", "1001", "1001", "1001", "0110"], "D": ["1110", "1001", "1001", "1001", "1001", "1110"], "E": ["1111", "1000", "1110", "1000", "1000", "1111"], } HALF = {(1, 1): "█", (1, 0): "▀", (0, 1): "▄", (0, 0): " "} def compact_logo(word): rows = [] for r in range(0, 6, 2): cells = [] for ch in word: g = PIXEL[ch] for x in range(4): cells.append(HALF[(int(g[r][x]), int(g[r + 1][x]))]) cells.append(" ") rows.append(cells[:-1]) return rows def show_logo(): W = term_w() word = "FLYNCODE" total = sum(len(LETTERS[ch][0]) for ch in word) small = sum(4 for _ in word) + len(word) - 1 println() if W >= total + 4 or W >= small + 4: rows = None if W >= total + 4: rows = [[chx for ch in word for chx in LETTERS[ch][r]] for r in range(6)] width = total else: rows = compact_logo(word) width = small pad = " " * ((W - width) // 2) for line_cells in rows: line, x = "", 0 for chx in line_cells: if chx == " ": line += " " else: line += fg(lerp(PAL["accent"], PAL["accent2"], x / max(1, width - 1))) + chx x += 1 println(pad + line + RESET) else: println(" " + c("accent", "FlynCode", bold=True)) sub = "AI Coding Agent in Terminal · v" + VERSION println() println(" " * max(0, (W - vlen(sub)) // 2) + c("muted", sub)) println() # ════════════════════════════════════════════════════════════════════════════ # СТАТУСЫ, ЗВУКИ, ПРЕДУПРЕЖДЕНИЯ О ЗАВИСАНИИ # ════════════════════════════════════════════════════════════════════════════ STATE_TEXT = {"wait": "Жду ответ", "think": "Думаю", "write": "Пишу ответ", "tool": "Выполняю", "verify": "Проверяю ответ", "perm": "Жду твоего разрешения", "compact": "Сжимаю память"} STATUS = {"state": "wait", "detail": "", "since": 0.0, "io": 0.0, "bytes": 0, "last": None} SOUND = {"on": True} NOTIFY = {"on": True} # всплывающее уведомление «ответ готов» / «ошибка» поверх окон def begin_status(state="wait"): """Начало шага запроса: сбрасываем таймер, счётчик байт и отметку активности.""" now = time.time() STATUS.update(state=state, detail="", since=now, io=now, bytes=0) def set_status(state, detail=""): STATUS["state"], STATUS["detail"] = state, detail def touch(nbytes=0): """Вызывается на каждом куске, пришедшем от сети: видно, что соединение живо.""" STATUS["io"] = time.time() STATUS["bytes"] += nbytes def set_result(kind, text): """Итог последнего хода: done / error / stopped. Показывается в строке статуса.""" STATUS["last"] = {"kind": kind, "text": text} def status_label(): now = time.time() st = STATUS base = STATE_TEXT.get(st["state"], "Работаю") if st["detail"]: base += " " + st["detail"] parts = [base + "…", fmt_dur(now - st["since"])] if st["bytes"]: parts.append("получено " + fmt_bytes(st["bytes"])) idle = now - st["io"] if idle >= 60: parts.append("⚠ нет данных %s — возможно, соединение зависло. Esc — отмена" % fmt_dur(idle)) return " · ".join(parts), "warn" if idle >= 15: parts.append("нет данных %s — ещё жду · Esc — отмена" % fmt_dur(idle)) return " · ".join(parts), "warn" parts.append("Esc — отмена") return " · ".join(parts), "muted" _tl = threading.local() # у каждого потока свой буфер вывода (для параллельных агентов) ACTIVE_ALL = set() # все открытые соединения: Esc обрывает сразу несколько агентов class _Router: """Подменяет sys.stdout: вывод потоков-агентов уходит в буфер, главного потока — в терминал.""" def __init__(self, real): self.real = real def write(self, s): buf = getattr(_tl, "buf", None) if buf is not None: buf.append(s) else: self.real.write(s) def flush(self): if getattr(_tl, "buf", None) is None: self.real.flush() def __getattr__(self, name): return getattr(self.real, name) def start_watchdog(sp): """Если спиннер не крутится и данных нет 20 с — один раз пишет строку-предупреждение.""" stop = threading.Event() if getattr(_tl, "buf", None) is not None: # параллельный агент: предупреждения не печатаем return stop WATCHDOGS.add(stop) def run(): warned = 0.0 while not stop.wait(2): io = STATUS["io"] idle = time.time() - io if idle >= 20 and not sp.running and warned != io: warned = io out("\n " + c("warn", "⚠ нет данных %s — ещё жду. Esc — отмена" % fmt_dur(idle)) + "\n") threading.Thread(target=run, daemon=True).start() return stop def desktop_notify(kind): """Системное уведомление: Windows (балун), macOS, Linux (notify-send). В фоне, не блокирует интерфейс.""" title = "FlynCode" body = "ИИ закончило ответ ✓" if kind == "done" else "Ошибка ✗ — посмотри в терминале" def run(): dn = subprocess.DEVNULL try: if IS_WIN: ps = ("Add-Type -AssemblyName System.Windows.Forms,System.Drawing; " "$n=New-Object System.Windows.Forms.NotifyIcon; " "$n.Icon=[System.Drawing.SystemIcons]::Information; $n.Visible=$true; " "$n.ShowBalloonTip(4000,'%s','%s',[System.Windows.Forms.ToolTipIcon]::Info); " "Start-Sleep -Seconds 5; $n.Dispose()" % (title, body)) subprocess.run(["powershell", "-NoProfile", "-Command", ps], stdout=dn, stderr=dn, timeout=30) elif sys.platform == "darwin" and shutil.which("osascript"): script = 'display notification "%s" with title "%s"' % (body, title) subprocess.run(["osascript", "-e", script], stdout=dn, stderr=dn, timeout=10) elif shutil.which("notify-send"): subprocess.run(["notify-send", "-a", title, title, body], stdout=dn, stderr=dn, timeout=10) except Exception: pass threading.Thread(target=run, daemon=True).start() def play_sound(kind): """kind: 'done' — ответ готов, 'error' — ошибка. Звук и уведомление в фоне, интерфейс не ждёт.""" if NOTIFY["on"]: desktop_notify(kind) if SOUND["on"]: threading.Thread(target=_play_sound, args=(kind,), daemon=True).start() def _play_mp3(path): """Проигрывает mp3 внешним плеером. True — если нашёлся плеер.""" if sys.platform == "darwin" and shutil.which("afplay"): cmd = ["afplay", path] elif shutil.which("ffplay"): cmd = ["ffplay", "-nodisp", "-autoexit", "-loglevel", "quiet", path] elif shutil.which("mpg123"): cmd = ["mpg123", "-q", path] else: return False subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=30) return True def _play_sound(kind): # Сначала пробуем файлы 1.mp3 (успех) и 2.mp3 (ошибка) рядом со скриптом. # Если файла нет или плеера нет — встроенный системный звук, как раньше. base = os.path.dirname(os.path.abspath(__file__)) if "__file__" in globals() else os.getcwd() fname = "1.mp3" if kind == "done" else "2.mp3" path = os.path.join(base, fname) if os.path.isfile(path): try: if IS_WIN: import ctypes alias = "flyncode_snd" opened = ctypes.windll.winmm.mciSendStringW( 'open "%s" type mpegvideo alias %s' % (path, alias), None, 0, None) if opened == 0: ctypes.windll.winmm.mciSendStringW("play %s wait" % alias, None, 0, None) ctypes.windll.winmm.mciSendStringW("close %s" % alias, None, 0, None) return elif _play_mp3(path): return except Exception: pass try: if IS_WIN: import winsound winsound.MessageBeep(winsound.MB_ICONASTERISK if kind == "done" else winsound.MB_ICONHAND) return if sys.platform == "darwin": name = "Glass" if kind == "done" else "Basso" subprocess.run(["afplay", "/System/Library/Sounds/%s.aiff" % name], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=10) return path2 = "/usr/share/sounds/freedesktop/stereo/%s" % ("complete.oga" if kind == "done" else "dialog-error.oga") for player in ("paplay", "pw-play"): if shutil.which(player) and os.path.exists(path2): subprocess.run([player, path2], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=10) return n = 1 if kind == "done" else 3 for i in range(n): out("\a") sys.stdout.flush() if i < n - 1: time.sleep(0.3) except Exception: pass SPINNERS = set() # все работающие спиннеры: при отмене и выходе из задачи гасим их разом WATCHDOGS = set() # сторожа «нет данных»: тоже гасим разом def stop_all_spinners(): for ev in list(WATCHDOGS): ev.set() WATCHDOGS.clear() for sp in list(SPINNERS): try: sp.stop() except Exception: pass class Spinner: FR = "|/-\\" def __init__(self, label_fn): self.label_fn = label_fn self.ev = threading.Event() self.t = None self.running = False def start(self): if self.running or getattr(_tl, "buf", None) is not None: return self.running = True SPINNERS.add(self) self.ev.clear() out("\x1b[?25l") self.t = threading.Thread(target=self._run, daemon=True) self.t.start() def _run(self): i = 0 while not self.ev.wait(0.08): lab = self.label_fn() text, col = lab if isinstance(lab, tuple) else (lab, "muted") text = clip(text, term_w() - 8) SCREEN.set_dyn([" " + c("accent", self.FR[i % len(self.FR)]) + " " + c(col, text)]) sys.stdout.flush() i += 1 def stop(self): if not self.running: return self.running = False SPINNERS.discard(self) self.ev.set() if self.t: self.t.join() SCREEN.set_dyn([]) def with_spinner(label, fn): t0 = time.time() sp = Spinner(lambda: "%s %s" % (label, fmt_dur(time.time() - t0))) sp.start() try: return fn() finally: sp.stop() # ════════════════════════════════════════════════════════════════════════════ # СТРОКА ВВОДА И МЕНЮ ВЫБОРА # ════════════════════════════════════════════════════════════════════════════ COMMANDS = [ ("/help", "список команд"), ("/model", "выбрать модель (с поиском)"), ("/provider", "сменить провайдера"), ("/think", "уровень размышлений"), ("/perm", "разрешения инструментов"), ("/cd", "сменить рабочую папку"), ("/settings", "настройки"), ("/usage", "статистика токенов"), ("/models", "обновить список моделей"), ("/agents", "агенты: Код, Поиск, Чат…"), ("/debate", "дискуссия агентов с итогом судьи"), ("/skills", "скиллы: @имя задача"), ("/key", "сменить API-ключ"), ("/clear", "новый чат"), ("/exit", "выход"), ("/tips", "как работать с агентами и скиллами"), ("/compact", "сжать память длинного чата"), ("/verify", "самопроверка ответа: вкл / выкл"), ("/runtime", "среды запуска: Bun, Node, Python, git"), ("/history", "история чатов: открыть, продолжить, удалить"), ("/chats", "список чатов: открыть или удалить"), ("/export", "экспорт чата в Markdown"), ("/mode", "режим: Build / Plan / Chat"), ("/plan", "план чата: показать или очистить"), ("/sidebar", "правая панель: вкл / выкл"), ("/attach", "прикрепить фото к следующему сообщению"), ("/paste", "вставить картинку из буфера"), ("/tray", "спрятать окно в трей (Windows, обычная консоль)"), ] HIST_DIR = os.path.join(os.path.expanduser("~"), ".flyncode", "chats") def _hist_path(chat_id): return os.path.join(HIST_DIR, chat_id + ".json") def _new_id(): import uuid return uuid.uuid4().hex[:8] SAVE_ERROR = {"msg": "", "shown": False} def save_chat(cid, messages, cwd, plan=None): """Сохраняет чат на диск атомарно: сначала во временный файл, потом переименование, чтобы сбой записи не портил прошлую версию. Картинки (base64) не пишем. True — если записалось.""" try: os.makedirs(HIST_DIR, exist_ok=True) clean = [{k: v for k, v in m.items() if k != "images"} for m in messages] data = {"id": cid, "messages": clean, "cwd": cwd, "ts": time.time(), "plan": plan or []} path = _hist_path(cid) tmp = path + ".tmp" with open(tmp, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=1, default=str) os.replace(tmp, path) SAVE_ERROR["shown"] = False return True except Exception as e: SAVE_ERROR["msg"] = str(e)[:200] return False def list_chats(): if not os.path.isdir(HIST_DIR): return [] res = [] for fn in os.listdir(HIST_DIR): if fn.endswith(".json"): cid = fn[:-5] p = os.path.join(HIST_DIR, fn) try: with open(p, encoding="utf-8") as f: d = json.load(f) msgs = d.get("messages", []) if not msgs: continue first = "" for m in msgs: if m.get("role") == "user" and not m.get("summary"): first = m.get("content", "")[:60].replace("\n", " ") break res.append({"id": cid, "ts": d.get("ts", 0), "cwd": d.get("cwd", ""), "n": len(msgs), "first": first, "path": p}) except Exception: pass return sorted(res, key=lambda x: -x["ts"]) def load_chat(cid): try: with open(_hist_path(cid), encoding="utf-8") as f: return json.load(f).get("messages", []) except Exception: return [] def read_chat_file(cid): """Полные данные чата (dict) или None, если файла нет или он не читается.""" try: with open(_hist_path(cid), encoding="utf-8") as f: d = json.load(f) return d if isinstance(d, dict) else None except Exception: return None def load_plan(cid): """План чата: список {text, done}. Лежит в том же файле, что и сам чат.""" try: with open(_hist_path(cid), encoding="utf-8") as f: d = json.load(f) return [t for t in d.get("plan", []) if isinstance(t, dict) and t.get("text")] except Exception: return [] def layout(pw, text, maxw): rows, pos, r, col = [""], [], 0, pw for ch in text: w = cw(ch) if col + w > maxw: rows.append("") r += 1 col = pw pos.append((r, col)) rows[r] += ch col += w pos.append((r, col)) return rows, pos def read_line(prefix=" ", mask=False, history=None, commands=False, placeholder="", initial="", esc_cancel=False, footer=None, on_tab=None, on_paste=None): """Возвращает str; None — Esc/Ctrl+C при esc_cancel; EOFError — выход. Строка ввода выглядит как карточка OpenCode: акцентный «┃» слева и тёмная подложка.""" buf = list(initial) pos = len(buf) hi = len(history) if history is not None else 0 st = {"sel": 0, "rows": 0, "armed": False} pw = vlen_ansi(prefix() if callable(prefix) else prefix) def sugg(): t = "".join(buf) if commands and t.startswith("/") and " " not in t: return [x for x in COMMANDS if x[0].startswith(t.lower())] return [] def render(): prefix_now = prefix() if callable(prefix) else prefix maxw = term_w() - 2 text = "".join("•" if mask else ch for ch in buf) rows, ps = layout(pw, text, maxw) if not buf and placeholder: ph = "Ctrl+C ещё раз — выход" if st["armed"] else placeholder lines = [box_line(prefix_now + c("muted", clip(ph, maxw - pw)))] cr, cc = 0, pw else: lines = [box_line((prefix_now if i == 0 else " " * pw) + r) for i, r in enumerate(rows)] cr, cc = ps[pos] if footer: lines = ([box_line(prefix_now)] + lines + [box_line(prefix_now)] + [box_line(prefix_now + fl) for fl in footer()]) cr += 1 sg = sugg() if sg: st["sel"] = min(st["sel"], len(sg) - 1) lines.append("") for i, (nm, ds) in enumerate(sg[:8]): if i == st["sel"]: lines.append(" " + bg(PAL["accent"]) + fg((10, 10, 10)) + padv(" " + padv(nm, 12) + ds, min(maxw - 2, 70)) + RESET) else: lines.append(" " + c("text", padv(nm, 12)) + c("muted", clip(ds, maxw - 16))) SCREEN.set_dyn(lines, (cr, cc)) def finish(): SCREEN.set_dyn([]) SCREEN.scroll = 0 st["rows"] = 0 with RawMode(): render() while True: k = read_key() sg = sugg() if k == "enter": t = "".join(buf) if sg and t not in dict(COMMANDS): t = sg[min(st["sel"], len(sg) - 1)][0] finish() if history is not None and t.strip() and (not history or history[-1] != t): history.append(t) return t if k == "ctrl-c": if esc_cancel: finish() return None if buf: buf.clear() pos = 0 elif st["armed"]: finish() raise EOFError else: st["armed"] = True elif k == "ctrl-d": if not buf: finish() raise EOFError elif k == "esc": if esc_cancel: finish() return None elif k in ("ctrl-v", "ctrl-g") and on_paste: for ch in (on_paste() or ""): buf.insert(pos, ch) pos += 1 st["sel"] = 0 elif k == "ctrl-n" and commands: finish() return "/clear" elif k == "ctrl-p" and commands: # палитра команд: открываем список всех команд через «/» buf[:] = list("/") pos = 1 st["sel"] = 0 elif k in ("pgup", "pgdn"): SCREEN.scroll_by(8 if k == "pgup" else -8) elif k == "backspace": if pos: del buf[pos - 1] pos -= 1 elif k == "delete": if pos < len(buf): del buf[pos] elif k == "left": pos = max(0, pos - 1) elif k == "right": pos = min(len(buf), pos + 1) elif k in ("home", "ctrl-a"): pos = 0 elif k in ("end", "ctrl-e"): pos = len(buf) elif k == "ctrl-u": del buf[:pos] pos = 0 elif k == "ctrl-w": j = pos while j and buf[j - 1] == " ": j -= 1 while j and buf[j - 1] != " ": j -= 1 del buf[j:pos] pos = j elif k == "tab" and sg: buf[:] = list(sg[min(st["sel"], len(sg) - 1)][0] + " ") pos = len(buf) elif k == "tab" and on_tab: on_tab() render() # цвет линии и название меняются сразу, без ожидания клавиши elif k == "up": if sg: st["sel"] = (st["sel"] - 1) % min(len(sg), 8) elif history and hi > 0: hi -= 1 buf[:] = list(history[hi]) pos = len(buf) elif k == "down": if sg: st["sel"] = (st["sel"] + 1) % min(len(sg), 8) elif history is not None and hi < len(history): hi += 1 buf[:] = list(history[hi]) if hi < len(history) else [] pos = len(buf) elif len(k) == 1 and k.isprintable(): buf.insert(pos, k) pos += 1 st["sel"] = 0 st["armed"] = False if not key_pending(): render() def select(title, items, current=None, searchable=True, hint="", custom=False): """items: [{'label','desc','value'}]. Возвращает value или None (Esc).""" q = [] for it in items: it["_s"] = (it["label"] + " " + it.get("desc", "") + " " + str(it["value"])).lower() def filt(): terms = "".join(q).lower().split() res = [it for it in items if all(t in it["_s"] for t in terms)] qs = "".join(q).strip() if custom and qs and not any(it["value"] == qs for it in res): res.append({"label": "Использовать «%s»" % qs, "desc": "ввести ID вручную", "value": qs, "_s": ""}) return res view = filt() sel = next((i for i, it in enumerate(view) if current is not None and it["value"] == current), 0) top = 0 vis = max(3, min(14, term_h() - 7)) height = vis + 3 first = [True] def draw(): nonlocal top W = term_w() if sel < top: top = sel if sel >= top + vis: top = sel - vis + 1 lines = [" " + c("accent", "◆ ") + c("text", clip(title, W - 6), bold=True)] if searchable: lines.append(" " + c("muted", "› ") + "".join(q) + c("accent", "|") + ("" if q else c("dim", " печатай для поиска…"))) else: lines.append(" " + c("muted", clip(hint, W - 6))) lw = min(max([vlen(it["label"]) for it in view] + [10]), max(20, W // 2)) dw = max(0, W - lw - 14) for i in range(vis): idx = top + i if idx >= len(view): lines.append(" " + c("muted", "ничего не найдено") if (not view and i == 0) else "") continue it = view[idx] mark = "● " if current is not None and it["value"] == current else " " label = padv(clip(it["label"], lw), lw) desc = clip(it.get("desc", ""), dw) if idx == sel: lines.append(" " + bg(PAL["accent"]) + fg((10, 10, 10)) + " " + padv(mark + label + " " + desc, W - 8) + " " + RESET) else: lines.append(" " + c("accent", mark) + c("text", label) + " " + c("muted", desc)) foot = "↑↓ выбор · Enter ок · Esc отмена %d/%d" % (min(sel + 1, len(view)), len(view)) lines.append(" " + c("dim", clip(foot, W - 4))) SCREEN.set_dyn(lines, None) first[0] = False def done(): SCREEN.set_dyn([]) out("\x1b[?25l") try: with RawMode(): draw() while True: k = read_key() if k == "enter" and view: done() return view[sel]["value"] if k in ("esc", "ctrl-c"): done() return None if k == "up": sel = (sel - 1) % len(view) if view else 0 elif k == "down": sel = (sel + 1) % len(view) if view else 0 elif k == "pgup": sel = max(0, sel - vis) elif k == "pgdn": sel = min(max(0, len(view) - 1), sel + vis) elif searchable and k == "backspace" and q: q.pop() view = filt() sel = 0 elif searchable and len(k) == 1 and k.isprintable(): q.append(k) view = filt() sel = 0 if not key_pending(): draw() except BaseException: SCREEN.set_dyn([]) raise # ════════════════════════════════════════════════════════════════════════════ # ПРОВАЙДЕРЫ И МОДЕЛИ # ════════════════════════════════════════════════════════════════════════════ PROVIDERS = [ dict(id="openai", name="ChatGPT (OpenAI)", kind="openai", base="https://api.openai.com/v1", env="OPENAI_API_KEY", url="https://platform.openai.com/api-keys", desc="GPT-5, GPT-4.1, o-серия"), dict(id="anthropic", name="Claude (Anthropic)", kind="anthropic", base="https://api.anthropic.com/v1", env="ANTHROPIC_API_KEY", url="https://console.anthropic.com/settings/keys", desc="Opus, Sonnet, Haiku"), dict(id="gemini", name="Gemini (Google)", kind="openai", base="https://generativelanguage.googleapis.com/v1beta/openai", env="GEMINI_API_KEY", url="https://aistudio.google.com/apikey", desc="Gemini 2.5 / 3, есть бесплатный тариф"), dict(id="openrouter", name="OpenRouter", kind="openai", base="https://openrouter.ai/api/v1", env="OPENROUTER_API_KEY", url="https://openrouter.ai/keys", desc="сотни моделей одним ключом"), dict(id="deepseek", name="DeepSeek", kind="openai", base="https://api.deepseek.com/v1", env="DEEPSEEK_API_KEY", url="https://platform.deepseek.com/api_keys", desc="V3, R1 — очень дёшево"), dict(id="groq", name="Groq", kind="openai", base="https://api.groq.com/openai/v1", env="GROQ_API_KEY", url="https://console.groq.com/keys", desc="сверхбыстрые open-source модели"), dict(id="xai", name="xAI (Grok)", kind="openai", base="https://api.x.ai/v1", env="XAI_API_KEY", url="https://console.x.ai", desc="Grok 3 / 4"), dict(id="mistral", name="Mistral", kind="openai", base="https://api.mistral.ai/v1", env="MISTRAL_API_KEY", url="https://console.mistral.ai/api-keys", desc="Mistral Large, Codestral", no_stream_usage=True), dict(id="together", name="Together AI", kind="openai", base="https://api.together.xyz/v1", env="TOGETHER_API_KEY", url="https://api.together.ai/settings/api-keys", desc="Llama, Qwen, DeepSeek и др."), dict(id="ollama", name="Ollama (локально)", kind="openai", base="http://localhost:11434/v1", env="", url="", desc="на твоём компьютере, ключ не нужен", nokey=True), dict(id="cerebras", name="Cerebras Cloud", kind="openai", base="https://api.cerebras.ai/v1", env="CEREBRAS_API_KEY", url="https://cloud.cerebras.ai/platform", desc="очень быстрые модели"), dict(id="fireworks", name="Fireworks AI", kind="openai", base="https://api.fireworks.ai/inference/v1", env="FIREWORKS_API_KEY", url="https://fireworks.ai/account/api-keys", desc="open-source модели, быстро"), dict(id="deepinfra", name="DeepInfra", kind="openai", base="https://api.deepinfra.com/v1/openai", env="DEEPINFRA_API_KEY", url="https://deepinfra.com/dash/api_keys", desc="open-source модели, дёшево"), dict(id="huggingface", name="Hugging Face Inference", kind="openai", base="https://router.huggingface.co/v1", env="HF_TOKEN", url="https://huggingface.co/settings/tokens", desc="тысячи open-source моделей"), dict(id="cohere", name="Cohere", kind="openai", base="https://api.cohere.ai/compatibility/v1", env="COHERE_API_KEY", url="https://dashboard.cohere.com/api-keys", desc="Command и Command R"), dict(id="sambanova", name="SambaNova Cloud", kind="openai", base="https://api.sambanova.ai/v1", env="SAMBANOVA_API_KEY", url="https://cloud.sambanova.ai/apis", desc="быстрые open-source модели"), dict(id="nebius", name="Nebius AI Studio", kind="openai", base="https://api.studio.nebius.com/v1", env="NEBIUS_API_KEY", url="https://studio.nebius.com/", desc="open-source модели"), dict(id="novita", name="Novita AI", kind="openai", base="https://api.novita.ai/v3/openai", env="NOVITA_API_KEY", url="https://novita.ai/settings/key-management", desc="open-source модели"), dict(id="glhf", name="Glhf.ai", kind="openai", base="https://glhf.chat/api/openai/v1", env="GLHF_API_KEY", url="https://glhf.chat/users/settings/api", desc="модели из Hugging Face"), dict(id="aimlapi", name="AI/ML API (Aim ML)", kind="openai", base="https://api.aimlapi.com/v1", env="AIMLAPI_API_KEY", url="https://aimlapi.com/app/keys", desc="сотни моделей одним ключом"), dict(id="lambda", name="Lambda", kind="openai", base="https://api.lambda.ai/v1", env="LAMBDA_API_KEY", url="https://cloud.lambda.ai/api-keys", desc="Llama и другие open-source"), dict(id="baseten", name="Baseten", kind="openai", base="https://inference.baseten.co/v1", env="BASETEN_API_KEY", url="https://app.baseten.co/settings/api_keys", desc="model APIs"), dict(id="nvidia", name="NVIDIA NIM", kind="openai", base="https://integrate.api.nvidia.com/v1", env="NVIDIA_API_KEY", url="https://build.nvidia.com", desc="модели NVIDIA и open-source"), dict(id="hyperbolic", name="Hyperbolic", kind="openai", base="https://api.hyperbolic.xyz/v1", env="HYPERBOLIC_API_KEY", url="https://app.hyperbolic.xyz/settings", desc="open-source модели, дёшево"), dict(id="featherless", name="Featherless AI", kind="openai", base="https://api.featherless.ai/v1", env="FEATHERLESS_API_KEY", url="https://featherless.ai/account/api-keys", desc="тысячи open-source моделей"), dict(id="chutes", name="Chutes", kind="openai", base="https://llm.chutes.ai/v1", env="CHUTES_API_KEY", url="https://chutes.ai", desc="open-source модели"), dict(id="friendli", name="Friendli", kind="openai", base="https://api.friendli.ai/serverless/v1", env="FRIENDLI_TOKEN", url="https://friendli.ai/suite/settings/tokens", desc="serverless open-source"), dict(id="scaleway", name="Scaleway Generative APIs", kind="openai", base="https://api.scaleway.ai/v1", env="SCW_SECRET_KEY", url="https://console.scaleway.com/iam/api-keys", desc="европейский провайдер"), dict(id="ovh", name="OVHcloud AI Endpoints", kind="openai", base="https://oai.endpoints.kepler.ai.cloud.ovh.net/v1", env="OVH_AI_ENDPOINTS_ACCESS_TOKEN", url="https://endpoints.ai.cloud.ovh.net", desc="европейский провайдер"), dict(id="perplexity", name="Perplexity Sonar", kind="openai", base="https://api.perplexity.ai", env="PERPLEXITY_API_KEY", url="https://www.perplexity.ai/settings/api", desc="модели с поиском в интернете"), dict(id="ai21", name="AI21 Labs", kind="openai", base="https://api.ai21.com/studio/v1", env="AI21_API_KEY", url="https://studio.ai21.com/account/api-key", desc="Jamba"), dict(id="writer", name="Writer", kind="openai", base="https://api.writer.com/v1", env="WRITER_API_KEY", url="https://app.writer.com/", desc="Palmyra, корпоративные модели"), dict(id="clarifai", name="Clarifai", kind="openai", base="https://api.clarifai.com/v2/ext/openai/v1", env="CLARIFAI_PAT", url="https://clarifai.com/settings/security", desc="модели Clarifai и партнёров"), dict(id="predictionguard", name="Prediction Guard", kind="openai", base="https://api.predictionguard.com", env="PREDICTIONGUARD_API_KEY", url="https://predictionguard.com", desc="адрес проверь в документации"), dict(id="moonshot", name="Moonshot (Kimi)", kind="openai", base="https://api.moonshot.ai/v1", env="MOONSHOT_API_KEY", url="https://platform.moonshot.ai/console/api-keys", desc="Kimi"), dict(id="zhipu", name="Zhipu (GLM)", kind="openai", base="https://open.bigmodel.cn/api/paas/v4", env="ZHIPUAI_API_KEY", url="https://open.bigmodel.cn/usercenter/apikeys", desc="GLM-4.5 и другие"), dict(id="minimax", name="MiniMax", kind="openai", base="https://api.minimax.io/v1", env="MINIMAX_API_KEY", url="https://platform.minimax.io", desc="MiniMax-M2 и другие"), dict(id="siliconflow", name="SiliconFlow", kind="openai", base="https://api.siliconflow.cn/v1", env="SILICONFLOW_API_KEY", url="https://cloud.siliconflow.cn/account/ak", desc="Qwen, DeepSeek и другие"), dict(id="qianfan", name="Baidu Qianfan", kind="openai", base="https://qianfan.baidubce.com/v2", env="QIANFAN_API_KEY", url="https://console.bce.baidu.com/qianfan/ais/console/apiKey", desc="ERNIE и другие"), dict(id="dashscope", name="Alibaba DashScope (Qwen)", kind="openai", base="https://dashscope-intl.aliyuncs.com/compatible-mode/v1", env="DASHSCOPE_API_KEY", url="https://bailian.console.alibabacloud.com/", desc="Qwen"), dict(id="hunyuan", name="Tencent Hunyuan", kind="openai", base="https://api.hunyuan.cloud.tencent.com/v1", env="HUNYUAN_API_KEY", url="https://console.cloud.tencent.com/hunyuan/api-key", desc="Hunyuan"), dict(id="spark", name="iFlytek Spark", kind="openai", base="https://spark-api-open.xf-yun.com/v1", env="SPARK_API_KEY", url="https://console.xfyun.cn", desc="ключ вида APIKey:APISecret"), dict(id="yandex", name="YandexGPT (Yandex Cloud)", kind="openai", base="https://llm.api.cloud.yandex.net/v1", env="YANDEX_API_KEY", url="https://console.yandex.cloud", auth="Api-Key", desc="модель вида gpt://<папка>/yandexgpt/latest"), dict(id="blackbox", name="Blackbox AI", kind="openai", base="https://api.blackbox.ai/v1", env="BLACKBOX_API_KEY", url="https://www.blackbox.ai", desc="адрес проверь в документации"), dict(id="runpod", name="RunPod (свой эндпоинт)", kind="openai", base="https://api.runpod.ai/v2/ENDPOINT_ID/openai/v1", env="RUNPOD_API_KEY", url="https://www.runpod.io/console/user/settings", desc="впиши свой ENDPOINT_ID в адрес", ask_base=True), dict(id="cloudflare", name="Cloudflare Workers AI", kind="openai", base="https://api.cloudflare.com/client/v4/accounts/ACCOUNT_ID/ai/v1", env="CLOUDFLARE_API_KEY", url="https://dash.cloudflare.com/profile/api-tokens", desc="впиши свой ACCOUNT_ID в адрес", ask_base=True), dict(id="modal", name="Modal (свой эндпоинт)", kind="openai", base="", env="", url="https://modal.com/docs", desc="адрес и ключ из твоего приложения", ask_base=True), dict(id="orcarouter", name="OrcaRouter", kind="openai", base="https://api.orcarouter.ai/v1", env="ORCAROUTER_API_KEY", url="https://www.orcarouter.ai/console/token", desc="200+ моделей одним ключом (sk-orca-…)"), dict(id="cleanapis", name="Cleanapis", kind="openai", base="", env="CLEANAPIS_API_KEY", url="", desc="адрес введи из их документации", ask_base=True), dict(id="freeinference", name="Freeinference", kind="openai", base="", env="FREEINFERENCE_API_KEY", url="", desc="адрес введи из их документации", ask_base=True), dict(id="tokenharbor", name="Tokenharbor", kind="openai", base="https://tokenharbor.ai/v1", env="TOKENHARBOR_API_KEY", url="https://tokenharbor.ai", desc="фронтир-модели одним ключом, OpenAI-совместимый"), dict(id="agentrouter", name="AgentRouter", kind="openai", base="", env="AGENTROUTER_API_KEY", url="", desc="адрес введи из их документации", ask_base=True), dict(id="anyscale", name="Anyscale Endpoints", kind="openai", base="", env="ANYSCALE_API_KEY", url="", desc="сервис мог закрыться, адрес введи сам", ask_base=True), dict(id="monsterapi", name="MonsterAPI", kind="openai", base="", env="MONSTER_API_KEY", url="", desc="адрес введи из их документации", ask_base=True), dict(id="agnes", name="Agnes AI", kind="openai", base="https://apihub.agnes-ai.com/v1", env="AGNES_API_KEY", url="https://platform.agnes-ai.com", desc="мультимодальные модели, OpenAI-совместимый"), dict(id="custom", name="Другой (OpenAI-совместимый)", kind="openai", base="", env="", url="", desc="LM Studio, vLLM, прокси…", optkey=True, ask_base=True), ] # ★ Рекомендуем: эти провайдеры показываются первыми в списке RECOMMENDED = ["agnes", "orcarouter", "tokenharbor", "nvidia", "dashscope", "openrouter"] PROVIDER_NOTES = [] # ошибки в providers.json — покажем при запуске USER_PROVIDERS_FILE = os.path.join(CFG_DIR, "providers.json") def load_user_providers(): """Свои провайдеры из ~/.flyncode/providers.json: любой OpenAI-совместимый сервис. Формат: [{"id": "...", "name": "...", "base": "https://.../v1", "env": "MY_KEY", "url": "страница ключа", "desc": "описание", "auth": "Bearer", "ask_base": false}]""" if not os.path.isfile(USER_PROVIDERS_FILE): return [] try: with open(USER_PROVIDERS_FILE, encoding="utf-8") as f: data = json.load(f) except (OSError, ValueError) as e: PROVIDER_NOTES.append("providers.json не читается: %s" % e) return [] if isinstance(data, dict): data = data.get("providers", []) if not isinstance(data, list): PROVIDER_NOTES.append("providers.json: нужен список провайдеров") return [] known = {p["id"] for p in PROVIDERS} out = [] for i, it in enumerate(data, 1): if not isinstance(it, dict) or not it.get("id") or not it.get("name"): PROVIDER_NOTES.append("providers.json: запись %d без id или name — пропущена" % i) continue pid = re.sub(r"[^a-z0-9_-]", "", str(it["id"]).lower()) if not pid or pid in known: PROVIDER_NOTES.append("providers.json: id «%s» пустой или уже занят — пропущен" % it["id"]) continue base = str(it.get("base") or "").strip().rstrip("/") if not base and not it.get("ask_base"): PROVIDER_NOTES.append("providers.json: у «%s» не указан base — пропущен" % it["name"]) continue known.add(pid) p = {"id": pid, "name": str(it["name"]), "kind": "openai", "base": base, "env": str(it.get("env") or ""), "url": str(it.get("url") or ""), "desc": str(it.get("desc") or "свой провайдер")} if it.get("auth"): p["auth"] = str(it["auth"]) if it.get("ask_base"): p["ask_base"] = True if it.get("nokey"): p["nokey"] = True out.append(p) return out PROVIDERS.extend(load_user_providers()) FALLBACK = { "openai": ["gpt-5", "gpt-5-mini", "gpt-4.1", "gpt-4o", "o4-mini"], "anthropic": ["claude-sonnet-4-5", "claude-opus-4-1", "claude-haiku-4-5"], "gemini": ["gemini-2.5-pro", "gemini-2.5-flash"], "openrouter": ["openai/gpt-5", "anthropic/claude-sonnet-4.5", "deepseek/deepseek-r1"], "deepseek": ["deepseek-chat", "deepseek-reasoner"], "groq": ["openai/gpt-oss-120b", "llama-3.3-70b-versatile"], "xai": ["grok-4", "grok-3-mini"], "mistral": ["mistral-large-latest", "codestral-latest"], "together": ["meta-llama/Llama-3.3-70B-Instruct-Turbo"], "ollama": ["llama3.2", "qwen2.5-coder"], "custom": [], "cerebras": ["llama3.1-8b", "llama-3.3-70b", "qwen-3-32b"], "fireworks": ["accounts/fireworks/models/llama-v3p3-70b-instruct"], "deepinfra": ["meta-llama/Meta-Llama-3.1-70B-Instruct"], "sambanova": ["Meta-Llama-3.3-70B-Instruct"], "perplexity": ["sonar", "sonar-pro", "sonar-reasoning"], "moonshot": ["kimi-k2-0711-preview"], "zhipu": ["glm-4.5"], "siliconflow": ["Qwen/Qwen2.5-72B-Instruct"], "nvidia": ["meta/llama-3.3-70b-instruct"], "hyperbolic": ["meta-llama/Llama-3.3-70B-Instruct"], "cohere": ["command-a-03-2025"], "orcarouter": ["orcarouter/auto"], "agnes": ["agnes-2.0-flash", "agnes-1.5-flash"], } EXCLUDE = ["embed", "whisper", "tts", "dall-e", "moderation", "guard", "rerank", "transcribe", "imagen", "davinci", "babbage", "realtime", "audio", "-image", "veo", "aqa", "native-audio", "computer-use", "deep-research", "codex", "instruct", "-live", "robotics", "learnlm"] LEVEL_DESC = {"auto": "как решит модель", "off": "без размышлений", "minimal": "минимум", "low": "быстро", "medium": "баланс скорости и качества", "high": "глубоко подумать"} ANTH_BUDGET = {"low": 2048, "medium": 8192, "high": 16000} def levels_for(pid, mid, think=False): m = mid.lower() if pid == "openai" and (m.startswith("gpt-5") or re.match(r"^o\d", m)): return ["auto", "minimal", "low", "medium", "high"] if m.startswith("gpt-5") else ["auto", "low", "medium", "high"] if pid == "anthropic" and re.search(r"claude-3-7|claude-(opus|sonnet|haiku)-4|claude-[a-z]+-[45]", m): return ["off", "low", "medium", "high"] if pid == "gemini" and ("2.5" in m or "gemini-3" in m): return ["auto", "low", "medium", "high"] if pid == "openrouter" and think: return ["auto", "minimal", "low", "medium", "high"] if pid == "groq" and "gpt-oss" in m: return ["auto", "low", "medium", "high"] if pid == "custom": return ["auto", "low", "medium", "high"] return ["auto"] if pid != "anthropic" else ["off"] class ApiError(Exception): """Ошибка API. code — HTTP-статус (например 429), transient — стоит ли перепробовать.""" def __init__(self, message, code=None, transient=False): super().__init__(message) self.code = code self.transient = transient # ════════════════════════════════════════════════════════════════════════════ # ПОВТОРЫ ПРИ ОШИБКАХ (включая HTTP 429): до 10 попыток, ожидание 10..60 секунд # ════════════════════════════════════════════════════════════════════════════ MAX_RETRIES = 10 RETRY_MIN_S = 10 # минимальная пауза между попытками RETRY_MAX_S = 60 # максимальная пауза (1 минута) def _retry_delay(attempt): """Пауза перед следующей попыткой: экспоненциальный рост от 10с до 60с + небольшой джиттер.""" import random exp = RETRY_MIN_S * (2 ** (attempt - 1)) base = min(RETRY_MAX_S, exp) delay = base + (base * random.uniform(-0.2, 0.2)) return min(RETRY_MAX_S, max(RETRY_MIN_S, delay)) def _wait_retry(status_fn, attempt, total, err): """Ждёт перед повтором, печатает обратный отсчёт и плавно обновляет спиннер. Поддерживает отмену по Esc.""" delay = _retry_delay(attempt) if status_fn is not None: status_fn(attempt, total, err, delay) deadline = time.time() + delay while time.time() < deadline: if CANCEL.is_set(): raise KeyboardInterrupt("отменено пользователем") try: STATUS.update(detail="retry %d/%d · %s · %dс" % (attempt, total, err, max(0, int(deadline - time.time())))) except Exception: pass time.sleep(min(0.2, max(0.05, deadline - time.time()))) try: STATUS.update(detail="") except Exception: pass def _ctx(): try: return ssl.create_default_context() except Exception: return None def err_text(e): if isinstance(e, urllib.error.HTTPError): try: raw = e.read().decode("utf-8", "replace") try: j = json.loads(raw) err = j.get("error", j) if isinstance(j, dict) else j msg = err.get("message", json.dumps(err, ensure_ascii=False)) if isinstance(err, dict) else str(err) except ValueError: msg = raw[:300] except Exception: msg = str(e.reason) hint = {401: " — неверный ключ", 403: " — доступ запрещён", 404: " — модель или адрес не найдены", 429: " — лимит или кончились деньги"}.get(e.code, "") return "HTTP %d%s: %s" % (e.code, hint, str(msg)[:400]) if isinstance(e, urllib.error.URLError): r = str(e.reason) if "CERTIFICATE" in r.upper(): return "Ошибка SSL (macOS: запусти «Install Certificates.command» или pip install certifi)" return "Нет соединения: " + r return str(e) def friendly_error(e): """Короткое объяснение ошибки и что делать. Возвращает (что случилось, что сделать) или (None, None).""" code = getattr(e, "code", None) raw = str(e) low = raw.lower() if code == 401: return ("Неверный API-ключ", "Проверь ключ: /settings → провайдер, или /forget и введи заново") if code == 403: return ("Доступ запрещён", "У ключа нет прав на эту модель или регион. Выбери другую: /model") if code == 404: return ("Модель или адрес не найдены", "Выбери модель из списка: /model") if code == 429: return ("Лимит запросов или закончились деньги", "Подожди немного или проверь баланс у провайдера") if code is not None and code >= 500: return ("Провайдер сейчас не отвечает (ошибка на его стороне)", "Попробуй через минуту, чат сохранён") if "ssl" in low or "certificate" in low: return ("Проблема с сертификатом SSL", "На macOS запусти «Install Certificates.command» или поставь certifi") if "bun" in low: return ("Bun не установлен или не запускается", "Поставь Bun вручную с bun.sh или включи «Ставить Bun сам» в /settings") if isinstance(e, (socket.timeout, TimeoutError)) or "timed out" in low: return ("Провайдер не ответил вовремя", "Проверь интернет и попробуй ещё раз") if isinstance(e, urllib.error.URLError) or "нет соединения" in low or "getaddrinfo" in low: return ("Нет соединения с интернетом или провайдер недоступен", "Проверь интернет, VPN или прокси и попробуй ещё раз") return (None, None) def _is_transient(e): """Стоит ли перепробовать: HTTP 429, 5xx и сетевые сбои — да; 401/403/404 — нет.""" if isinstance(e, urllib.error.HTTPError): return e.code in (408, 425, 429) or 500 <= e.code < 600 if isinstance(e, urllib.error.URLError): if isinstance(e.reason, Exception) and "CERTIFICATE" in str(e.reason).upper(): return False return True return False def _api_err(e): """Заворачивает исключение в ApiError с кодом и флагом transient.""" txt = err_text(e) code = e.code if isinstance(e, urllib.error.HTTPError) else None return ApiError(txt, code=code, transient=_is_transient(e)) def _open_cancelable(req, timeout): """urlopen, который прерывается по Esc: запрос идёт в фоне, а ожидание смотрит на CANCEL.""" box = {} def go(): try: box["resp"] = urllib.request.urlopen(req, timeout=timeout, context=_ctx()) except BaseException as e: box["err"] = e t = threading.Thread(target=go, daemon=True) t.start() while t.is_alive(): if CANCEL.is_set(): def late_close(): t.join() r = box.get("resp") if r is not None: force_close(r) try: r.close() except Exception: pass threading.Thread(target=late_close, daemon=True).start() raise KeyboardInterrupt("отменено пользователем") t.join(0.1) if "err" in box: raise box["err"] return box["resp"] def http_open(url, headers=None, body=None, timeout=180, method=None): h = {"User-Agent": "FlynCode/" + VERSION} h.update(headers or {}) data = None if body is not None: data = json.dumps(body).encode("utf-8") h["Content-Type"] = "application/json" req = urllib.request.Request(url, data=data, headers=h, method=method or ("POST" if body is not None else "GET")) # до 10 попыток при transient-ошибках (включая 429): пауза 10..60с, Esc отменяет attempt = 0 while True: try: resp = _open_cancelable(req, timeout) ACTIVE_ALL.add(resp) return resp except Exception as e: attempt += 1 ae = _api_err(e) if not ae.transient or attempt > MAX_RETRIES: raise ApiError(str(ae), code=ae.code, transient=ae.transient) if CANCEL.is_set(): raise KeyboardInterrupt("отменено пользователем") try: STATUS.update(detail="retry %d/%d · %s" % (attempt, MAX_RETRIES, str(ae)[:120])) except Exception: pass _wait_retry(None, attempt, MAX_RETRIES, str(ae)[:120]) def http_json(url, headers=None): with http_open(url, headers, timeout=30) as r: try: return json.loads(r.read().decode("utf-8", "replace")) except ValueError: raise ApiError("Сервер вернул не JSON (проверь адрес API)") def sse(resp): try: for raw in resp: touch(len(raw)) line = raw.decode("utf-8", "replace").strip() if not line.startswith("data:"): continue d = line[5:].strip() if d == "[DONE]": return try: yield json.loads(d) except ValueError: continue except (OSError, ValueError) as e: raise ApiError("Соединение прервано: %s" % e) def _ts(v): if isinstance(v, (int, float)): return float(v) try: return datetime.fromisoformat(str(v).replace("Z", "+00:00")).timestamp() except Exception: return 0.0 def fetch_models(p, key, base): pid = p["id"] if p["kind"] == "anthropic": data = http_json(base + "/models?limit=1000", {"x-api-key": key, "anthropic-version": "2023-06-01"}) raw = data.get("data", []) else: h = {"Authorization": p.get("auth", "Bearer") + " " + key} if key else {} data = http_json(base.rstrip("/") + "/models", h) raw = data.get("data", []) if isinstance(data, dict) else [] models = [] for it in raw: mid = re.sub(r"^models/", "", it.get("id") or "") if not mid or (pid != "openrouter" and any(x in mid.lower() for x in EXCLUDE)): continue m = {"id": mid, "ctx": it.get("context_length") or it.get("inputTokenLimit") or 0, "ts": _ts(it.get("created_at") or it.get("created")), "think": False, "vision": False, "price": None} ml = mid.lower() desc_lower = it.get("description", "").lower() if any(kw in ml for kw in ("vision", "image", "multimodal", "-img")): m["vision"] = True elif "flash" in ml or ("gemini" in ml and "flash" in ml): m["vision"] = True elif "instruct" in ml and any(kw in desc_lower for kw in ("image", "vision", "multimodal")): m["vision"] = True elif pid == "custom": if any(kw in ml for kw in ("vision", "image", "multimodal", "flash", "pro")): m["vision"] = True if pid == "openrouter": arch = it.get("architecture") or {} outm = arch.get("output_modalities") or ["text"] if "text" not in outm or "image" in outm: continue m["think"] = "reasoning" in (it.get("supported_parameters") or []) if "image" in (arch.get("input_modalities") or []): m["vision"] = True pr = it.get("pricing") or {} try: m["price"] = (float(pr.get("prompt", 0)) * 1e6, float(pr.get("completion", 0)) * 1e6) except (TypeError, ValueError): pass elif pid == "anthropic": if "thinking" in ml or "op" in ml: m["think"] = True elif pid == "openai": if any(kw in ml for kw in ("o1", "o3", "reason")): m["think"] = True if "vision" in ml or "image" in ml: m["vision"] = True elif pid == "gemini": if "vision" in ml or "flash" in ml or "pro" in ml: m["vision"] = True if "reason" in ml or "flash" in ml: m["think"] = True elif pid == "deepseek": if "r1" in ml or "reasoner" in ml: m["think"] = True elif pid == "groq": if "reason" in ml or "r1" in ml: m["think"] = True m["levels"] = levels_for(pid, mid, m["think"]) models.append(m) if not models: raise ApiError("Список моделей пуст") models.sort(key=lambda m: (-m["ts"], m["id"])) return models def fallback_models(pid): return [{"id": i, "ctx": 0, "ts": 0.0, "think": False, "vision": False, "price": None, "levels": levels_for(pid, i)} for i in FALLBACK.get(pid, [])] # ════════════════════════════════════════════════════════════════════════════ # КОНФИГ # ════════════════════════════════════════════════════════════════════════════ DEFAULT_SETTINGS = { "show_thinking": True, "tools": True, "perms": {}, "theme": "flyn", "sound": True, "notify": True, "verify": True, "autocheck": True, "autocheck_rounds": 10, "vision": "auto", "big_logo": True, "compact_at": 60000, "bun_auto": True, "token_saving": "auto", # выкл / auto (с половины истории) / всегда — экономия токенов "system": "Ты FlynCode — точный и дружелюбный AI-агент в терминале. Отвечай на языке пользователя. " "Код оформляй в блоках ```.", } def load_cfg(): cfg = {"keys": {}, "bases": {}, "last": {}, "settings": {}} try: with open(CFG_FILE, "r", encoding="utf-8") as f: cfg.update(json.load(f)) except Exception: pass s = copy.deepcopy(DEFAULT_SETTINGS) s.update(cfg.get("settings") or {}) cfg["settings"] = s return cfg def save_cfg(cfg): try: os.makedirs(CFG_DIR, exist_ok=True) with open(CFG_FILE, "w", encoding="utf-8") as f: json.dump(cfg, f, ensure_ascii=False, indent=2) try: os.chmod(CFG_FILE, 0o600) except Exception: pass except Exception: pass def load_cache(pid): try: with open(os.path.join(CFG_DIR, "models_%s.json" % pid), "r", encoding="utf-8") as f: return json.load(f).get("models", []) except Exception: return [] def save_cache(pid, models): try: os.makedirs(CFG_DIR, exist_ok=True) with open(os.path.join(CFG_DIR, "models_%s.json" % pid), "w", encoding="utf-8") as f: json.dump({"models": models}, f, ensure_ascii=False) except Exception: pass # ════════════════════════════════════════════════════════════════════════════ # ИНСТРУМЕНТЫ АГЕНТА # ════════════════════════════════════════════════════════════════════════════ TOOL_DEFS = [ {"name": "read_file", "ask": False, "desc": "Прочитать текстовый файл с номерами строк. offset/limit — для больших файлов.", "props": {"path": "string", "offset": "integer", "limit": "integer"}, "req": ["path"]}, {"name": "list_files", "ask": False, "desc": "Найти файлы и папки по glob-шаблону, например **/*.py.", "props": {"pattern": "string", "path": "string"}, "req": ["pattern"]}, {"name": "grep", "ask": False, "desc": "Поиск регулярного выражения по содержимому файлов.", "props": {"pattern": "string", "path": "string", "include": "string"}, "req": ["pattern"]}, {"name": "write_file", "ask": True, "desc": "Создать файл или полностью перезаписать существующий. Для большого файла — только первая часть, остальное добавляй через append_file.", "props": {"path": "string", "content": "string"}, "req": ["path", "content"]}, {"name": "append_file", "ask": True, "desc": "Дописать текст в конец файла (или создать файл, если его нет). Для записи большого файла по частям.", "props": {"path": "string", "content": "string"}, "req": ["path", "content"]}, {"name": "edit_file", "ask": True, "desc": "Заменить один уникальный фрагмент текста в файле (old → new). Сначала прочитай файл.", "props": {"path": "string", "old": "string", "new": "string"}, "req": ["path", "old", "new"]}, {"name": "bash", "ask": True, "desc": "Выполнить shell-команду в рабочей папке: тесты, сборка, git, установка пакетов, запуск скриптов.", "props": {"command": "string", "timeout": "integer"}, "req": ["command"]}, {"name": "fetch_url", "ask": False, "desc": "Открыть веб-страницу по ссылке и получить её текст. Используй, когда просят открыть сайт, прочитать статью или проверить ссылку.", "props": {"url": "string"}, "req": ["url"]}, {"name": "download_file", "ask": True, "desc": "Скачать файл по ссылке на диск (нужно разрешение пользователя).", "props": {"url": "string", "save_to": "string"}, "req": ["url", "save_to"]}, {"name": "web_search", "ask": False, "desc": "Поиск в интернете: заголовки, ссылки и сниппеты. Используй для актуальных фактов и поиска нужных сайтов.", "props": {"query": "string"}, "req": ["query"]}, {"name": "load_skill", "ask": False, "desc": "Загрузить инструкции скилла (навыка) по имени. Список доступных скиллов дан в системном промпте.", "props": {"name": "string"}, "req": ["name"]}, {"name": "create_agent", "ask": True, "desc": "Создать помощника-агента с ролью (например, Код, Идейщик, Дизайнер) для части сложной задачи. Без model агент работает на текущей модели. Создавай, только если задача действительно делится на роли.", "props": {"name": "string", "role": "string", "model": "string"}, "req": ["name", "role"]}, {"name": "ask_agent", "ask": False, "desc": "Поручить задачу созданному агенту по имени и получить его ответ. Агенты работают по очереди: пока один работает, другой не запускается.", "props": {"name": "string", "task": "string"}, "req": ["name", "task"]}, {"name": "multi_edit", "ask": True, "desc": "Сделать несколько правок одного файла за один вызов (old → new, каждый фрагмент уникален). Либо применяются все, либо ни одна.", "props": {"path": "string", "edits": {"type": "array", "items": {"type": "object", "properties": { "old": {"type": "string"}, "new": {"type": "string"}}, "required": ["old", "new"]}}}, "req": ["path", "edits"]}, {"name": "apply_patch", "ask": True, "desc": "Применить unified diff (формат git diff) сразу к нескольким файлам. Контекст в патче должен совпадать с текущими файлами. Новый файл — заголовок --- /dev/null.", "props": {"patch": "string"}, "req": ["patch"]}, {"name": "move_path", "ask": True, "desc": "Переместить или переименовать файл или папку.", "props": {"src": "string", "dst": "string"}, "req": ["src", "dst"]}, {"name": "delete_path", "ask": True, "desc": "Удалить файл или папку. Папку с содержимым удаляет только при recursive=true. Необратимо.", "props": {"path": "string", "recursive": "boolean"}, "req": ["path"]}, {"name": "git", "ask": False, "desc": "Git только для чтения: status, diff, log, show, branch, blame, ls-files, rev-parse, describe, remote -v. Коммиты и ветки делай через bash.", "props": {"args": "string"}, "req": ["args"]}, {"name": "run_script", "ask": True, "desc": "Запустить файл-скрипт, среду выбирает сам: .py — Python, .js/.mjs — Bun или Node, .ts — Bun, .sh — bash.", "props": {"path": "string", "args": "string", "timeout": "integer"}, "req": ["path"]}, {"name": "todo_write", "ask": False, "desc": "Вести план из нескольких шагов: передай полный список с полями text и done. Вызывай для сложной задачи и отмечай шаги по мере выполнения.", "props": {"items": {"type": "array", "items": {"type": "object", "properties": { "text": {"type": "string"}, "done": {"type": "boolean"}}, "required": ["text"]}}}, "req": ["items"]}, {"name": "screenshot", "ask": False, "desc": "Сделать скриншот HTML-файла проекта (path) или сайта (url) через Chrome или Edge. Нужен, чтобы проверить вёрстку глазами: если модель умеет смотреть картинки, снимок придёт ей сразу.", "props": {"path": "string", "url": "string", "width": "integer", "height": "integer"}, "req": []}, {"name": "ask_user", "ask": False, "desc": ("Задать пользователю вопрос с вариантами ответа и возможностью написать свой. Используй, когда не хватает " "ответа для хорошего результата: стиль дизайна, цвета, формат, объём, выбор между подходами. Варианты " "придумывай сам по контексту задачи: 2–4 конкретных. Не спрашивай то, что понятно из задачи. " "До 4 вопросов за один вызов."), "props": {"questions": {"type": "array", "items": {"type": "object", "properties": { "question": {"type": "string"}, "options": {"type": "array", "items": {"type": "string"}}, "allow_custom": {"type": "boolean"}}, "required": ["question"]}}}, "req": ["questions"]}, ] ASK = {t["name"]: t["ask"] for t in TOOL_DEFS} DEFAULT_PERMS = {t["name"]: ("ask" if t["ask"] else "allow") for t in TOOL_DEFS} SKIP_DIRS = {".git", "node_modules", "__pycache__", ".venv", "venv"} RO_DENY = {"write_file", "append_file", "edit_file", "bash", "download_file", "fetch_url", "web_search", "multi_edit", "apply_patch", "move_path", "delete_path", "run_script"} # пресет «только чтение» def _prop(v): return v if isinstance(v, dict) else {"type": v} def oa_tools(): return [{"type": "function", "function": {"name": t["name"], "description": t["desc"], "parameters": { "type": "object", "properties": {k: _prop(v) for k, v in t["props"].items()}, "required": t["req"]}}} for t in TOOL_DEFS] def an_tools(): return [{"name": t["name"], "description": t["desc"], "input_schema": { "type": "object", "properties": {k: _prop(v) for k, v in t["props"].items()}, "required": t["req"]}} for t in TOOL_DEFS] def resolve(app, path): return os.path.normpath(os.path.join(app.cwd, os.path.expanduser(path or "."))) PROTECTED_MSG = "доступ запрещён: папка FlynCode с ключами API инструментам недоступна" def protected(path): """Папка FlynCode (ключи API, настройки, скиллы): инструменты модели её не читают и не меняют.""" ap = os.path.normcase(os.path.abspath(path)) base = os.path.normcase(os.path.abspath(CFG_DIR)) return ap == base or ap.startswith(base + os.sep) def strip_tags(s): return html.unescape(re.sub(r"<[^>]+>", "", s)).strip() def t_read_file(app, a): path = resolve(app, a["path"]) if protected(path): raise PermissionError(PROTECTED_MSG) with open(path, "r", encoding="utf-8", newline="") as f: lines = f.read().splitlines() off = max(1, int(a.get("offset") or 1)) lim = int(a.get("limit") or 400) chunk = lines[off - 1: off - 1 + lim] body = "\n".join("%6d\t%s" % (i + off, ln) for i, ln in enumerate(chunk)) if off - 1 + lim < len(lines): body += "\n… (всего строк: %d)" % len(lines) return body or "(пусто)", "%d строк" % len(chunk) def t_list_files(app, a): base = resolve(app, a.get("path")) found = glob.glob(os.path.join(base, a["pattern"]), recursive=True) found = [f for f in found if not protected(f)] rel = sorted(os.path.relpath(f, app.cwd) + ("/" if os.path.isdir(f) else "") for f in found)[:300] return "\n".join(rel) or "(ничего не найдено)", "%d шт." % len(rel) def t_grep(app, a): rx = re.compile(a["pattern"]) base = resolve(app, a.get("path")) inc = a.get("include") or "*" hits = [] for root, dirs, files in os.walk(base): dirs[:] = [d for d in dirs if d not in SKIP_DIRS and not protected(os.path.join(root, d))] for fn in files: if len(hits) >= 200: break if not fnmatch.fnmatch(fn, inc): continue fp = os.path.join(root, fn) if protected(fp): continue try: if os.path.getsize(fp) > 1_000_000: continue with open(fp, "r", encoding="utf-8") as f: for n, line in enumerate(f, 1): if rx.search(line): hits.append("%s:%d: %s" % (os.path.relpath(fp, app.cwd), n, line.rstrip()[:200])) if len(hits) >= 200: break except (OSError, UnicodeDecodeError): continue return "\n".join(hits) or "нет совпадений", "%d совпадений" % len(hits) def t_write_file(app, a): path = resolve(app, a["path"]) if protected(path): raise PermissionError(PROTECTED_MSG) os.makedirs(os.path.dirname(path) or ".", exist_ok=True) with open(path, "w", encoding="utf-8") as f: f.write(a["content"]) return "Записано: %s (%s)" % (path, fmt_bytes(len(a["content"].encode("utf-8")))), os.path.basename(path) def t_append_file(app, a): path = resolve(app, a["path"]) if protected(path): raise PermissionError(PROTECTED_MSG) os.makedirs(os.path.dirname(path) or ".", exist_ok=True) content = a["content"] prefix = "" total = 0 if os.path.isfile(path) and os.path.getsize(path) > 0: with open(path, "rb") as f: total = f.read().count(b"\n") with open(path, "rb") as f: f.seek(-1, 2) last = f.read(1) if last != b"\n": # последняя строка без перевода строки — не склеиваем с новым куском prefix = "\n" total += 1 with open(path, "a", encoding="utf-8", newline="") as f: f.write(prefix + content) added = content.count("\n") + (1 if content and not content.endswith("\n") else 0) return ("Дописано в %s: +%d строк, всего ≈%d" % (path, added, total + added), os.path.basename(path)) def t_edit_file(app, a): path = resolve(app, a["path"]) if protected(path): raise PermissionError(PROTECTED_MSG) old, new = a["old"], a["new"] if not old: raise ValueError("пустой фрагмент нельзя заменить") with open(path, "r", encoding="utf-8", newline="") as f: # newline="" — не трогаем переводы строк src = f.read() if "\r\n" in src: # файл с CRLF: подгоняем фрагмент под его переводы строк old = old.replace("\r\n", "\n").replace("\n", "\r\n") new = new.replace("\r\n", "\n").replace("\n", "\r\n") n = src.count(old) if n != 1: raise ValueError("фрагмент встречается %d раз(а), нужен уникальный" % n) with open(path, "w", encoding="utf-8", newline="") as f: f.write(src.replace(old, new, 1)) return "Изменено: " + path, os.path.basename(path) def kill_tree(proc): """Убивает shell вместе с дочерними процессами (иначе они держат пайп и зависают).""" try: if IS_WIN: subprocess.run(["taskkill", "/F", "/T", "/PID", str(proc.pid)], capture_output=True) else: os.killpg(proc.pid, signal.SIGKILL) except Exception: try: proc.kill() except Exception: pass def t_bash(app, a): kw = {} if IS_WIN else {"start_new_session": True} proc = subprocess.Popen(a["command"], shell=True, cwd=app.cwd, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, encoding="utf-8", errors="replace", **kw) deadline = time.time() + int(a.get("timeout") or 120) try: while True: try: so, se = proc.communicate(timeout=0.3) break except subprocess.TimeoutExpired: if CANCEL.is_set() or time.time() > deadline: raise RuntimeError("остановлено (Esc)" if CANCEL.is_set() else "таймаут") except BaseException: kill_tree(proc) raise rc = proc.returncode o = so + ("\n[stderr]\n" + se if se.strip() else "") o = o.strip() if len(o) > 8000: o = o[:4000] + "\n…\n" + o[-3000:] return "exit %d\n%s" % (rc, o or "(нет вывода)"), "exit %d" % rc def norm_url(u): u = (u or "").strip() return u if u.startswith(("http://", "https://")) else "https://" + u def t_fetch_url(app, a): with http_open(norm_url(a["url"]), UA, timeout=60, method="GET") as r: data = r.read(5_000_000) ctype = r.headers.get("Content-Type", "") text = data.decode("utf-8", "replace") if "html" in ctype or "", " ", text) text = html.unescape(re.sub(r"(?s)<[^>]+>", " ", text)) text = re.sub(r"[ \t]+", " ", text) text = re.sub(r"\n\s*\n+", "\n\n", text).strip() return text[:15000] or "(пусто)", "%d симв." % len(text) def t_download_file(app, a): path = resolve(app, a["save_to"]) if protected(path): raise PermissionError(PROTECTED_MSG) os.makedirs(os.path.dirname(path) or ".", exist_ok=True) n = 0 tmp = path + ".part" try: with http_open(norm_url(a["url"]), UA, timeout=120, method="GET") as r, open(tmp, "wb") as f: while True: if CANCEL.is_set(): raise RuntimeError("остановлено (Esc)") chunk = r.read(65536) if not chunk: break n += len(chunk) if n > 500_000_000: raise RuntimeError("файл больше 500 МБ, загрузка остановлена") f.write(chunk) os.replace(tmp, path) except BaseException: if os.path.exists(tmp): os.remove(tmp) raise return "Скачано: %s (%s)" % (path, fmt_bytes(n)), "скачано " + fmt_bytes(n) def t_web_search(app, a): url = "https://html.duckduckgo.com/html/?q=" + urllib.parse.quote(a["query"]) with http_open(url, UA, timeout=30, method="GET") as r: page = r.read().decode("utf-8", "replace") links = re.findall(r'class="result__a"[^>]*href="([^"]+)"[^>]*>(.*?)', page, re.S) snips = re.findall(r'class="result__snippet"[^>]*>(.*?)', page, re.S) res = [] for i, (href, title) in enumerate(links[:8]): m = re.search(r"uddg=([^&]+)", href) link_ = urllib.parse.unquote(m.group(1)) if m else href snip = strip_tags(snips[i]) if i < len(snips) else "" res.append("%d. %s\n %s\n %s" % (i + 1, strip_tags(title), link_, snip)) if not res: return ("Поиск не вернул результатов (возможна блокировка поисковика). " "Открой нужный сайт через fetch_url по прямой ссылке."), "0 результатов" return "\n".join(res), "%d результатов" % len(res) # ════════════════════════════════════════════════════════════════════════════ # СКИЛЛЫ — папки с инструкциями (SKILL.md) # ════════════════════════════════════════════════════════════════════════════ # Где лежат: <проект>/.flyncode/skills/<имя>/SKILL.md (важнее) и ~/.flyncode/skills/<имя>/SKILL.md # Формат файла: сверху блок --- name: ... description: ... ---, ниже инструкции в Markdown. SKILL_MAX = 30000 def open_folder(path): """Открывает папку в проводнике / Finder / файловом менеджере. True — если получилось.""" try: os.makedirs(path, exist_ok=True) if IS_WIN: os.startfile(path) elif sys.platform == "darwin": subprocess.Popen(["open", path], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) else: subprocess.Popen(["xdg-open", path], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, start_new_session=True) return True except Exception: return False def skill_dirs(cwd): return [os.path.join(cwd, ".flyncode", "skills"), os.path.join(CFG_DIR, "skills")] def parse_skill(path): try: with open(path, "r", encoding="utf-8", errors="replace") as f: text = f.read(SKILL_MAX + 4000) except OSError: return None meta, body = {}, text m = re.match(r"\ufeff?---[ \t]*\r?\n(.*?)\r?\n---[ \t]*(?:\r?\n|$)", text, re.S) if m: for line in m.group(1).splitlines(): if ":" in line and not line[:1].isspace(): k, v = line.split(":", 1) meta[k.strip().lower()] = v.strip().strip("\"'") body = text[m.end():] fname = os.path.basename(path) default = os.path.basename(os.path.dirname(path)) if fname.lower() == "skill.md" else os.path.splitext(fname)[0] name = (meta.get("name") or default).strip().replace(" ", "-").lower() return {"name": name, "description": meta.get("description") or "(нет описания)", "path": path, "body": body.strip()[:SKILL_MAX]} def list_skills(cwd): """Имя → скилл. Скилл из папки проекта перекрывает глобальный с тем же именем.""" found = {} for d in skill_dirs(cwd): try: subs = sorted(os.listdir(d)) except OSError: continue for sub in subs: full = os.path.join(d, sub) if os.path.isdir(full): p = os.path.join(full, "SKILL.md") elif sub.lower().endswith(".md"): p = full else: continue if os.path.isfile(p): s = parse_skill(p) if s and s["name"] not in found: found[s["name"]] = s return dict(sorted(found.items())) def skill_report(cwd): """Диагностика для /skills: какие папки проверяются и что в них найдено.""" lines = [] loaded = list_skills(cwd) def status(p): s = parse_skill(p) if s is None: return "не читается" if loaded.get(s["name"], {}).get("path") == p: return "загружен как @%s" % s["name"] return "пропущен: имя @%s уже занято другим файлом" % s["name"] for d in skill_dirs(cwd): if not os.path.isdir(d): lines.append("%s — папки нет" % d) continue try: items = sorted(os.listdir(d)) except OSError as e: lines.append("%s — не открывается: %s" % (d, e)) continue lines.append("%s — элементов: %d" % (d, len(items))) for it in items: full = os.path.join(d, it) if os.path.isdir(full): p = os.path.join(full, "SKILL.md") lines.append(" %s/ — %s" % (it, status(p) if os.path.isfile(p) else "нет файла SKILL.md")) elif it.lower().endswith(".md"): lines.append(" %s — %s" % (it, status(full))) return lines def find_skill(cwd, name): return list_skills(cwd).get((name or "").strip().lower()) def skills_mentioned(cwd, text): """Скиллы, чьи имена встречаются в сообщении словом: их инструкции идут в запрос сразу.""" if not text or "[Скилл" in text: return [] low = text.lower() found = [] for name, sk in list_skills(cwd).items(): if re.search(r"(? deadline: raise RuntimeError("остановлено (Esc)" if CANCEL.is_set() else "таймаут %d с" % timeout) except BaseException: kill_tree(proc) raise o = (so + ("\n[stderr]\n" + se if se.strip() else "")).strip() if len(o) > 8000: o = o[:4000] + "\n…\n" + o[-3000:] return proc.returncode, o or "(нет вывода)" def t_multi_edit(app, a): path = resolve(app, a["path"]) if protected(path): raise PermissionError(PROTECTED_MSG) with open(path, "r", encoding="utf-8", newline="") as f: src = f.read() crlf = "\r\n" in src edits = a.get("edits") or [] if not edits: raise ValueError("нет правок") for i, ed in enumerate(edits, 1): old, new = ed.get("old", ""), ed.get("new", "") if not old: raise ValueError("правка %d: пустой фрагмент нельзя заменить" % i) if crlf: old = old.replace("\r\n", "\n").replace("\n", "\r\n") new = new.replace("\r\n", "\n").replace("\n", "\r\n") n = src.count(old) if n != 1: raise ValueError("правка %d: фрагмент встречается %d раз(а), нужен уникальный — файл не изменён" % (i, n)) src = src.replace(old, new, 1) with open(path, "w", encoding="utf-8", newline="") as f: f.write(src) return "Правок применено: %d — %s" % (len(edits), path), os.path.basename(path) def _strip_ab(p): p = p.strip().split("\t")[0] return p[2:] if p.startswith(("a/", "b/")) else p def parse_patch(text): """unified diff → [{'old', 'new', 'hunks': [{'start', 'lines': [(тег, текст)]}]}].""" files, cur, hunk = [], None, None lines = text.replace("\r\n", "\n").split("\n") i = 0 while i < len(lines): ln = lines[i] if ln.startswith("--- ") and i + 1 < len(lines) and lines[i + 1].startswith("+++ "): old_p, new_p = _strip_ab(ln[4:]), _strip_ab(lines[i + 1][4:]) cur = {"old": None if old_p == "/dev/null" else old_p, "new": None if new_p == "/dev/null" else new_p, "hunks": []} files.append(cur) hunk = None i += 2 continue m = HUNK_RE.match(ln) if m: if cur is None: raise ValueError("блок @@ без заголовка файла (нужны строки --- и +++)") hunk = {"start": int(m.group(1)), "lines": []} cur["hunks"].append(hunk) i += 1 continue if hunk is not None: if ln[:1] in (" ", "-", "+"): hunk["lines"].append((ln[:1], ln[1:])) elif ln == "" and any(x for x in lines[i:]): hunk["lines"].append((" ", "")) elif ln.startswith("\\"): pass # «\ No newline at end of file» elif ln != "": hunk = None i += 1 return files def apply_patch_files(app, files): """Применяет патч в памяти: [(путь, новый текст, отн. имя)]. Файлы на диск пишутся только если всё сошлось.""" results = [] for f in files: if f["new"] is None: raise ValueError("удаление через патч не поддержано — используй delete_path") path = resolve(app, f["new"]) if protected(path): raise PermissionError(PROTECTED_MSG) if f["old"] is None: # новый файл if os.path.exists(path) and os.path.getsize(path) > 0: raise ValueError("файл уже есть: %s — для изменения нужен контекст из него" % f["new"]) text = "" else: if not os.path.isfile(path): raise FileNotFoundError("нет такого файла: %s" % f["new"]) with open(path, "r", encoding="utf-8", newline="") as fh: text = fh.read() crlf = "\r\n" in text body = text.replace("\r\n", "\n") endnl = body == "" or body.endswith("\n") lines = body.split("\n") if body.endswith("\n"): lines = lines[:-1] if body == "": lines = [] delta = 0 for hi, h in enumerate(f["hunks"], 1): old = [t for tag, t in h["lines"] if tag in (" ", "-")] new = [t for tag, t in h["lines"] if tag in (" ", "+")] pos = (h["start"] - 1 if old else h["start"]) + delta pos = max(0, min(pos, len(lines))) if not old: found = pos elif lines[pos:pos + len(old)] == old: found = pos else: cands = [p for p in range(len(lines) - len(old) + 1) if lines[p:p + len(old)] == old] if not cands: raise ValueError("%s: блок %d не совпал с текущим файлом — проверь контекст" % (f["new"], hi)) found = min(cands, key=lambda p: abs(p - pos)) lines[found:found + len(old)] = new delta += len(new) - len(old) out_s = "\n".join(lines) + ("\n" if lines and endnl else "") if crlf: out_s = out_s.replace("\n", "\r\n") results.append((path, out_s, f["new"])) return results def t_apply_patch(app, a): files = parse_patch(a.get("patch") or "") if not files: raise ValueError("в патче нет файлов (нужны строки --- и +++)") results = apply_patch_files(app, files) for path, text, _ in results: os.makedirs(os.path.dirname(path) or ".", exist_ok=True) with open(path, "w", encoding="utf-8", newline="") as fh: fh.write(text) return "Патч применён: " + ", ".join(rel for _, _, rel in results), "%d файл(ов)" % len(results) def t_move_path(app, a): src, dst = resolve(app, a["src"]), resolve(app, a["dst"]) if protected(src) or protected(dst): raise PermissionError(PROTECTED_MSG) if not os.path.exists(src): raise FileNotFoundError("нет такого пути: " + a["src"]) if os.path.exists(dst): raise FileExistsError("цель уже существует: " + a["dst"]) os.makedirs(os.path.dirname(dst) or ".", exist_ok=True) shutil.move(src, dst) return "Перемещено: %s → %s" % (a["src"], a["dst"]), os.path.basename(dst) def t_delete_path(app, a): p = resolve(app, a["path"]) if protected(p): raise PermissionError(PROTECTED_MSG) guard = {os.path.normcase(os.path.abspath(x)) for x in (app.cwd, os.sep, os.path.expanduser("~"))} if os.path.normcase(os.path.abspath(p)) in guard: raise PermissionError("рабочую папку, корень диска или домашнюю папку удалять нельзя") if not os.path.lexists(p): raise FileNotFoundError("нет такого пути: " + a["path"]) if os.path.isdir(p) and not os.path.islink(p): if not a.get("recursive"): raise ValueError("это папка: для удаления с содержимым укажи recursive=true") shutil.rmtree(p) else: os.remove(p) return "Удалено: " + a["path"], os.path.basename(p) def t_git(app, a): if not detect_runtimes().get("git"): raise RuntimeError("git не найден в системе") args = shlex.split(a.get("args") or "", posix=not IS_WIN) if not args: raise ValueError("укажи подкоманду, например: status или diff") sub = args[0] if sub not in GIT_READONLY: raise PermissionError("git %s не разрешён: только чтение (%s)" % (sub, ", ".join(sorted(GIT_READONLY)))) for x in args: if x in GIT_BAD or any(x.startswith(b + "=") for b in GIT_BAD if b.startswith("--")): raise PermissionError("опция %s запрещена" % x) if sub == "branch" and any(x not in GIT_BRANCH_OK for x in args[1:]): raise PermissionError("создавать и менять ветки через git нельзя — используй bash") if sub == "remote" and any(x not in ("-v", "--verbose") for x in args[1:]): raise PermissionError("менять remote через git нельзя — используй bash") rc, out_s = run_proc(["git", "--no-pager"] + args, app.cwd, timeout=60) return "exit %d\n%s" % (rc, out_s), "git %s · exit %d" % (sub, rc) def t_run_script(app, a): path = resolve(app, a["path"]) if protected(path): raise PermissionError(PROTECTED_MSG) if not os.path.isfile(path): raise FileNotFoundError("нет такого файла: " + a["path"]) prefix, how = runner_for(path) if prefix is None: raise RuntimeError(how) argv = prefix + [path] + shlex.split(a.get("args") or "", posix=not IS_WIN) rc, out_s = run_proc(argv, app.cwd, timeout=int(a.get("timeout") or 120)) return "[%s] exit %d\n%s" % (how, rc, out_s), "%s · exit %d" % (how, rc) def t_todo_write(app, a): todos = [] for it in a.get("items") or []: if isinstance(it, str) and it.strip(): todos.append({"text": it.strip(), "done": False}) elif isinstance(it, dict) and str(it.get("text", "")).strip(): todos.append({"text": str(it["text"]).strip(), "done": bool(it.get("done"))}) app.plan = todos println() if not todos: println(" " + c("muted", "План очищен")) for t in todos: mark = c("ok", "[x]") if t["done"] else c("muted", "[ ]") println(" " + mark + " " + c("muted" if t["done"] else "text", t["text"])) println() done = sum(1 for t in todos if t["done"]) lines = ["[%s] %s" % ("x" if t["done"] else " ", t["text"]) for t in todos] return "План: %d из %d выполнено\n%s" % (done, len(todos), "\n".join(lines)), "%d/%d" % (done, len(todos)) # ════════════════════════════════════════════════════════════════════════════ # АВТОПРОВЕРКА ПРАВОК И СКРИНШОТЫ # ════════════════════════════════════════════════════════════════════════════ AUTOCHECK_LEVELS = [5, 10, 20] # сколько попыток исправления после автопроверки (меняется в настройках) OK_RE = re.compile(r"\s*(OK|ОК|Ok)[.!]?\s*") WRITE_TOOLS = {"write_file", "append_file", "edit_file", "multi_edit", "apply_patch", "move_path"} VISION_RE = re.compile(r"gpt-4o|gpt-4\.1|gpt-4-turbo|gpt-4-vision|gpt-5|^o[134](-|$)|claude|gemini|grok-4|" r"vision|-vl\b|llava|pixtral|llama-4|gemma-3|qwen2\.5-vl|qwen3-vl", re.I) VISION_LABEL = {"auto": "по названию модели", "on": "всегда", "off": "выкл"} VOID_TAGS = {"area", "base", "br", "col", "embed", "hr", "img", "input", "link", "meta", "param", "source", "track", "wbr"} BLOCK_TAGS = {"div", "section", "main", "header", "footer", "nav", "article", "aside", "ul", "ol", "table", "form", "button"} SHOT_DIR = os.path.join(CFG_DIR, "shots") SHOT_MAX = 4 * 1024 * 1024 SHOT_TL = threading.local() # путь последнего скриншота в этом потоке: агенты работают параллельно def vision_ok(cfg, mid, models=()): """Понимает ли модель картинки: по настройке «Скриншоты для зрения» или по названию модели.""" mode = (cfg.get("settings") or {}).get("vision", "auto") if mode in ("on", "off"): return mode == "on" for m in models or (): if m.get("id") == mid and m.get("vision"): return True return bool(VISION_RE.search(mid or "")) ATTACH_DIR = os.path.join(CFG_DIR, "attachments") IMG_EXT = {".png": "image/png", ".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".webp": "image/webp", ".gif": "image/gif"} IMG_MAX = 3_500_000 # байт: после base64 (+33%) остаётся в лимитах провайдеров def detect_image_mime(raw): """Тип картинки по первым байтам (расширению не доверяем). None — если это не картинка.""" if raw.startswith(b"\x89PNG\r\n\x1a\n"): return "image/png" if raw.startswith(b"\xff\xd8\xff"): return "image/jpeg" if raw.startswith((b"GIF87a", b"GIF89a")): return "image/gif" if raw[:4] == b"RIFF" and raw[8:12] == b"WEBP": return "image/webp" return None def load_image_file(path): """Читает картинку с диска. Возвращает {'mime', 'data', 'name'} или поднимает ValueError с понятным текстом.""" name = os.path.basename(path) if os.path.splitext(path)[1].lower() not in IMG_EXT: raise ValueError("%s — не картинка (нужны png, jpg, webp или gif)" % name) size = os.path.getsize(path) if size > IMG_MAX: raise ValueError("%s весит %s — больше лимита 3.5 МБ, сожми картинку" % (name, fmt_bytes(size))) with open(path, "rb") as f: raw = f.read() mime = detect_image_mime(raw) if not mime: raise ValueError("%s повреждён или это не картинка" % name) return {"mime": mime, "data": base64.b64encode(raw).decode("ascii"), "name": name} def split_image_paths(line, cwd): """Ищет в сообщении пути к существующим картинкам: в кавычках, с «\\ » перед пробелом или просто через пробел (так терминал вставляет перетащенный файл). Возвращает (текст без путей, [полные пути]).""" found = [] def take(m): raw = m.group(1) or m.group(2) or (m.group(3) or "") p = raw.replace("\\ ", " ") full = os.path.normpath(os.path.join(cwd, os.path.expanduser(p))) if os.path.splitext(p)[1].lower() in IMG_EXT and os.path.isfile(full): found.append(full) return " " return m.group(0) rest = re.sub(r"\"([^\"]+)\"|'([^']+)'|((?:\\ |\S)+)", take, line) return re.sub(r"\s+", " ", rest).strip(), found def clipboard_image(): """Сохраняет картинку из буфера обмена во временный PNG. Возвращает (путь, '') или (None, причина).""" os.makedirs(ATTACH_DIR, exist_ok=True) dst = os.path.join(ATTACH_DIR, "clip_%d.png" % int(time.time() * 1000)) kw = {"creationflags": 0x08000000} if IS_WIN else {} # без окна PowerShell if IS_WIN: ps = ("Add-Type -AssemblyName System.Windows.Forms; Add-Type -AssemblyName System.Drawing; " "$i=[System.Windows.Forms.Clipboard]::GetImage(); if($i -eq $null){exit 3}; " "$i.Save('%s',[System.Drawing.Imaging.ImageFormat]::Png)" % dst.replace("'", "''")) argv = ["powershell", "-NoProfile", "-STA", "-Command", ps] elif sys.platform == "darwin": if not shutil.which("pngpaste"): return None, "для вставки из буфера на macOS поставь pngpaste: brew install pngpaste" argv = ["pngpaste", dst] else: if shutil.which("wl-paste"): argv = ["sh", "-c", 'wl-paste -t image/png > "$0"', dst] elif shutil.which("xclip"): argv = ["sh", "-c", 'xclip -selection clipboard -t image/png -o > "$0"', dst] else: return None, "для вставки из буфера установи xclip или wl-clipboard" try: subprocess.run(argv, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=20, **kw) except (OSError, subprocess.TimeoutExpired): return None, "не получилось прочитать буфер обмена" if not os.path.isfile(dst) or os.path.getsize(dst) == 0: return None, "в буфере обмена нет картинки" return dst, "" def clipboard_text(): """Текст из буфера обмена (для Ctrl+V). Пустая строка, если текста нет или нужной утилиты нет.""" kw = {"creationflags": 0x08000000} if IS_WIN else {} if IS_WIN: argv = ["powershell", "-NoProfile", "-Command", "[Console]::OutputEncoding=[Text.Encoding]::UTF8; Get-Clipboard -Raw"] elif sys.platform == "darwin": argv = ["pbpaste"] elif shutil.which("wl-paste"): argv = ["wl-paste", "-n"] elif shutil.which("xclip"): argv = ["xclip", "-selection", "clipboard", "-o"] else: return "" try: r = subprocess.run(argv, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, timeout=10, **kw) except (OSError, subprocess.TimeoutExpired): return "" txt = r.stdout.decode("utf-8", "replace") return " ".join(txt.replace("\r", "").split("\n")).strip() def image_part(path): with open(path, "rb") as f: return {"mime": "image/png", "data": base64.b64encode(f.read()).decode("ascii")} def js_check_src(code, ext=".js"): """Синтаксис JavaScript через Node, если он установлен. Возвращает текст ошибки или None.""" node = shutil.which("node") if not node or not code.strip(): return None fd, tmp = tempfile.mkstemp(suffix=".cjs" if ext == ".cjs" else ".mjs") try: with os.fdopen(fd, "w", encoding="utf-8") as f: f.write(code) r = subprocess.run([node, "--check", tmp], stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, encoding="utf-8", errors="replace", timeout=20) except (OSError, subprocess.TimeoutExpired): return None finally: try: os.unlink(tmp) except OSError: pass if r.returncode == 0: return None lines = [ln.strip() for ln in (r.stderr or "").splitlines() if ln.strip()] loc = re.search(r":(\d+)$", lines[0]) if lines else None errs = [ln for ln in lines if "Error" in ln] msg = errs[0] if errs else (lines[-1] if lines else "синтаксическая ошибка") return ("строка %s: " % loc.group(1) if loc else "") + msg class _TagCheck(HTMLParser): def __init__(self): super().__init__(convert_charrefs=True) self.stack, self.errors = [], [] def handle_starttag(self, tag, attrs): if tag not in VOID_TAGS: self.stack.append((tag, self.getpos()[0])) def handle_endtag(self, tag): if tag in VOID_TAGS: return for i in range(len(self.stack) - 1, -1, -1): if self.stack[i][0] == tag: for t, ln in self.stack[i + 1:]: if t in BLOCK_TAGS: self.errors.append("<%s> на строке %d не закрыт" % (t, ln)) del self.stack[i:] return if tag in BLOCK_TAGS: self.errors.append("лишний на строке %d" % (tag, self.getpos()[0])) def html_check(src): """Грубая проверка HTML: незакрытые блоки, лишние закрывающие теги и синтаксис встроенных скриптов.""" problems = [] p = _TagCheck() try: p.feed(src) p.close() except Exception as e: problems.append("HTML не разобрался: %s" % e) problems += p.errors[:3] problems += ["<%s> на строке %d не закрыт" % (t, ln) for t, ln in p.stack if t in BLOCK_TAGS][-3:] for attrs, code in re.findall(r"]*)>(.*?)", src, re.S | re.I): if re.search(r"\bsrc\s*=", attrs) or not code.strip(): continue ext = ".mjs" if re.search(r"type\s*=\s*[\"']module", attrs, re.I) else ".js" err = js_check_src(code, ext) if err: problems.append("ошибка в скрипте, " + err) return "; ".join(problems) or None def check_file(path): """Проверка одного изменённого файла. Возвращает текст проблемы или None.""" ext = os.path.splitext(path)[1].lower() try: if os.path.getsize(path) > 2_000_000: return None with open(path, "r", encoding="utf-8", errors="replace") as f: src = f.read() except OSError as e: return "не читается: %s" % e if ext == ".py": try: compile(src, path, "exec") except SyntaxError as e: return "строка %s: %s" % (e.lineno or "?", e.msg) except (ValueError, TypeError) as e: return str(e) return None if ext == ".json": try: json.loads(src) except ValueError as e: return "некорректный JSON: %s" % e return None if ext in (".js", ".mjs", ".cjs"): return js_check_src(src, ext) if ext == ".css": plain = re.sub(r"/\*.*?\*/", "", src, flags=re.S) o, cl = plain.count("{"), plain.count("}") return None if o == cl else "не сходятся фигурные скобки: открыто %d, закрыто %d" % (o, cl) if ext in (".html", ".htm"): return html_check(src) return None def find_browser(): for n in ("google-chrome", "google-chrome-stable", "chromium", "chromium-browser", "chrome", "msedge", "microsoft-edge"): p = shutil.which(n) if p: return p if IS_WIN: roots = [os.environ.get("ProgramFiles", ""), os.environ.get("ProgramFiles(x86)", ""), os.environ.get("LOCALAPPDATA", "")] rel = [os.path.join("Google", "Chrome", "Application", "chrome.exe"), os.path.join("Microsoft", "Edge", "Application", "msedge.exe")] cands = [os.path.join(r, x) for r in roots if r for x in rel] elif sys.platform == "darwin": cands = ["/Applications/Google Chrome.app/Contents/MacOS/Google Chrome", "/Applications/Chromium.app/Contents/MacOS/Chromium", "/Applications/Microsoft Edge.app/Contents/MacOS/Microsoft Edge"] else: cands = [] return next((p for p in cands if os.path.isfile(p)), None) def run_cancelable(argv, timeout=60, **kw): """subprocess.run, который прерывается по Esc и убивает всё дерево процессов.""" kw = dict(kw) if not IS_WIN: kw.setdefault("start_new_session", True) proc = subprocess.Popen(argv, **kw) deadline = time.time() + timeout try: while True: try: return proc.wait(timeout=0.2) except subprocess.TimeoutExpired: if CANCEL.is_set(): raise KeyboardInterrupt("отменено пользователем") if time.time() > deadline: raise subprocess.TimeoutExpired(argv, timeout) except BaseException: kill_tree(proc) raise def take_screenshot(url, w=1280, h=800): """Снимок страницы через headless Chrome/Edge. Возвращает путь к PNG в ~/.flyncode/shots.""" exe = find_browser() if not exe: raise RuntimeError("не нашёл Chrome, Chromium или Edge: поставь любой из них, и скриншоты заработают") os.makedirs(SHOT_DIR, exist_ok=True) png = os.path.join(SHOT_DIR, "shot_%d.png" % int(time.time() * 1000)) args = ["--disable-gpu", "--hide-scrollbars", "--no-first-run", "--no-default-browser-check", "--user-data-dir=" + os.path.join(SHOT_DIR, "profile"), "--window-size=%d,%d" % (w, h), "--virtual-time-budget=3000", "--screenshot=" + png, url] if not IS_WIN and hasattr(os, "geteuid") and os.geteuid() == 0: args.insert(0, "--no-sandbox") kw = {} if IS_WIN else {"start_new_session": True} for head in ("--headless=new", "--headless"): try: run_cancelable([exe, head] + args, timeout=60, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, **kw) except subprocess.TimeoutExpired: continue if os.path.isfile(png) and os.path.getsize(png) > 0: break if not os.path.isfile(png) or os.path.getsize(png) == 0: raise RuntimeError("браузер не сохранил скриншот") if os.path.getsize(png) > SHOT_MAX: raise RuntimeError("скриншот слишком большой для модели (%s)" % fmt_bytes(os.path.getsize(png))) return png def t_screenshot(app, a): src = (a.get("url") or a.get("path") or "").strip() if not src: raise ValueError("укажи path (HTML-файл) или url") if src.lower().startswith(("http://", "https://")): url = src else: path = resolve(app, src) if protected(path): raise PermissionError(PROTECTED_MSG) if not os.path.isfile(path): raise FileNotFoundError("нет файла: " + src) url = pathlib.Path(path).as_uri() w = min(1920, max(320, int(a.get("width") or 1280))) h = min(1600, max(240, int(a.get("height") or 800))) png = take_screenshot(url, w, h) SHOT_TL.path = png if vision_ok(app.cfg, app.model, app.models): note = "Изображение приложено ниже: посмотри и оцени вёрстку." else: note = "Модель не умеет смотреть картинки, поэтому оцени по коду." return "Скриншот %dx%d снят. %s" % (w, h, note), "скриншот %dx%d" % (w, h) def track_run(run, name, args, root): """Запоминает, какие файлы поменял ход агента и какие скриншоты сделаны (отдельно для каждого агента).""" if name in WRITE_TOOLS: if name in ("write_file", "append_file", "edit_file", "multi_edit"): paths = [args.get("path")] elif name == "move_path": paths = [args.get("dst")] else: paths = re.findall(r"^\+\+\+ (?:b/)?(\S+)", args.get("patch") or "", re.M) for p in paths: if p and p != "/dev/null": full = os.path.normpath(os.path.join(root, os.path.expanduser(p))) run["changed"].add(full) run["touched"].add(full) elif name == "screenshot": p = getattr(SHOT_TL, "path", None) SHOT_TL.path = None if p: run["shots"].append(p) AGENT_LOCK = threading.RLock() # агенты работают по очереди: один за раз, никто никому не мешает AGENT_DEPTH = [0] # агент может поручить другому агенту, но не бесконечно MAX_AGENT_DEPTH = 2 def t_create_agent(app, a): name = re.sub(r"[^\w-]", "", str(a.get("name") or "").strip().replace(" ", "-"))[:24] role = str(a.get("role") or "").strip() if not name or not role: raise ValueError("нужны name и role") if not app.p: raise ValueError("сначала подключи провайдера (/provider)") model = str(a.get("model") or "").strip() or app.model if not model: raise ValueError("сначала выбери модель (/model)") agents = app.ensure_agents() rec = next((x for x in agents if x["name"].lower() == name.lower()), None) if rec is None: rec = {"name": name, "pid": app.p["id"], "model": model, "role": role, "on": True, "tools": True, "auto": True} agents.append(rec) action = "создан" else: rec.update(model=model, role=role, on=True) action = "обновлён" save_cfg(app.cfg) return "Агент @%s %s: модель %s. Роль: %s" % (rec["name"], action, model, role), "агент @%s" % rec["name"] def t_ask_agent(app, a): name = str(a.get("name") or "").strip() task = str(a.get("task") or "").strip() alive = [x for x in app.ensure_agents() if x.get("on", True)] rec = next((x for x in alive if x["name"].lower() == name.lower()), None) if rec is None: known = ", ".join("@" + x["name"] for x in alive) or "нет" raise ValueError("нет агента «%s». Есть: %s. Создай его через create_agent." % (name, known)) if AGENT_DEPTH[0] >= MAX_AGENT_DEPTH: raise ValueError("агенты вложены слишком глубоко — реши эту часть сам") AGENT_DEPTH[0] += 1 try: with AGENT_LOCK: ans = app.agent_run(rec, task) finally: AGENT_DEPTH[0] -= 1 if not ans or not ans.strip(): raise ValueError("агент @%s не дал ответа (возможно, остановлен)" % rec["name"]) return "[Ответ агента @%s]\n%s" % (rec["name"], ans.strip()), "ответ @%s" % rec["name"] def t_ask_user(app, a): """Опрос пользователя: модель задаёт вопросы с вариантами, пользователь выбирает или пишет свой ответ.""" if getattr(_tl, "buf", None) is not None: raise ValueError("в параллельном режиме спрашивать нельзя — реши сам и скажи, что выбрал") qs = a.get("questions") or [] if isinstance(qs, dict): qs = [qs] qs = [q for q in qs if isinstance(q, dict) and str(q.get("question") or "").strip()][:4] if not qs: raise ValueError("нет вопросов: укажи questions с полем question") lines = [] for q in qs: text = str(q["question"]).strip() opts = [str(o).strip() for o in (q.get("options") or []) if str(o).strip()][:8] custom = bool(q.get("allow_custom", True)) or not opts ans = app.poll_one(text, opts, custom) if ans is None: lines.append("%s → пользователь пропустил вопрос: реши сам и скажи, что выбрал" % text) else: lines.append("%s → %s" % (text, ans)) return "Ответы пользователя:\n" + "\n".join(lines), "%d ответ(ов)" % len(qs) TOOL_FN = {"read_file": t_read_file, "list_files": t_list_files, "grep": t_grep, "write_file": t_write_file, "append_file": t_append_file, "edit_file": t_edit_file, "bash": t_bash, "fetch_url": t_fetch_url, "web_search": t_web_search, "download_file": t_download_file, "load_skill": t_load_skill, "create_agent": t_create_agent, "ask_agent": t_ask_agent, "multi_edit": t_multi_edit, "apply_patch": t_apply_patch, "move_path": t_move_path, "delete_path": t_delete_path, "git": t_git, "run_script": t_run_script, "todo_write": t_todo_write, "screenshot": t_screenshot, "ask_user": t_ask_user} def run_tool(app, name, args): fn = TOOL_FN.get(name) if fn is None: return False, "Неизвестный инструмент: " + name, "" try: text, summ = fn(app, args) return True, text, summ except Exception as e: return False, "Ошибка: %s" % e, "" def describe(name, a): if name == "bash": s = a.get("command", "") elif name in ("fetch_url", "download_file"): s = a.get("url", "") + ((" → " + a["save_to"]) if a.get("save_to") else "") elif name == "web_search": s = "поиск: " + a.get("query", "") elif name == "create_agent": s = "агент @%s (%s): %s" % (a.get("name", ""), a.get("model") or "текущая модель", a.get("role", "")) elif name == "ask_agent": s = "агенту @%s: %s" % (a.get("name", ""), a.get("task", "")) elif name == "grep": s = "grep %s %s" % (a.get("pattern", ""), a.get("path", ".")) elif name == "list_files": s = "list_files " + a.get("pattern", "") elif name == "apply_patch": names = re.findall(r"^\+\+\+ (?:b/)?(\S+)", a.get("patch") or "", re.M) s = "патч: " + (", ".join(names[:5]) or "?") elif name == "multi_edit": s = "правки %s (%d)" % (a.get("path", ""), len(a.get("edits") or [])) elif name == "move_path": s = "%s → %s" % (a.get("src", ""), a.get("dst", "")) elif name == "delete_path": s = "удалить " + a.get("path", "") + (" (с содержимым)" if a.get("recursive") else "") elif name == "git": s = "git " + a.get("args", "") elif name == "run_script": s = "запуск " + a.get("path", "") + ((" " + a["args"]) if a.get("args") else "") elif name == "todo_write": s = "план: %d шагов" % len(a.get("items") or []) elif name == "screenshot": s = "скриншот " + (a.get("path") or a.get("url") or "") elif name == "ask_user": qs = a.get("questions") or [] first = qs[0].get("question", "") if qs and isinstance(qs[0], dict) else "" s = "вопрос: " + str(first) elif name == "load_skill": s = "скилл " + a.get("name", "") else: s = "%s %s" % (name, a.get("path", "")) return clip(s.replace("\n", " ⏎ "), max(40, term_w() - 14)) def parse_args(s): if isinstance(s, dict): return s if not (s or "").strip(): return {} try: v = json.loads(s) return v if isinstance(v, dict) else {} except ValueError: return {"_raw": s} # ════════════════════════════════════════════════════════════════════════════ # ФОРМАТЫ СООБЩЕНИЙ И СТРИМИНГ # ════════════════════════════════════════════════════════════════════════════ # Внутренний формат истории: user {content}; assistant {content, tool_calls[{id,name,args}], blocks}; # tool {tool_call_id, name, content}. Здесь он переводится в формат провайдера. DUMMY_SIG = "skip_thought_signature_validator" def oa_tool_call(tc, pid=""): item = {"id": tc["id"], "type": "function", "function": { "name": tc["name"], "arguments": json.dumps(tc["args"], ensure_ascii=False)}} extra = tc.get("extra") if not extra and pid == "gemini": extra = {"google": {"thought_signature": DUMMY_SIG}} if extra: item["extra_content"] = extra return item def oa_user_content(m): if not m.get("images"): return m["content"] parts = [{"type": "image_url", "image_url": {"url": "data:%s;base64,%s" % (im["mime"], im["data"])}} for im in m["images"]] return parts + [{"type": "text", "text": m["content"]}] def oa_messages(msgs, system, pid="", token_saving="auto"): res = [{"role": "system", "content": system}] for m in msgs: if m["role"] == "user": res.append({"role": "user", "content": oa_user_content(m)}) elif m["role"] == "assistant": d = {"role": "assistant", "content": m.get("content") or None} if m.get("tool_calls"): d["tool_calls"] = [oa_tool_call(tc, pid) for tc in m["tool_calls"]] else: d["content"] = d["content"] or "" res.append(d) elif m["role"] == "tool": cnt = m["content"] if token_saving != "off" and isinstance(cnt, str) and len(cnt) > 10000: cnt = cnt[:5000] + "\n... (обрезано) ...\n" + cnt[-2500:] res.append({"role": "tool", "tool_call_id": m["tool_call_id"], "content": cnt}) return res def an_messages(msgs, token_saving="auto"): res = [] for m in msgs: if m["role"] == "user" and m.get("images"): blocks = [{"type": "image", "source": {"type": "base64", "media_type": im["mime"], "data": im["data"]}} for im in m["images"]] blocks.append({"type": "text", "text": m["content"]}) if res and res[-1]["role"] == "user" and isinstance(res[-1]["content"], list): res[-1]["content"].extend(blocks) else: res.append({"role": "user", "content": blocks}) elif m["role"] == "user": res.append({"role": "user", "content": m["content"]}) elif m["role"] == "assistant": blocks = m.get("blocks") if not blocks: blocks = [] if m.get("content"): blocks.append({"type": "text", "text": m["content"]}) for tc in m.get("tool_calls") or []: blocks.append({"type": "tool_use", "id": tc["id"], "name": tc["name"], "input": tc["args"]}) if blocks: res.append({"role": "assistant", "content": blocks}) elif m["role"] == "tool": cnt = m["content"] if token_saving == "off" else ( m["content"][:5000] + "\n... (обрезано) ...\n" + m["content"][-2500:] if isinstance(m["content"], str) and len(m["content"]) > 10000 else m["content"]) blk = {"type": "tool_result", "tool_use_id": m["tool_call_id"], "content": cnt} if res and res[-1]["role"] == "user" and isinstance(res[-1]["content"], list): res[-1]["content"].append(blk) else: res.append({"role": "user", "content": [blk]}) return res class ThinkSplit: """Разделяет … внутри обычного текста.""" def __init__(self): self.inside = False self.buf = "" def feed(self, text): self.buf += text res = [] while True: tag = "" if self.inside else "" i = self.buf.find(tag) if i >= 0: if i: res.append(("think" if self.inside else "text", self.buf[:i])) self.buf = self.buf[i + len(tag):] self.inside = not self.inside continue keep = 0 for k in range(min(len(tag) - 1, len(self.buf)), 0, -1): if tag.startswith(self.buf[-k:]): keep = k break emit = self.buf[:len(self.buf) - keep] self.buf = self.buf[len(self.buf) - keep:] if emit: res.append(("think" if self.inside else "text", emit)) return res def flush(self): if self.buf: r = [("think" if self.inside else "text", self.buf)] self.buf = "" return r return [] def tools_on(app, msgs): # Инструменты отправляем, если включены, либо если в истории уже есть вызовы инструментов. has_hist = any(m["role"] == "tool" for m in msgs) if getattr(app, "mode", "build") == "chat": return has_hist return app.cfg["settings"]["tools"] or has_hist def stream_openai(app, msgs): p, lvl = app.p, app.effort body = {"model": app.model, "stream": True, "messages": oa_messages(msgs, app.system_prompt(), p["id"], (app.cfg.get("settings") or {}).get("token_saving", "auto"))} if not p.get("no_stream_usage"): body["stream_options"] = {"include_usage": True} if tools_on(app, msgs): body["tools"] = oa_tools() if p["id"] == "openrouter": body["usage"] = {"include": True} if lvl not in ("auto", "off"): body["reasoning"] = {"effort": lvl} elif lvl not in ("auto", "off"): body["reasoning_effort"] = lvl headers = {"Accept": "text/event-stream"} if app.key: headers["Authorization"] = p.get("auth", "Bearer") + " " + app.key if p["id"] == "openrouter": headers.update({"HTTP-Referer": "https://github.com/flyncode", "X-Title": "FlynCode"}) resp = http_open(app.base.rstrip("/") + "/chat/completions", headers, body) ACTIVE["resp"] = resp ACTIVE_ALL.add(resp) split = ThinkSplit() acc, usage = {}, None with resp: for ch in sse(resp): if ch.get("error"): e = ch["error"] raise ApiError(e.get("message", str(e)) if isinstance(e, dict) else str(e)) if ch.get("usage"): u = ch["usage"] det = u.get("completion_tokens_details") or {} usage = {"in": u.get("prompt_tokens", 0), "out": u.get("completion_tokens", 0), "think": det.get("reasoning_tokens", 0) or 0, "cost": u.get("cost")} for choice in ch.get("choices") or []: d = choice.get("delta") or {} rt = d.get("reasoning_content") or d.get("reasoning") if isinstance(rt, str) and rt: yield ("think", rt) tx = d.get("content") if isinstance(tx, str) and tx: for ev in split.feed(tx): yield ev for tc in d.get("tool_calls") or []: slot = acc.setdefault(tc.get("index") or 0, {"id": "", "name": "", "args": ""}) if tc.get("id"): slot["id"] = tc["id"] if tc.get("extra_content"): slot["extra"] = tc["extra_content"] fn = tc.get("function") or {} nm = fn.get("name") or "" if nm and nm != slot["name"]: # провайдер может повторять имя в каждом куске slot["name"] += nm slot["args"] += fn.get("arguments") or "" for ev in split.flush(): yield ev calls = [{"id": s["id"] or "call_%d" % i, "name": s["name"], "args": parse_args(s["args"]), "extra": s.get("extra")} for i, s in sorted(acc.items()) if s["name"]] if usage: yield ("usage", usage) yield ("final", {"calls": calls, "blocks": None}) def stream_anthropic(app, msgs): lvl = app.effort max_tokens = 8192 body = {"model": app.model, "system": app.system_prompt(), "messages": an_messages(msgs, (app.cfg.get("settings") or {}).get("token_saving", "auto")), "stream": True} b = ANTH_BUDGET.get(lvl) if b: body["thinking"] = {"type": "enabled", "budget_tokens": b} max_tokens = max(max_tokens, b + 4096) if tools_on(app, msgs): body["tools"] = an_tools() body["max_tokens"] = max_tokens headers = {"x-api-key": app.key, "anthropic-version": "2023-06-01", "Accept": "text/event-stream"} resp = http_open(app.base.rstrip("/") + "/messages", headers, body) ACTIVE["resp"] = resp ACTIVE_ALL.add(resp) u_in = u_out = think_chars = 0 blocks, cur = [], None with resp: for ev in sse(resp): t = ev.get("type") if t == "message_start": u = ev["message"].get("usage", {}) u_in = sum(u.get(k, 0) or 0 for k in ("input_tokens", "cache_read_input_tokens", "cache_creation_input_tokens")) elif t == "content_block_start": cb = ev["content_block"] bt = cb.get("type") if bt == "text": cur = {"type": "text", "text": cb.get("text", "")} elif bt == "thinking": cur = {"type": "thinking", "thinking": "", "signature": ""} elif bt == "tool_use": cur = {"type": "tool_use", "id": cb["id"], "name": cb["name"], "input": ""} else: cur = None elif t == "content_block_delta" and cur is not None: d = ev.get("delta", {}) dt = d.get("type") if dt == "text_delta": cur["text"] += d.get("text", "") yield ("text", d.get("text", "")) elif dt == "thinking_delta": cur["thinking"] += d.get("thinking", "") think_chars += len(d.get("thinking", "")) yield ("think", d.get("thinking", "")) elif dt == "signature_delta": cur["signature"] = d.get("signature", "") elif dt == "input_json_delta": cur["input"] += d.get("partial_json", "") elif t == "content_block_stop" and cur is not None: if cur["type"] == "tool_use": cur["input"] = parse_args(cur["input"]) keep = (cur["type"] == "text" and cur["text"]) or \ (cur["type"] == "thinking" and cur["signature"]) or cur["type"] == "tool_use" if keep: blocks.append(cur) cur = None elif t == "message_delta": u_out = ev.get("usage", {}).get("output_tokens", u_out) elif t == "error": raise ApiError(ev.get("error", {}).get("message", "ошибка Anthropic")) calls = [{"id": bl["id"], "name": bl["name"], "args": bl["input"]} for bl in blocks if bl["type"] == "tool_use"] yield ("usage", {"in": u_in, "out": u_out, "think": int(think_chars / 3.5), "cost": None}) yield ("final", {"calls": calls, "blocks": blocks}) # ════════════════════════════════════════════════════════════════════════════ # РЕНДЕР MARKDOWN (потоковый) # ════════════════════════════════════════════════════════════════════════════ class MD: def __init__(self): self.ls = True self.lb = "" self.in_code = False self.code_bol = True self.fence = False self.bold = self.ic = self.hdr = self.star = False def w(self, s): out(s) # через out(), чтобы правая панель перерисовывалась def style(self): s = RESET if self.hdr: s += "\x1b[1m" + fg(PAL["accent"]) if self.bold: s += "\x1b[1m" if self.ic: s += fg(PAL["accent2"]) return s def feed(self, text): for ch in text: self._ch(ch) sys.stdout.flush() def _resolve(self): self.ls = False lb, self.lb = self.lb, "" for ch in lb: self._body(ch) def _ch(self, ch): if self.fence: if ch == "\n": self.fence = False lang = self.lb[3:].strip() if not self.in_code: self.in_code = True self.code_bol = True self.w(" " + c("dim", "┌─ ") + c("muted", lang or "code") + "\n") else: self.in_code = False self.w(" " + c("dim", "└─") + "\n") self.ls, self.lb = True, "" else: self.lb += ch return if not self.ls: self._body(ch) return if ch == "\n": self._resolve() self._body("\n") return self.lb += ch lb = self.lb if lb in ("`", "``"): return if lb == "```": self.fence = True return if self.in_code: self._resolve() return if lb[0] in "-*": if len(lb) == 1: return if lb[1] == " " and len(lb) == 2: self.w(c("accent", " • ")) self.lb, self.ls = "", False return self._resolve() return if lb[0] == "#": if set(lb) == {"#"}: return if lb.rstrip("#") == "" or lb[-1] != " " or set(lb[:-1]) != {"#"}: self._resolve() return self.hdr = True self.w(self.style()) self.lb, self.ls = "", False return self._resolve() def _body(self, ch): if self.in_code: if ch == "\n": self.w(RESET + "\n") self.ls, self.code_bol = True, True return if self.code_bol: self.w(" " + c("dim", "│ ") + fg(PAL["info"])) self.code_bol = False self.w(ch) return if ch == "\n": self.w(RESET + "\n") self.bold = self.ic = self.hdr = self.star = False self.ls = True return if ch == "`": if self.star: self.w("*") self.star = False self.ic = not self.ic self.w(self.style()) return if self.ic: self.w(ch) return if ch == "*": if self.star: self.star = False self.bold = not self.bold self.w(self.style()) else: self.star = True return if self.star: self.star = False self.w("*") self.w(ch) def finish(self): self.fence = False if self.ls and self.lb: self._resolve() if self.star: self.w("*") self.star = False self.w(RESET) if not self.ls and not (self.in_code and self.code_bol): self.w("\n") sys.stdout.flush() # ════════════════════════════════════════════════════════════════════════════ # ОТМЕНА (Esc), АГЕНТЫ, ПРИЛОЖЕНИЕ # ════════════════════════════════════════════════════════════════════════════ CANCEL = threading.Event() # ставится при Esc / Ctrl+C во время ответа ACTIVE = {"resp": None} # текущее HTTP-соединение, его можно оборвать def force_close(resp): """Обрывает соединение, чтобы блокирующее чтение (стрим, загрузка страницы) сразу вернулось.""" if resp is None: return try: sock = resp.fp.raw._sock own = False except Exception: sock, own = None, False if sock is None: try: sock = socket.socket(fileno=os.dup(resp.fp.fileno())) own = True except Exception: return try: sock.shutdown(socket.SHUT_RDWR) except Exception: pass if own: try: sock.close() except Exception: pass class EscWatcher: """Пока модель отвечает, слушает Esc и отменяет запрос.""" def __init__(self): self.stop_ev = threading.Event() self.th = None self.old = None def start(self): if self.th or not sys.stdin.isatty(): return self.stop_ev.clear() if not IS_WIN: try: fd = sys.stdin.fileno() self.old = termios.tcgetattr(fd) new = termios.tcgetattr(fd) new[3] &= ~(termios.ECHO | termios.ICANON) new[6][termios.VMIN] = 1 new[6][termios.VTIME] = 0 termios.tcsetattr(fd, termios.TCSADRAIN, new) except Exception: self.old = None self.th = threading.Thread(target=self._loop, daemon=True) self.th.start() def _loop(self): while not self.stop_ev.is_set(): try: if IS_WIN: if msvcrt.kbhit(): if read_key() == "esc": self._cancel() else: time.sleep(0.05) elif _sel.select([sys.stdin.fileno()], [], [], 0.1)[0]: if read_key() == "esc": self._cancel() except Exception: time.sleep(0.2) # ошибка чтения не должна навсегда отключить Esc def _cancel(self): CANCEL.set() for r in list(ACTIVE_ALL): force_close(r) force_close(ACTIVE["resp"]) def stop(self): if not self.th: return self.stop_ev.set() self.th.join(timeout=1) self.th = None if not IS_WIN and self.old is not None: try: termios.tcsetattr(sys.stdin.fileno(), termios.TCSADRAIN, self.old) except Exception: pass self.old = None def prov_key(cfg, p): return cfg["keys"].get(p["id"]) or (os.environ.get(p["env"], "") if p["env"] else "") def prov_base(cfg, p): return cfg["bases"].get(p["id"]) or p["base"] # Короткая версия TOOLS_RULES — экономим токены: зачем пересказывать то, что уже в TOOL_DEFS (схемы инструментов). TOOLS_RULES_SHORT = ( "У тебя есть инструменты FlynCode (см. схемы): fetch_url, web_search, download_file, read_file, list_files, grep, " "write_file, append_file, edit_file, bash, multi_edit, apply_patch, move_path, delete_path, run_script, git, todo_write, screenshot, load_skill, create_agent, ask_agent, ask_user (опрос пользователя с вариантами).\n" "Правки — кусками (сначала read_file, потом edit_file/multi_edit); большой файл — частями write_file+append_file.\n" "В чате код целиком не выводи, если он записан в файл. Если просят открыть сайт/файл/найти — ВЫЗОВИ инструмент, не говори «не могу».\n" "Не выдумывай факты и имена файлов/функций — сначала открой/найди через инструмент.") TOOLS_RULES = ( "У тебя ЕСТЬ доступ к инструментам FlynCode, они подключены к этому чату:\n" "- fetch_url(url) — открыть сайт или страницу и прочитать текст;\n" "- web_search(query) — поиск в интернете;\n" "- download_file(url, save_to) — скачать файл (нужно разрешение пользователя);\n" "- read_file, list_files, grep — читать и искать файлы; write_file, edit_file — правка (с разрешением);\n" "- bash(command) — выполнить команду, например тесты или запуск (с разрешением).\n" "- multi_edit(path, edits) — несколько правок одного файла разом; apply_patch(patch) — diff по нескольким файлам " "(формат git diff, с разрешением); move_path, delete_path — перемещение и удаление (с разрешением).\n" "- run_script(path) — запуск .py/.js/.ts/.sh: среду выбирает сам (Bun, Node, Python, bash); " "git(args) — только чтение истории и состояния.\n" "- todo_write(items) — для сложной задачи заведи список шагов и отмечай выполненные.\n" "- ask_user(questions) — ОПРОС: если не хватает данных для хорошего результата (стиль дизайна, цвета, формат, " "объём, выбор подхода), задай вопрос. Предлагай 2–4 конкретных варианта, которые придумал сам по контексту, " "и allow_custom=true, чтобы пользователь мог написать свой. Не спрашивай то, что понятно из задачи, " "и не больше 4 вопросов за раз. Если пользователь пропустил вопрос — реши разумно и скажи, что выбрал.\n" "- create_agent(name, role, model?) и ask_agent(name, task) — если задача сложная и делится на роли " "(например, @Код пишет код, @Идейщик предлагает идеи, @Дизайнер делает интерфейс), создай агентов и поручай им части. " "Агенты работают по очереди и не мешают друг другу; итог собери сам и ответь пользователю. " "Для простых задач агентов не создавай.\n" "- screenshot(path или url) — снимок страницы, чтобы проверить вёрстку глазами. Используй после правок интерфейса.\n" "ПРАВКИ КУСКАМИ: не переписывай файл целиком ради небольшого изменения. Чтобы исправить баг или добавить " "функцию в уже написанный файл, сначала прочитай нужный участок (read_file с offset/limit), затем меняй только " "нужный фрагмент через edit_file или multi_edit. write_file — для нового файла или когда файл меняется почти весь.\n" "БОЛЬШОЙ ФАЙЛ ПИШИ ПО ЧАСТЯМ: больше примерно 150 строк — сначала write_file с первой частью (импорты, каркас, " "первые функции), потом добавляй блоки через append_file, по одному смысловому куску за вызов.\n" "В ЧАТЕ код целиком не выводи, если он уже записан в файл: покажи только изменённый фрагмент и коротко, что " "и зачем поменял.\n" "После твоих правок FlynCode сам проверяет синтаксис и теги. Если придёт сообщение «[Автопроверка…]», исправь найденное.\n" "Если пользователь просит открыть сайт, найти информацию или проверить файлы — ВЫЗОВИ нужный инструмент. " "Никогда не говори, что не можешь открыть сайт или выполнить действие, не попробовав инструмент. " "Если инструмент вернул ошибку или отказ — честно скажи об этом и предложи альтернативу. " "Не выдумывай содержимое страниц и файлов: сначала открой их.") ACCURACY_RULES = ("Точность важнее скорости: не выдумывай факты, цифры, ссылки, названия функций и файлов. " "Если не уверен — прямо скажи об этом и отдели проверенное от предположений. " "Для данных, которые могут устареть (цены, версии, новости, даты, состояние сайтов), " "опирайся на поиск или открытие страниц, а не на память.") VERIFY_TEMPLATE = ("Ты проверяешь ответ ИИ-ассистента перед тем, как его покажут пользователю. Инструментов у тебя нет.\n\n" "Вопрос пользователя:\n%s\n\n%sОтвет для проверки:\n%s\n\n" "Найди: фактические ошибки; выдуманные детали (ссылки, цифры, даты, имена, названия функций и файлов); " "противоречия с результатами инструментов; уверенные утверждения, которые ничем не подтверждены.\n" "Не придирайся к стилю и не меняй то, что верно. Ничего нового от себя не добавляй.\n" "Если ошибок нет — ответь ровно одним словом: OK\n" "Если есть — первой строкой напиши «ИСПРАВЛЕНО: » и кратко, что именно исправлено, " "затем с новой строки выдай ИСПРАВЛЕННЫЙ ответ целиком, в том же языке и формате. " "То, что проверить нельзя, но что выглядит правдоподобно, помечай словами «не проверено».") COMPACT_KEEP_TURNS = 2 # сколько последних реплик пользователя оставлять дословно COMPACT_TEMPLATE = ("Ниже история диалога с ИИ-ассистентом. Сожми её в сводку, по которой можно продолжить работу " "без потери смысла.\n" "Сохрани: цель и задачи пользователя; принятые решения и договорённости; имена файлов, путей, " "функций, чисел и ссылок; что уже сделано и с каким результатом; открытые вопросы и ошибки. " "Если в истории есть прежняя сводка — учти её и перенеси важное.\n" "Не добавляй ничего от себя и не оценивай. Пиши коротко, пунктами, не больше 700 слов.\n\n" "История:\n%s") # Сжатие посреди задачи (когда контекст почти полон, а работа ещё не закончена) MIDRUN_AT = 0.8 # при каком заполнении контекста сжимать прямо посреди задачи MIDRUN_KEEP_STEPS = 3 # сколько последних шагов модели оставлять дословно MIDRUN_TEMPLATE = ("Ниже ход работы ИИ-ассистента над задачей: что ему поручили, какие инструменты он вызывал " "и что получил. Работа прервана, потому что контекст почти заполнен. Сожми ход работы в сводку, " "по которой можно продолжить без потерь.\n" "Сохрани: что уже сделано и с каким результатом; какие файлы созданы или изменены и как; найденные " "ошибки и их статус; имена файлов, путей, функций, чисел и ссылок; что ещё осталось сделать. " "Не добавляй ничего от себя. Пиши коротко, пунктами, не больше 800 слов.\n\n" "История:\n%s") def msgs_tokens(msgs): """Грубая оценка размера истории в токенах (без служебных блоков модели).""" text = json.dumps([{k: v for k, v in m.items() if k not in ("blocks", "images")} for m in msgs], ensure_ascii=False, default=str) return est_tokens(text) + 1600 * sum(len(m.get("images") or []) for m in msgs) class AgentCtx: """Параметры одного агента. Тот же интерфейс, что у App, поэтому стриминг общий.""" def __init__(self, app, a): self.p = next(x for x in PROVIDERS if x["id"] == a["pid"]) self.key = prov_key(app.cfg, self.p) self.base = prov_base(app.cfg, self.p) self.model = a["model"] self.effort = levels_for(self.p["id"], self.model)[0] self.cfg = {"settings": {"tools": a.get("tools", True)}} self.name = a["name"] self.role = a.get("role", "") self.cwd = app.cwd def system_prompt(self): rules = (TOOLS_RULES if self.cfg["settings"]["tools"] else "") + "\n" + ACCURACY_RULES return ("Ты — агент «%s» в команде FlynCode. %s\n%s\n" "Сегодня: %s. Модель: %s. Рабочая папка: %s." % ( self.name, self.role, rules, datetime.now().strftime("%Y-%m-%d"), self.model, self.cwd)) class App: def __init__(self): self.cfg = load_cfg() set_theme(self.cfg["settings"]["theme"]) SOUND["on"] = self.cfg["settings"].get("sound", True) NOTIFY["on"] = self.cfg["settings"].get("notify", True) S = self.cfg["settings"] P = dict(DEFAULT_PERMS, **S.get("perms", {})) if S.get("perms_v") != 3: P.update({"fetch_url": "allow", "web_search": "allow", "download_file": "ask"}) S["perms_v"] = 3 S["perms"] = P self.p = None self.key = "" self.base = "" self.models = [] self.model = "" self.effort = "auto" self.messages = [] self.history = [] self.cwd = os.getcwd() self.notes = list(PROVIDER_NOTES) self.chat_id = self.cfg.get("chats_current") or _new_id() self.cfg["chats_current"] = self.chat_id save_cfg(self.cfg) data = read_chat_file(self.chat_id) if data is None and os.path.exists(_hist_path(self.chat_id)): broken = _hist_path(self.chat_id) + ".broken" try: os.replace(_hist_path(self.chat_id), broken) self.notes.append("Файл чата повреждён, его копия сохранена: " + broken) except OSError: pass data = data or {} self.messages = list(data.get("messages", [])) self.plan = [t for t in data.get("plan", []) if isinstance(t, dict) and t.get("text")] APP_REF["app"] = self SIDEBAR["on"] = bool(self.cfg["settings"].get("sidebar", True)) self.allowed = set() self.allow_all = False self.todos = [] self.pending_images = [] # фото, которые уйдут с следующим сообщением self.watch = EscWatcher() self._in_task = False self.tot = {"in": 0, "out": 0, "think": 0, "cost": 0.0, "time": 0.0, "turns": 0} self.mode = self.cfg["settings"].get("mode", "build") if self.mode not in MODES: self.mode = "build" # ---------- сохранение чата def persist_chat(self): """Пишет текущий чат на диск (сообщения, план). Вызывается после каждого действия и при выходе.""" if getattr(_tl, "buf", None) is not None: # параллельный агент: сохраняет главный поток return ok = save_chat(self.chat_id, self.messages, self.cwd, plan=self.plan) if not ok and not SAVE_ERROR["shown"]: SAVE_ERROR["shown"] = True println(" " + c("err", "✗ Не удалось сохранить чат: ") + c("muted", SAVE_ERROR["msg"])) self.cfg["chats_current"] = self.chat_id save_cfg(self.cfg) def start_autosave(self): """Фоновое сохранение чата раз в 10 секунд, если он изменился: защищает от закрытия окна, зависания и потери связи во время долгого ответа. Пишет только файл чата, без настроек.""" def loop(): last = None while True: time.sleep(10) try: msgs = list(self.messages) tail = str(msgs[-1].get("content", "")) if msgs else "" sig = (self.chat_id, len(msgs), len(tail), tail[-40:]) if sig != last: if save_chat(self.chat_id, msgs, self.cwd, plan=list(self.plan)): last = sig except Exception: pass threading.Thread(target=loop, daemon=True).start() def new_chat(self, keep_old=True): """Единая точка создания нового чата: старый сохраняется (если нужно), сообщения и план обнуляются, новый чат сразу получает свой файл. План и контекст не переходят между чатами.""" if keep_old: self.persist_chat() self.messages = [] self.plan = [] self.pending_images = [] self.chat_id = _new_id() self.cfg["chats_current"] = self.chat_id save_cfg(self.cfg) self.persist_chat() # ---------- режимы: Build / Plan / Chat def set_mode(self, key): if key not in MODES: return self.mode = key self.cfg["settings"]["mode"] = key save_cfg(self.cfg) def cycle_mode(self): i = MODE_ORDER.index(self.mode) if self.mode in MODE_ORDER else 0 self.set_mode(MODE_ORDER[(i + 1) % len(MODE_ORDER)]) def mode_bar(self): """Линия слева у строки ввода: её цвет задаёт режим (Build синий, Plan оранжевый, Chat серый).""" return rgbtxt(MODE_COLOR[self.mode], "┃", bold=True) + " " def composer_footer(self): """Одна строка под полем: название режима цветом режима, модель и подсказка справа.""" name = MODES[self.mode][0] mode_s = rgbtxt(MODE_COLOR[self.mode], name, bold=True) prov = self.p["name"].split(" (")[0] if self.p else "" avail = term_w() - 3 model = self.model or "нет модели" full = "tab режим ctrl+p команды" for hint_t, show_prov in ((full, True), (full, False), ("tab режим", False)): pv = (" (" + prov + ")") if (show_prov and prov) else "" room = avail - vlen(name) - 2 - vlen(hint_t) - 2 - len(pv) if room >= 8: left = mode_s + " " + c("text", clip(model, room)) + (c("muted", pv) if pv else "") gap = avail - vlen_ansi(left) - vlen(hint_t) return [left + " " * max(2, gap) + c("muted", hint_t)] return [mode_s] def cmd_mode(self, arg): a = (arg or "").strip().lower() alias = {"build": "build", "билд": "build", "plan": "plan", "план": "plan", "chat": "chat", "чат": "chat"} if a in alias: self.set_mode(alias[a]) elif a: println(" " + c("err", "Неизвестный режим: " + a) + c("muted", " (build / plan / chat)")) println() return else: self.cycle_mode() name, desc = MODES[self.mode] println(" " + c("ok", "✓ Режим: ") + c("text", name) + c("muted", " — " + desc)) println() # ---------- общее def system_prompt(self): ts = (self.cfg.get("settings") or {}).get("token_saving", "auto") use_short = False if ts == "always": use_short = True elif ts == "auto": try: lim = (self.cfg.get("settings") or {}).get("compact_at", 60000) or 0 used = msgs_tokens(self.messages) if hasattr(self, "messages") else 0 use_short = bool(lim and used > lim * 0.5) except Exception: use_short = False if self.cfg["settings"]["tools"]: tools = TOOLS_RULES_SHORT if use_short else TOOLS_RULES else: tools = "Инструменты выключены в настройках: доступа к сайтам и файлам нет, отвечай из своих знаний." skills = self.skills_block() tools = tools + "\n\n" + ACCURACY_RULES if self.plan: tools = tools + "\n\nСохранённый план этого чата (держись его и по ходу отмечай шаги через todo_write):\n" + \ "\n".join("[%s] %s" % ("x" if t["done"] else " ", t["text"]) for t in self.plan) if MODE_RULES.get(self.mode): tools = tools + "\n\n" + MODE_RULES[self.mode] return ("%s\n\n%s%s\n\nСегодня: %s. Модель: %s. Рабочая папка: %s. ОС: %s." % ( self.cfg["settings"]["system"], tools, ("\n\n" + skills) if skills else "", datetime.now().strftime("%Y-%m-%d"), self.model, self.cwd, "Windows" if IS_WIN else sys.platform)) def levels(self): for m in self.models: if m["id"] == self.model: return m.get("levels") or ["auto"] return levels_for(self.p["id"], self.model) def save_last(self): self.cfg["last"]["provider"] = self.p["id"] self.cfg["last"].setdefault("models", {})[self.p["id"]] = {"model": self.model, "effort": self.effort} save_cfg(self.cfg) def refresh_models(self): self.models = fetch_models(self.p, self.key, self.base) save_cache(self.p["id"], self.models) return self.models def ask(self, label, mask=False, initial="", placeholder=""): println(" " + label) return read_line(" › ", mask=mask, initial=initial, esc_cancel=True, placeholder=placeholder) # ---------- подключение def pick_provider(self): items = [] rec = [p for pid in RECOMMENDED for p in PROVIDERS if p["id"] == pid] rest = [p for p in PROVIDERS if p["id"] not in RECOMMENDED] for p in rec + rest: has = bool(self.cfg["keys"].get(p["id"])) or bool(p["env"] and os.environ.get(p["env"])) star = p["id"] in RECOMMENDED items.append({"label": ("★ " if star else "") + p["name"], "desc": ("рекомендуем · " if star else "") + p["desc"] + (" ✓ ключ есть" if has else ""), "value": p["id"]}) pid = select("Выбери провайдера", items, current=self.cfg["last"].get("provider"), searchable=True, hint="печатай, чтобы найти · ↑↓ выбор · Enter") return next((p for p in PROVIDERS if p["id"] == pid), None) def ensure_key(self, p): if p.get("nokey") or prov_key(self.cfg, p): return True println(" " + c("muted", "Нужен API-ключ для %s: " % p["name"]) + link(p["url"])) v = self.ask(c("text", "Вставь API-ключ", bold=True), mask=True) if not v or not v.strip(): return False self.cfg["keys"][p["id"]] = v.strip() save_cfg(self.cfg) return True def connect(self, p): pid = p["id"] base = self.cfg["bases"].get(pid) or p["base"] if p.get("ask_base"): v = self.ask(c("text", "Адрес API ") + c("muted", "(например http://localhost:1234/v1)"), initial=base) if not v or not v.strip(): return False base = v.strip().rstrip("/") self.cfg["bases"][pid] = base key = "" if not p.get("nokey"): env = os.environ.get(p["env"], "") if p["env"] else "" saved = self.cfg["keys"].get(pid) or env use = False if saved: mask = saved[:4] + "…" + saved[-4:] if len(saved) > 10 else "••••" ch = select("API-ключ " + mask, [ {"label": "Использовать его", "desc": "", "value": "use"}, {"label": "Ввести новый", "desc": "", "value": "new"}], searchable=False, hint=p["name"]) if ch is None: return False use = ch == "use" key = saved if use else None while key is None: if p["url"]: println(" " + c("muted", "Получить ключ: ") + link(p["url"])) v = self.ask(c("text", "Вставь API-ключ для %s" % p["name"], bold=True) + (c("muted", " (можно пусто)") if p.get("optkey") else ""), mask=True, placeholder="API key…") if v is None: return False v = v.strip() if v or p.get("optkey"): key = v else: println(" " + c("err", "Ключ не может быть пустым")) self.p, self.key, self.base = p, key, base try: with_spinner("Проверяю подключение…", self.refresh_models) except ApiError as e: println(" " + c("err", "✗ " + clip(str(e), term_w() - 6))) ch = select("Что делаем?", [ {"label": "Ввести ключ заново", "desc": "", "value": "retry"}, {"label": "Продолжить со встроенным списком", "desc": "ключ не проверен", "value": "go"}, {"label": "Назад", "desc": "", "value": "back"}], searchable=False) if ch == "retry": self.cfg["keys"].pop(pid, None) return self.connect(p) if ch != "go": return False self.models = load_cache(pid) or fallback_models(pid) if key: self.cfg["keys"][pid] = key save_cfg(self.cfg) println(" " + c("ok", "✓ Подключено · ") + c("muted", "моделей: %d" % len(self.models))) return True def pick_model(self, query=""): items = [] for m in self.models: id_part = m["id"] parts = [] if m.get("ctx"): parts.append("%s" % fmt_tok(m["ctx"])) lv = m.get("levels") or ["auto"] if len(lv) > 1: parts.append("◆" + "/".join(x for x in lv if x != "auto")) if m.get("vision"): parts.append("👁") if m.get("think"): parts.append("💭") if m.get("price") and any(m["price"]): parts.append("$%.2f/$%.2f" % tuple(m["price"])) desc = " · ".join(parts) if parts else "" items.append({"label": id_part, "desc": desc, "value": m["id"]}) if query: items = [it for it in items if query.lower() in it["label"].lower() or query.lower() in it["desc"].lower()] or items mid = select("Модель · %s · %d" % (self.p["name"], len(items)), items, current=self.model or None, custom=True) if mid is None: return False self.model = mid if self.effort not in self.levels(): self.effort = self.levels()[0] return True def pick_effort(self): lv = self.levels() items = [{"label": x, "desc": LEVEL_DESC.get(x, ""), "value": x} for x in lv] v = select("Насколько думать · %s" % self.model, items, current=self.effort, searchable=False) if v: self.effort = v def pick_after_connect(self): last = (self.cfg["last"].get("models") or {}).get(self.p["id"], {}) self.model = last.get("model") or (self.models[0]["id"] if self.models else "") self.effort = last.get("effort") or "auto" if not self.pick_model(): return False if len(self.levels()) > 1: self.pick_effort() self.save_last() return True def onboarding(self): while True: clear_screen() show_logo() p = self.pick_provider() if p is None: println(" " + c("muted", "Пока! 👋")) sys.exit(0) if self.connect(p) and self.pick_after_connect(): return # ---------- разрешения def allowed_now(self, name, args): pol = self.cfg["settings"]["perms"].get(name, "ask") if pol == "deny": return False if pol == "allow" or self.allow_all or name in self.allowed: return True if getattr(_tl, "buf", None) is not None: return False # параллельный агент не может спрашивать — отказ set_status("perm") self.watch.stop() # на время вопроса Esc означает «нет» try: ch = select("Разрешить действие?", [ {"label": "Да, один раз", "desc": "", "value": "once"}, {"label": "Да, «%s» в этой сессии" % name, "desc": "", "value": "tool"}, {"label": "Да, всё в этой сессии", "desc": "полный доступ к инструментам", "value": "all"}, {"label": "Нет", "desc": "отказать модели", "value": "no"}, ], searchable=False, hint=describe(name, args)) finally: self.watch.start() if ch is None: CANCEL.set() # Esc на вопросе — остановить весь ответ return False if ch == "no": return False if ch == "tool": self.allowed.add(name) if ch == "all": self.allow_all = True return True def perm_menu(self): P = self.cfg["settings"]["perms"] names = {"ask": "спрашивать", "allow": "разрешено", "deny": "запрещено"} nxt = {"ask": "allow", "allow": "deny", "deny": "ask"} def preset(kind): for t in TOOL_DEFS: if kind == "ask": P[t["name"]] = "ask" if t["ask"] else "allow" elif kind == "allow": P[t["name"]] = "allow" else: P[t["name"]] = "deny" if t["name"] in RO_DENY else "allow" self.allow_all = False self.allowed.clear() while True: items = [ {"label": "Пресет: спрашивать всё", "desc": "безопасный режим", "value": "@ask"}, {"label": "Пресет: полный доступ", "desc": "всё без вопросов", "value": "@allow"}, {"label": "Пресет: только чтение", "desc": "запись, правки, bash и сеть запрещены", "value": "@ro"}, ] items += [{"label": t["name"], "desc": names[P[t["name"]]], "value": t["name"]} for t in TOOL_DEFS] v = select("Разрешения инструментов", items, searchable=False, hint="Enter — сменить: спрашивать → разрешено → запрещено") if v is None: save_cfg(self.cfg) return if v in ("@ask", "@allow", "@ro"): preset({"@ask": "ask", "@allow": "allow", "@ro": "ro"}[v]) else: P[v] = nxt[P[v]] save_cfg(self.cfg) # ---------- агенты def ensure_agents(self): return self.cfg.setdefault("agents", []) def agents_menu(self): while True: A = self.ensure_agents() items = [{"label": "+ Добавить агента", "desc": "имя, провайдер, модель, роль", "value": "@add"}] for i, a in enumerate(A): pname = next((p["name"] for p in PROVIDERS if p["id"] == a["pid"]), a["pid"]) desc = "%s · %s" % (pname.split(" (")[0], a["model"]) if a.get("role"): desc += " · " + clip(a["role"], 28) if not a.get("on", True): desc = "[выкл] " + desc items.append({"label": "@" + a["name"], "desc": desc, "value": i}) v = select("Агенты FlynCode", items, searchable=False, hint="Enter — открыть · в чате: @имя текст · /debate тема · /tips — подсказки") if v is None: save_cfg(self.cfg) return if v == "@add": a = self.agent_form(None) if a: A.append(a) else: self.agent_actions(v) save_cfg(self.cfg) def agent_actions(self, i): A = self.ensure_agents() a = A[i] items = [ {"label": "Изменить", "desc": "имя, провайдер, модель, роль", "value": "edit"}, {"label": "Включить / выключить", "desc": "сейчас: " + ("вкл" if a.get("on", True) else "выкл"), "value": "on"}, {"label": "Инструменты для агента", "desc": "сейчас: " + ("вкл" if a.get("tools", True) else "выкл"), "value": "tools"}, {"label": "Удалить", "desc": "", "value": "del"}, ] v = select("@" + a["name"], items, searchable=False, hint="Enter — выбрать") if v == "edit": na = self.agent_form(a) if na: A[i] = na elif v == "on": a["on"] = not a.get("on", True) elif v == "tools": a["tools"] = not a.get("tools", True) elif v == "del": del A[i] def agent_form(self, old): old = old or {} println() name = self.ask(c("text", "Имя агента", bold=True) + c("muted", " (например Код, Поиск, Чат)"), initial=old.get("name", "")) if not name or not name.strip(): return None name = name.strip().replace(" ", "_").lstrip("@") p = self.pick_provider() if p is None or not self.ensure_key(p): return None key, base = prov_key(self.cfg, p), prov_base(self.cfg, p) try: models = with_spinner("Загружаю модели…", lambda: fetch_models(p, key, base)) except ApiError as e: println(" " + c("warn", "⚠ " + clip(str(e), term_w() - 8)) + c("muted", " — введи ID вручную")) models = fallback_models(p["id"]) items = [{"label": m["id"], "desc": (fmt_tok(m["ctx"]) + " ctx") if m.get("ctx") else "", "value": m["id"]} for m in models] same = old.get("pid") == p["id"] model = select("Модель для @%s" % name, items, current=old.get("model") if same else None, custom=True) if not model: return None role = self.ask(c("text", "Роль / инструкция", bold=True) + c("muted", " (Enter — пропустить)"), initial=old.get("role", "")) if role is None: return None return {"name": name, "pid": p["id"], "model": model, "role": role.strip(), "on": old.get("on", True), "tools": old.get("tools", True)} def chat_context(self, limit=3000): lines = [] for m in self.messages[-10:]: if m["role"] in ("user", "assistant") and m.get("content") and not m.get("tool_calls") \ and not m.get("summary"): who = "Пользователь" if m["role"] == "user" else "FlynCode" lines.append("%s: %s" % (who, m["content"][:600])) return "\n".join(lines)[-limit:] def agent_run(self, a, task): """Ход агента по поручению главного агента. Возвращает текст ответа, в чат ничего не пишет.""" ctx = AgentCtx(self, a) body = "Задача от главного агента:\n" + task return self.run_guarded(ctx, [{"role": "user", "content": body}], "@%s · %s" % (a["name"], a["model"]), notify=False) def call_agent(self, a, text): ctx = AgentCtx(self, a) recent = self.chat_context() body = ("Последние реплики чата:\n%s\n\n" % recent if recent else "") + "Запрос: " + text ans = self.run_guarded(ctx, [{"role": "user", "content": body}], "@%s · %s" % (a["name"], a["model"])) if ans and ans.strip(): self.messages.append({"role": "user", "content": "[Ответ агента @%s на «%s»]:\n%s" % ( a["name"], text, ans.strip())}) def mention(self, line): stages, task, err = self.parse_team(line) if err: println(" " + c("err", err)) println() return if stages: self.print_user(line) self.run_task(lambda: self.run_team(stages, task)) return body = line[1:] A = [a for a in self.cfg.get("agents", []) if a.get("on", True)] hit = None for a in sorted(A, key=lambda x: -len(x["name"])): n = len(a["name"]) if body.lower().startswith(a["name"].lower()) and (len(body) == n or body[n] == " "): hit = a break if hit is None: tok = body.split(None, 1) sk = find_skill(self.cwd, tok[0].rstrip(".,;:!?)")) if tok else None if sk is not None: self.run_skill(sk, tok[1].strip() if len(tok) > 1 else "") return found = ", ".join("@" + n for n in list_skills(self.cwd)) or "ни одного" println(" " + c("err", "✗ Нет агента или скилла с таким именем: «%s»" % (tok[0] if tok else ""))) if tok: close = difflib.get_close_matches(tok[0].lower(), list(list_skills(self.cwd)), n=1, cutoff=0.6) if close: println(" " + c("warn", "Может, ты имел в виду @%s?" % close[0])) println(" " + c("muted", "Найденные скиллы: " + found)) println(" " + c("muted", "Открой /skills — там видно, какие папки проверяются и что в них найдено.")) println() return text = body[len(hit["name"]):].strip() if not text: println(" " + c("muted", "Напиши запрос после @%s" % hit["name"])) println() return self.print_user(line) self.run_task(lambda: self.call_agent(hit, text)) def parse_team(self, line): """@A + @B > @C задача → (этапы, задача, ошибка). Этапы идут по «>», агенты в этапе — по «+». Если это не команда из нескольких агентов — (None, None, None) и работает обычный @агент.""" A = [a for a in self.cfg.get("agents", []) if a.get("on", True)] rest = line.strip() def find(s): for a in sorted(A, key=lambda x: -len(x["name"])): n = a["name"] if s[1:1 + len(n)].lower() == n.lower() and (len(s) == 1 + len(n) or s[1 + len(n)] in " \t+>"): return a, 1 + len(n) return None, 0 if not rest.startswith("@"): return None, None, None stages, stage, first = [], [], True while rest.startswith("@"): a, used = find(rest) if a is None: if first: return None, None, None # не агент — пусть разбирает обычный путь (скиллы) return None, None, "✗ Нет агента с таким именем: «%s»" % rest.split()[0] first = False stage.append(a) rest = rest[used:].lstrip() if rest.startswith("+") or rest.startswith(">"): sep = rest[0] rest = rest[1:].lstrip() if sep == ">": stages.append(stage) stage = [] if not rest.startswith("@"): return None, None, "✗ После «%s» нужен агент" % sep else: break stages.append(stage) task = rest.strip() if len(stages) == 1 and len(stages[0]) == 1: return None, None, None # один агент без «+» и «>» — обычный путь if not task: return None, None, "✗ Напиши задачу после агентов" return stages, task, None def run_parallel(self, stage, body, stage_no): """Агенты одного этапа работают одновременно; их ответы выводятся блоками по порядку.""" names = ", ".join("@" + a["name"] for a in stage) println(" " + c("accent2", "◆ Параллельно", bold=True) + c("muted", " этап %d: %s" % (stage_no, names))) results = [(a["name"], None, "") for a in stage] def work(i, a): _tl.buf, _tl.quiet = [], True txt = None try: txt = self.run_guarded(AgentCtx(self, a), [{"role": "user", "content": body}], "@%s · %s" % (a["name"], a["model"]), notify=False) except Exception: pass results[i] = (a["name"], txt, "".join(_tl.buf)) t0 = time.time() sp = Spinner(lambda: ("Параллельно: %s · %s · Esc — отмена" % (names, fmt_dur(time.time() - t0)), "muted")) threads = [threading.Thread(target=work, args=(i, a), daemon=True) for i, a in enumerate(stage)] real = sys.stdout sys.stdout = _Router(real) try: for t in threads: t.start() sp.start() for t in threads: t.join() finally: sp.stop() sys.stdout = real for _, _, buf in results: real.write(buf) real.flush() println() return [(n, t) for n, t, _ in results] def run_team(self, stages, task): names = " → ".join(" + ".join("@" + a["name"] for a in st) for st in stages) println(" " + c("accent2", "◆ Команда", bold=True) + c("muted", " " + names)) println() recent = self.chat_context() log = [] for si, stage in enumerate(stages): if CANCEL.is_set(): break body = ("Последние реплики чата:\n%s\n\n" % recent if recent else "") + "Задача: " + task if log: # каждый следующий этап видит ответы всех предыдущих body += "\n\nРезультаты предыдущих этапов команды:\n" + "\n\n".join( "[@%s]: %s" % (n, t) for n, t in log) if len(stage) == 1: a = stage[0] txt = self.run_guarded(AgentCtx(self, a), [{"role": "user", "content": body}], "@%s · %s · этап %d" % (a["name"], a["model"], si + 1), notify=False) results = [(a["name"], txt)] else: results = self.run_parallel(stage, body, si + 1) got = [(n, t.strip()) for n, t in results if t and t.strip()] if not got: if not CANCEL.is_set(): println(" " + c("err", "✗ Этап %d не дал ответа — команда остановлена" % (si + 1))) println() break log += got if log: self.messages.append({"role": "user", "content": "[Команда агентов по задаче «%s»]\n%s" % ( task, "\n\n".join("[@%s]: %s" % (n, t) for n, t in log))}) if CANCEL.is_set(): set_result("stopped", "команда остановлена") elif log: set_result("done", "команда готова") play_sound("done") else: set_result("error", "команда не ответила") play_sound("error") println() def debate(self, topic, A, rounds): names = ", ".join("@" + a["name"] for a in A) println(" " + c("accent2", "◆ Дебаты", bold=True) + c("muted", " %s · раундов: %d" % (names, rounds))) println(" " + c("muted", "Тема: ") + c("text", topic)) println() log = [] for rnd in range(rounds): for a in A: if CANCEL.is_set(): break ctx = AgentCtx(self, a) ctx.cfg = {"settings": {"tools": False}} # в дебатах без инструментов said = "\n\n".join("[%s]: %s" % (n, t) for n, t in log[-10:]) or "(пока никто не высказался)" others = ", ".join(x["name"] for x in A if x is not a) prompt = ("Тема дискуссии: %s\n\nРаунд %d из %d. Ты — %s. Другие участники: %s.\n\n" "Реплики до тебя:\n%s\n\n" "Ответь кратко, до 120 слов. Выскажи свою позицию, прямо оспорь или поддержи " "конкретные тезисы других и не повторяй уже сказанное. Если согласен с большинством, " "так и скажи." % (topic, rnd + 1, rounds, a["name"], others, said)) txt = self.run_guarded(ctx, [{"role": "user", "content": prompt}], "@%s · %s" % (a["name"], a["model"]), notify=False) if txt and txt.strip(): log.append((a["name"], txt.strip())) if CANCEL.is_set(): break if CANCEL.is_set() or not log: return transcript = "\n\n".join("[%s]: %s" % (n, t) for n, t in log) judge = AgentCtx(self, {"name": "Судья", "pid": self.p["id"], "model": self.model, "role": "Ты беспристрастный судья дискуссии.", "tools": False}) req = ("Тема: %s\n\nДискуссия:\n%s\n\nПодведи итог: 1) в чём участники согласны; 2) в чём спорят и чья " "позиция аргументированнее; 3) конкретная рекомендация с обоснованием. До 200 слов." % (topic, transcript)) verdict = self.run_guarded(judge, [{"role": "user", "content": req}], "Решение судьи · " + self.model) if verdict and verdict.strip(): self.messages.append({"role": "user", "content": "[Дебаты по теме «%s»]\n%s\n\n[Итог судьи]\n%s" % ( topic, transcript, verdict.strip())}) def cmd_debate(self, arg): A = [a for a in self.cfg.get("agents", []) if a.get("on", True)] if len(A) < 2: println(" " + c("err", "✗ Для дебатов нужно минимум 2 включённых агента (/agents)")) println() return topic = arg or ((self.ask(c("text", "Тема дебатов", bold=True)) or "").strip()) if not topic: return rounds = int(self.cfg["settings"].get("rounds", 2)) self.run_task(lambda: self.debate(topic, A, rounds)) # ---------- чат def welcome_rows(self): """Блок приветствия: команды как в /help и советы. Выводится по центру экрана.""" cmds = [("/help", "все команды"), ("/model", "выбрать модель и провайдера"), ("/mode", "режим: Build · Plan · Chat"), ("/plan", "план этого чата"), ("/clear", "новый чат"), ("/settings", "настройки")] tips = [ "Tab переключает режим: Build делает, Plan составит план, Chat просто общается", "@имя вызывает агента или скилл", "Перетащи фото в окно или /paste — прикрепить картинку к сообщению", "Esc отменяет ответ, ctrl+p показывает все команды", ] def head(title): return c("text", title, bold=True) + " " + c("dim", "─" * 28) rows = [head("Команды"), ""] for name, desc in cmds: rows.append(c("accent", padv(name, 11)) + c("muted", desc)) rows += ["", head("Советы")] for tip in tips: for k, ln in enumerate(textwrap.wrap(tip, 60) or [tip]): rows.append(c("accent2", "• " if k == 0 else " ") + c("muted", ln)) return rows def banner(self): clear_screen() SCREEN.center = not self.messages show_logo() W = term_w() block = self.welcome_rows() bw = max(vlen_ansi(r) for r in block) pad = " " * max(0, (W - bw) // 2) for r in block: println(pad + r) println() def status_line(self): """Над полем ввода — только проблемы: остальное показано в правой панели.""" parts = [] bs = BUN_STATE.get("state") if bs == "installing": parts.append(c("warn", "ставлю Bun…")) elif bs == "error": parts.append(c("err", "Bun: " + clip(BUN_STATE.get("msg", ""), 40))) if self.pending_images: parts.append(c("accent2", "📎 фото: %d — уйдут с следующим сообщением (/attach clear — убрать)" % len(self.pending_images))) last = STATUS.get("last") if last and last["kind"] in ("error", "stopped"): parts.append(c("err" if last["kind"] == "error" else "warn", last["text"])) if parts: println(" " + c("dim", " · ").join(parts)) println() def print_user(self, text): """Сообщение пользователя: «›» и жирный текст, как в OpenCode.""" SCREEN.center = False inner = max(20, term_w() - 4) first = True for para in text.split("\n"): for ln in (textwrap.wrap(para, inner) or [""]): if first: println(c("text", "› ", bold=True) + c("text", ln, bold=True)) first = False else: println(" " + c("text", ln, bold=True)) println() def close_pending(self, msgs): """После отмены дописывает заглушки для вызовов инструментов без результата.""" ids = set() for m in reversed(msgs): if m["role"] == "tool": ids.add(m["tool_call_id"]) continue if m["role"] == "assistant" and m.get("tool_calls"): for tc in m["tool_calls"]: if tc["id"] not in ids: msgs.append({"role": "tool", "tool_call_id": tc["id"], "name": tc["name"], "content": "(прервано пользователем)"}) break if msgs and msgs[-1]["role"] != "assistant": msgs.append({"role": "assistant", "content": "(прервано пользователем)"}) def stream_step(self, ctx, msgs, t_step): show = self.cfg["settings"]["show_thinking"] gen = stream_anthropic if ctx.p["kind"] == "anthropic" else stream_openai st = {"think": False} begin_status("wait") sp = Spinner(status_label) md = MD() body, think, usage = "", "", None final = {"calls": [], "blocks": None} think_open, bol = False, True sp.start() wd = start_watchdog(sp) attempt = 0 while True: try: for kind, payload in gen(ctx, msgs): if kind == "think": think += payload st["think"] = True set_status("think") if show: if not think_open: sp.stop() think_open = True println(" " + c("muted", "💭 Размышление…", italic=True)) s = "" for ch in payload: if ch == "\n": s += "\n" bol = True continue if bol: s += " " bol = False s += "\x1b[3m" + fg(PAL["muted"]) + ch + RESET out(s) elif kind == "text": sp.stop() set_status("write") if think_open: out(("" if bol else "\n") + "\n") think_open = False body += payload md.feed(payload) elif kind == "usage": usage = payload elif kind == "final": final = payload break # успешно except ApiError as e: # если текст уже печатался — дальше не повторяем, чтобы не задвоить вывод if body or think: raise attempt += 1 transient = getattr(e, "transient", False) or e.code in (408, 425, 429) or (e.code is not None and 500 <= e.code < 600) if not transient or attempt > MAX_RETRIES: raise if CANCEL.is_set(): raise KeyboardInterrupt("отменено пользователем") delay = _retry_delay(attempt) err_short = str(e)[:120] out("\n") println(" " + c("warn", "⚠ Ошибка: ") + c("muted", err_short)) println(" " + c("muted", "Повтор %d/%d через %dс · Esc — отменить" % (attempt, MAX_RETRIES, int(delay)))) try: STATUS.update(detail="retry %d/%d · %s · %dс" % (attempt, MAX_RETRIES, err_short, int(delay))) except Exception: pass _wait_retry(status_fn=lambda a, t, er, d: None, attempt=attempt, total=MAX_RETRIES, err=err_short) body, think, usage = "", "", None final = {"calls": [], "blocks": None} think_open, bol = False, True try: md = MD() except Exception: pass sp.start() sp.stop() wd.set() md.finish() return body, think, final, usage def footer(self, model, acc, t0): dt = time.time() - t0 self.tot["in"] += acc["in"] self.tot["out"] += acc["out"] self.tot["think"] += acc["think"] self.tot["cost"] += acc["cost"] self.tot["time"] += dt self.tot["turns"] += 1 wave = "~" if acc["est"] else "" bits = [c("accent", "▣ ") + c("text", model), c("muted", "↑ %s%s" % (wave, fmt_tok(acc["in"]))), c("muted", "↓ %s%s" % (wave, fmt_tok(acc["out"])))] if acc["think"]: bits.append(c("accent2", "◆ %s%s думал" % (wave, fmt_tok(acc["think"])))) bits.append(c("muted", "⏱ " + fmt_dur(dt))) if acc["cost"]: bits.append(c("ok", "$%.4f" % acc["cost"])) println(" " + c("dim", " · ").join(bits)) println() def agent_loop(self, ctx, msgs, title): """Один ход агента: стрим, вызовы инструментов, автопроверка правок, повтор. Перед каждым запросом проверяет заполнение контекста и при необходимости сжимает память прямо посреди задачи. Возвращает текст ответа.""" acc = {"in": 0, "out": 0, "think": 0, "cost": 0.0, "est": False} t0 = time.time() println(" " + c("accent", "◆ ", bold=True) + c("muted", title)) S = self.cfg["settings"] limit = S.get("autocheck_rounds", 10) if S.get("autocheck", True) else 0 run = {"changed": set(), "touched": set(), "shots": [], "failed": False, "sig": None, "same": 0, "midrun_stop": False} anchor = max(0, len(msgs) - 1) # задача этого хода — последнее сообщение пользователя text, checks, after_check = "", 0, False for _ in range(MAX_STEPS + limit): new_anchor = self.midrun_compact(ctx, msgs, anchor, run) if new_anchor is not None: anchor = new_anchor body, think, final, usage = self.stream_step(ctx, msgs, time.time()) if usage is None: acc["est"] = True usage = {"in": msgs_tokens(msgs), "out": est_tokens(body + think), "think": est_tokens(think) if think else 0, "cost": None} for k in ("in", "out", "think"): acc[k] += usage.get(k) or 0 acc["cost"] += usage.get("cost") or 0 calls = final.get("calls") or [] msgs.append({"role": "assistant", "content": body, "tool_calls": calls, "blocks": final.get("blocks")}) # ответ «OK» на автопроверку в итоговый текст не попадает if not (after_check and OK_RE.fullmatch(body)): text += body if CANCEL.is_set(): break if calls: after_check = False for call in calls: self.run_call(call, msgs, run) if CANCEL.is_set(): break if run["shots"]: shots = [] if vision_ok(self.cfg, ctx.model, self.models): shots = [image_part(p) for p in run["shots"]] run["shots"].clear() if shots: msgs.append({"role": "user", "content": "Скриншот из инструмента screenshot. " "Посмотри и скажи, что в вёрстке выглядит криво.", "images": shots}) continue if after_check and OK_RE.fullmatch(body): run["failed"] = False # модель подтвердила, что всё в порядке after_check = False if limit and (run["changed"] or run["failed"]) and checks < limit: checks += 1 fb = self.autocheck(ctx, run, checks, limit) if fb: msgs.append(fb) after_check = True continue break self.footer(ctx.model, acc, t0) return text def verify_answer(self, ctx, msgs, body): """Самопроверка: второй проход без инструментов сверяет ответ с вопросом и результатами инструментов.""" if len(body) > 12000: println(" " + c("muted", "Ответ слишком длинный для самопроверки — показываю как есть")) println() return body start = max((i for i, m in enumerate(msgs) if m["role"] == "user"), default=0) q = msgs[start].get("content") if msgs else "" q = q if isinstance(q, str) else "" ev = "\n\n".join("[%s]\n%s" % (m.get("name") or "инструмент", str(m.get("content") or "")[:1200]) for m in msgs[start:] if m["role"] == "tool") prompt = VERIFY_TEMPLATE % (q[:2000], ("Результаты инструментов в этом ходе:\n%s\n\n" % ev) if ev else "", body) q_ctx = copy.copy(ctx) q_ctx.cfg = {"settings": dict(ctx.cfg["settings"], tools=False, show_thinking=False)} gen = stream_anthropic if ctx.p["kind"] == "anthropic" else stream_openai res, usage = "", None begin_status("verify") sp = Spinner(status_label) sp.start() try: for kind, payload in gen(q_ctx, [{"role": "user", "content": prompt}]): if kind == "text": res += payload elif kind == "usage": usage = payload except Exception as e: sp.stop() if CANCEL.is_set(): return body println(" " + c("warn", "⚠ Самопроверка не удалась: %s — показываю ответ как есть" % clip(str(e), 100))) println() return body finally: sp.stop() if usage: for k in ("in", "out", "think"): self.tot[k] += usage.get(k) or 0 self.tot["cost"] += usage.get("cost") or 0 res = res.strip() if re.fullmatch(r"(OK|ОК|Ok)[.!]?", res): println(" " + c("ok", "✓ Самопроверка: факты подтверждены")) println() return body m = re.match(r"ИСПРАВЛЕНО:\s*(.*?)\n+(.+)\Z", res, re.S) if m and m.group(2).strip(): note, fixed = m.group(1).strip(), m.group(2).strip() println(" " + c("warn", "⚠ Самопроверка нашла неточности: ") + c("muted", clip(note or "см. ниже", term_w() - 40))) println() md = MD() md.feed(fixed + "\n") md.finish() println() if msgs and msgs[-1]["role"] == "assistant": msgs[-1]["content"], msgs[-1]["blocks"] = fixed, None return fixed println(" " + c("warn", "⚠ Самопроверка не дала однозначного ответа — показываю как есть")) println() return body def transcript_of(self, msgs): """Текстовая расшифровка кусков истории для сводки (общая для /compact и сжатия посреди задачи).""" lines = [] for m in msgs: if m.get("summary"): lines.append("[Прежняя сводка]\n" + str(m.get("content") or "")) elif m["role"] == "user": lines.append("Пользователь: " + str(m.get("content") or "")) elif m["role"] == "assistant": t = str(m.get("content") or "") calls = ", ".join("%s(%s)" % (tc["name"], clip(json.dumps(tc.get("args") or {}, ensure_ascii=False), 120)) for tc in m.get("tool_calls") or []) if t or calls: lines.append("Ассистент: " + t + ("\n[вызовы: %s]" % calls if calls else "")) elif m["role"] == "tool": lines.append("[результат %s]: %s" % (m.get("name") or "", str(m.get("content") or "")[:400])) transcript = "\n".join(lines) if len(transcript) > 60000: transcript = transcript[:15000] + "\n…(середина пропущена)…\n" + transcript[-45000:] return transcript def midrun_compact(self, ctx, msgs, anchor, run): """Сжимает ход работы посреди задачи, если контекст заполнен на MIDRUN_AT. Задача и последние шаги остаются дословно. Возвращает новый индекс задачи или None.""" lim_cfg = self.cfg["settings"].get("compact_at", 60000) or 0 if not lim_cfg or run.get("midrun_stop") or anchor >= len(msgs) or msgs[anchor]["role"] != "user": return None limit = next((m.get("ctx") for m in self.models if m.get("id") == ctx.model and m.get("ctx")), 0) or lim_cfg used = msgs_tokens(msgs) if used < limit * MIDRUN_AT: return None steps = [i for i in range(anchor + 1, len(msgs)) if msgs[i]["role"] == "assistant"] if len(steps) <= MIDRUN_KEEP_STEPS: return None tail = steps[-MIDRUN_KEEP_STEPS] # с этого шага хвост остаётся дословно old = [m for i, m in enumerate(msgs[:tail]) if i != anchor] if msgs_tokens(old) < 1500: return None println(" " + c("accent2", "◎ ") + c("warn", "Контекст заполнен на %d%% — сжимаю память посреди задачи…" % (used * 100 // limit))) try: summary = self.quiet_generate(MIDRUN_TEMPLATE % self.transcript_of(old), "compact") except Exception as e: if CANCEL.is_set(): raise KeyboardInterrupt("отменено пользователем") println(" " + c("warn", "⚠ Не удалось сжать память посреди задачи: %s — продолжаю как есть" % clip(str(e), 100))) println() run["midrun_stop"] = True return None if CANCEL.is_set(): raise KeyboardInterrupt("отменено пользователем") if not summary: println(" " + c("warn", "⚠ Сводка пустая — продолжаю как есть")) println() run["midrun_stop"] = True return None head = [ {"role": "user", "content": "[Сводка хода работы — память сжата автоматически посреди задачи]\n" + summary, "summary": True}, {"role": "assistant", "content": "Понял, продолжаю с учётом этой сводки."}, ] task = msgs[anchor] msgs[:tail] = head + [task] # на месте старых шагов — сводка, задача на своём месте after = msgs_tokens(msgs) println(" " + c("ok", "✓ Память сжата посреди задачи: ") + c("muted", "~%s → ~%s токенов · задача и последние шаги остались дословно" % (fmt_tok(used), fmt_tok(after)))) println() if after >= limit * MIDRUN_AT: run["midrun_stop"] = True # хвост и так занимает почти всё окно: больше не сжимаем в этом ходе return len(head) def autocheck(self, ctx, run, attempt, limit): """Автопроверка после правок: синтаксис, теги, скрипты и скриншоты HTML (если модель видит картинки). Возвращает сообщение для модели с просьбой исправить или None, если проверка прошла (или дальше не имеет смысла).""" files = sorted(run["changed"] or run["touched"]) run["changed"].clear() if not files: return None def rel(p): return os.path.relpath(p, self.cwd) println(" " + c("accent2", "◎ ") + c("muted", "автопроверка %d из %d: " % (attempt, limit) + ", ".join(rel(f) for f in files[:5]))) problems = [] for f in files: err = check_file(f) if os.path.isfile(f) else None if err: problems.append((rel(f), err)) shots, names = [], [] if vision_ok(self.cfg, ctx.model, self.models): pages = [f for f in files if f.lower().endswith((".html", ".htm")) and os.path.isfile(f)] for f in pages[:2]: try: shots.append(image_part(take_screenshot(pathlib.Path(f).as_uri()))) names.append(rel(f)) except Exception as e: println(" " + c("warn", "⚠ скриншот не снят: ") + c("muted", clip(str(e), term_w() - 30))) if not problems and not shots: println(" " + c("ok", "✓ ошибок не найдено")) println() run["failed"], run["sig"], run["same"] = False, None, 0 return None # одни и те же ошибки три проверки подряд: модель не справляется, не крутим дальше sig = tuple(problems) run["same"] = run["same"] + 1 if sig and sig == run["sig"] else 0 run["sig"] = sig if run["same"] >= 2: for f, err in problems: println(" " + c("err", "✗ ") + c("text", f) + c("muted", " — " + clip(err, term_w() - 30))) println(" " + c("warn", "⚠ одна и та же ошибка три проверки подряд: автоисправление остановлено")) println() run["failed"] = True # ошибка осталась: в конце хода покажем предупреждение return None for f, err in problems: println(" " + c("err", "✗ ") + c("text", f) + c("muted", " — " + clip(err, term_w() - 30))) if shots: println(" " + c("accent2", "▣ скриншотов для проверки: %d" % len(shots))) println() run["failed"] = True lines = ["[Автопроверка %d из %d]" % (attempt, limit)] if attempt > 1: lines.append("Прошлые замечания ещё не исправлены или появились новые.") lines += ["- %s: %s" % p for p in problems] or ["- синтаксис и теги в порядке"] if shots: lines.append("Во вложении скриншот интерфейса: %s." % ", ".join(names)) lines.append("Посмотри на него и найди, что выглядит криво: обрезанный или налезающий текст, " "сломанная сетка и выравнивание, пустые или слишком плотные места, слабый контраст, " "лишние элементы.") lines.append("Исправляй файлами, пока всё не будет в порядке. Когда проблем не осталось, ответь одним словом: OK.") msg = {"role": "user", "content": "\n".join(lines)} if shots: msg["images"] = shots return msg def run_guarded(self, ctx, msgs, title, rollback_to=None, notify=True): """Ход агента: ошибки, Esc, статус, самопроверка и звук. Возвращает текст или None.""" text = None try: text = self.agent_loop(ctx, msgs, title) except KeyboardInterrupt: CANCEL.set() except Exception as e: if not CANCEL.is_set(): println() what, how = friendly_error(e) println(" " + c("err", "✗ " + (what or "Ошибка"))) if how: println(" " + c("warn", "→ " + how)) println(" " + c("dim", clip(str(e), max(40, term_w() - 6)))) println() set_result("error", "ошибка") play_sound("error") if rollback_to is not None: if len(msgs) == rollback_to + 1: msgs.pop() else: msgs.append({"role": "assistant", "content": "(ошибка API: %s)" % e}) return None finally: stop_all_spinners() if not CANCEL.is_set() and notify and text and text.strip() and self.cfg["settings"].get("verify", True) \ and self.mode != "chat": text = self.verify_answer(ctx, msgs, text) if CANCEL.is_set(): println() println(" " + c("warn", "■ остановлено (Esc)")) self.close_pending(msgs) println() set_result("stopped", "остановлено") return None set_result("done", "готово") if notify: play_sound("done") return text def run_task(self, fn): """Верхний уровень: сбрасывает отмену, включает слушатель Esc на время задачи. Вложенные вызовы (например, скилл внутри @-строки) слушатель не трогают.""" outer = not self._in_task if outer: CANCEL.clear() ACTIVE_ALL.clear() self._in_task = True self.watch.start() try: fn() finally: self.persist_chat() if outer: self._in_task = False self.watch.stop() stop_all_spinners() def run_call(self, call, msgs, run=None): name, args = call["name"], call["args"] or {} res = "(не выполнено)" println(" " + c("accent", "⚙ ") + c("text", describe(name, args))) try: if self.mode == "chat": res = "Режим Chat: инструменты не используются. Ответь текстом." println(" " + c("err", "✗ режим Chat")) elif self.mode == "plan" and name in PLAN_BLOCKED: res = ("Режим Plan: это действие меняет файлы или запускает команды, поэтому запрещено. " "Опиши шаги словами; изменения сделаем в режиме Build.") println(" " + c("err", "✗ режим Plan")) elif not self.cfg["settings"]["tools"]: res = "Инструменты выключены в настройках FlynCode. Ответь без них." println(" " + c("err", "✗ инструменты выключены")) elif not self.allowed_now(name, args): res = ("Действие не разрешено: отказ пользователя или запрет в настройках. " "Не повторяй его без нового запроса; предложи альтернативу.") println(" " + c("err", "✗ отказано")) else: set_status("tool", name) t0 = time.time() sp = Spinner(lambda: ("Выполняю %s · %s · Esc — отмена" % (name, fmt_dur(time.time() - t0)), "muted")) if name not in ("ask_agent", "ask_user"): sp.start() try: SHOT_TL.path = None ok, res, summ = run_tool(self, name, args) if ok and run is not None: track_run(run, name, args, self.cwd) finally: sp.stop() println(" " + (c("ok", "✓ ") if ok else c("err", "✗ ")) + c("muted", clip(summ, term_w() - 10))) if name in ("bash", "grep", "list_files") and ok: for ln in res.splitlines()[:4]: println(" " + c("dim", clip(ln, term_w() - 10))) finally: msgs.append({"role": "tool", "tool_call_id": call["id"], "name": name, "content": res}) self.persist_chat() # ---------- память: сжатие длинного чата def compact_cut(self, keep_turns): """Индекс, с которого история остаётся дословной: начало одной из последних keep_turns реплик.""" idx = [i for i, m in enumerate(self.messages) if m["role"] == "user" and not m.get("summary")] if len(idx) <= keep_turns: return 0 return idx[-keep_turns] def quiet_generate(self, prompt, state): """Запрос без инструментов и без печати: возвращает ответ целиком (нужен для сжатия памяти).""" q = copy.copy(self) q.cfg = {"settings": dict(self.cfg["settings"], tools=False, show_thinking=False)} gen = stream_anthropic if self.p["kind"] == "anthropic" else stream_openai res, usage = "", None begin_status(state) sp = Spinner(status_label) sp.start() try: for kind, payload in gen(q, [{"role": "user", "content": prompt}]): if kind == "text": res += payload elif kind == "usage": usage = payload finally: sp.stop() if usage: for k in ("in", "out", "think"): self.tot[k] += usage.get(k) or 0 self.tot["cost"] += usage.get("cost") or 0 return res.strip() def compact(self, auto=False): """Старая часть истории → сводка; последние реплики остаются дословно. True — если сжали.""" cut = self.compact_cut(COMPACT_KEEP_TURNS) if cut <= 0: if not auto: println(" " + c("muted", "Пока нечего сжимать: реплик мало.")) println() return False before = msgs_tokens(self.messages) transcript = self.transcript_of(self.messages[:cut]) try: summary = self.quiet_generate(COMPACT_TEMPLATE % transcript, "compact") except Exception as e: if CANCEL.is_set(): return False println(" " + c("warn", "⚠ Не удалось сжать память: %s — история осталась полной" % clip(str(e), 100))) println() return False if CANCEL.is_set() or not summary: if not CANCEL.is_set(): println(" " + c("warn", "⚠ Сводка пустая — история осталась полной")) println() return False self.messages[:cut] = [ {"role": "user", "content": "[Сводка предыдущего разговора — память сжата автоматически]\n" + summary, "summary": True}, {"role": "assistant", "content": "Понял, продолжаю с учётом этой сводки."}, ] after = msgs_tokens(self.messages) println(" " + c("ok", "✓ Память сжата: ") + c("muted", "~%s → ~%s токенов · последние реплики остались дословно" % (fmt_tok(before), fmt_tok(after)))) println() return True def maybe_compact(self): lim = self.cfg["settings"].get("compact_at", 60000) or 0 if not lim or msgs_tokens(self.messages) <= lim: return cut = self.compact_cut(COMPACT_KEEP_TURNS) # если сжимать почти нечего (вся тяжесть в последних репликах) — не дёргаем модель на каждом ходе if cut <= 0 or msgs_tokens(self.messages[:cut]) < 1500: return self.compact(auto=True) def cmd_runtime(self): R = detect_runtimes() println() println(" " + c("accent", "Среды запуска", bold=True)) labels = [("bun", "Bun (JS и TypeScript)"), ("node", "Node.js"), ("python", "Python"), ("git", "Git"), ("bash", "bash")] for k, lab in labels: v = R.get(k) println(" " + c("muted", padv(lab, 24)) + (c("ok", "✓ " + v) if v else c("warn", "не найден"))) println() println(" " + c("muted", "run_script сам выбирает среду: .js → Bun или Node, .ts → Bun, .py → Python.")) bs = BUN_STATE.get("state") if bs == "installing": println(" " + c("warn", "Bun ставится в фоне… подожди немного")) elif bs == "error": println(" " + c("err", "Bun не встал: " + BUN_STATE.get("msg", ""))) println(" " + c("muted", "Поставь вручную с bun.sh или проверь интернет — FlynCode попробует снова при запуске.")) elif not R.get("bun"): println(" " + c("muted", "Bun поставится сам при следующем запуске (настройка «Ставить Bun сам»).")) println() def cmd_verify(self, arg=""): """/verify — переключить самопроверку; /verify on|off — явно.""" S = self.cfg["settings"] a = (arg or "").strip().lower() on_words = ("on", "вкл", "да", "1", "yes") off_words = ("off", "выкл", "нет", "0", "no") if a in on_words: S["verify"] = True elif a in off_words: S["verify"] = False elif a: println(" " + c("err", "✗ Используй: /verify on или /verify off (без аргумента — переключить)")) println() return else: S["verify"] = not S.get("verify", True) save_cfg(self.cfg) if S["verify"]: println(" " + c("ok", "✓ Самопроверка включена") + c("muted", " (ответ сверяется с фактами, расход токенов x2)")) else: println(" " + c("warn", "■ Самопроверка выключена") + c("muted", " (ответы показываются сразу)")) println() def cmd_compact(self): self.run_task(lambda: self.compact(auto=False)) # ---------- подсказки: агенты и скиллы def cmd_tips(self): A = [a for a in self.cfg.get("agents", []) if a.get("on", True)] sk = list_skills(self.cwd) names_a = ", ".join("@" + a["name"] for a in A) or "пока нет — создай в /agents" names_s = ", ".join("@" + n for n in sk) or "пока нет — /skills покажет, куда положить папку" def H(t): println(" " + c("accent", t, bold=True)) def T(t): println(" " + c("text", t)) def M(t): println(" " + c("muted", t)) println() H("АГЕНТ — отдельная модель со своей ролью, провайдером и моделью") T(" @Код напиши тесты вызвать агента; его ответ увидит и основная модель") T(" /debate тема несколько агентов спорят, судья пишет итог") T(" /agents создать и настроить агентов") M(" Сейчас: " + names_a) println() H("СКИЛЛ — готовая инструкция (папка со SKILL.md), по которой работает основная модель") T(" @имя задача инструкция целиком попадёт в запрос, модель обязана ей следовать") T(" просто опиши задачу если подходящий скилл есть, модель подключит его сама") T(" /skills какие скиллы найдены и где лежат") M(" Сейчас: " + names_s) println() H("КАК ВЫБРАТЬ") T(" нужна другая модель, провайдер или роль → агент") T(" нужны правила, шаблон, формат, чек-лист → скилл") T(" нужно просто ответить → обычный вопрос, ничего вызывать не надо") println() H("НЕСКОЛЬКО АГЕНТОВ СРАЗУ") T(" @Код > @Ревью > @План задача по очереди: каждый видит ответ предыдущего") T(" @Ревью + @План задача параллельно: оба получают одну задачу") T(" @Код + @Ревью > @Итог задача сначала параллельно, потом следующий этап") M(" В параллельном режиме инструменты с вопросом не запускаются: разреши их в /perm") println() H("ЕСЛИ ЗАПУТАЛСЯ") T(" «Нет агента или скилла» → /skills покажет точные имена") T(" Имя из SKILL.md важнее имени папки. Если имена совпадают, сначала ищется агент.") T(" Долго нет ответа → строка статуса покажет этап и время; Esc отменит") T(" Чат стал длинным → /compact сожмёт старую часть в сводку (это делается и само)") println() def turn(self, text, shown=None, title=None): self.maybe_compact() n0 = len(self.messages) auto = skills_mentioned(self.cwd, text) content = ("\n\n".join(auto) + "\n\n" + text) if auto else text imgs, self.pending_images = self.pending_images, [] msg = {"role": "user", "content": content} if imgs: names = ", ".join(i["name"] for i in imgs) msg["content"] = "[Фото: %s]\n%s" % (names, content or "Что на фото?") msg["images"] = imgs println(" " + c("accent2", "📎 ") + c("muted", names)) self.messages.append(msg) self.print_user(text if shown is None else shown) self.run_guarded(self, self.messages, title or ("FlynCode · " + self.model), rollback_to=n0) def run_skill(self, sk, task): """@скилл задача: инструкции скилла идут в запрос, модель обязана им следовать.""" ask = task or "примени скилл к текущему контексту чата." text = ("[Скилл «%s». Строго следуй этим инструкциям:]\n%s\n[конец скилла]\n\nЗадача: %s" % (sk["name"], sk["body"], ask)) shown = "@%s%s" % (sk["name"], (" " + task) if task else "") title = "скилл %s · %s" % (sk["name"], self.model) self.run_task(lambda: self.turn(text, shown=shown, title=title)) def skills_block(self): """Список скиллов для системного промпта: модель сама решает, когда загрузить инструкции.""" if not self.cfg["settings"]["tools"]: return "" sk = list_skills(self.cwd) if not sk: return "" lines = ["Доступные скиллы — готовые инструкции для типовых задач. Если задача подходит под описание " "скилла, вызови load_skill(name), прочитай инструкции и строго следуй им:"] lines += ["- %s: %s" % (s["name"], s["description"][:300]) for s in sk.values()] return "\n".join(lines) def cmd_skills(self): sk = list_skills(self.cwd) println() if not sk: println(" " + c("muted", "Скиллов пока нет. Создай папку со SKILL.md в одном из мест:")) println(" " + c("text", os.path.join(CFG_DIR, "skills", "<имя>")) + c("muted", " — для всех проектов")) println(" " + c("text", os.path.join(self.cwd, ".flyncode", "skills", "<имя>")) + c("muted", " — только для этой папки")) else: for s in sk.values(): println(" " + c("accent", padv("@" + s["name"], 22), bold=True) + c("muted", clip(s["description"], term_w() - 26))) println() println(" " + c("muted", "Вызов: ") + c("text", "@имя твоя задача") + c("muted", " или просто опиши задачу — модель сама подключит подходящий скилл.")) println() println(" " + c("muted", "Агенты или скиллы, что выбрать — ") + c("text", "/tips")) println(" " + c("muted", "Где ищу скиллы:")) for ln in skill_report(self.cwd): println(" " + c("dim", clip(ln, term_w() - 4))) println() # ---------- команды def cmd_plan(self, arg=""): if (arg or "").strip().lower() in ("clear", "очистить", "сброс"): self.plan = [] println(" " + c("ok", "✓ План очищен")) println() return if not self.plan: println(" " + c("muted", "Плана пока нет. Переключись в Plan (tab) и опиши задачу — план сохранится здесь.")) println() return done = sum(1 for t in self.plan if t["done"]) println(" " + c("text", "План: %d из %d выполнено" % (done, len(self.plan)), bold=True)) for t in self.plan: mark = c("ok", "[x]") if t["done"] else c("muted", "[ ]") println(" " + mark + " " + c("muted" if t["done"] else "text", clip(t["text"], term_w() - 8))) println() def attach_paths(self, paths): """Прикрепляет картинки к следующему сообщению. Проверяет формат, размер и то, умеет ли модель смотреть картинки.""" if not paths: return if not vision_ok(self.cfg, self.model, self.models): println(" " + c("err", "✗ Модель «%s» не понимает картинки" % (self.model or "—")) + c("muted", " выбери другую (/model) или в настройках «Скриншоты для зрения» → всегда")) println() return added = [] for p in paths: try: img = load_image_file(p) except (ValueError, OSError) as e: println(" " + c("err", "✗ " + str(e))) continue self.pending_images.append(img) added.append(img["name"]) if added: println(" " + c("ok", "✓ Прикреплено: ") + c("text", ", ".join(added)) + c("muted", " — уйдут с следующим сообщением")) println() def poll_one(self, question, options, custom): """Один вопрос пользователю: меню с вариантами и, если нужно, свой ответ. None — вопрос пропущен (Esc).""" items = [{"label": o, "desc": "", "value": ("opt", i)} for i, o in enumerate(options)] if custom and options: items.append({"label": "✎ Свой вариант…", "desc": "написать ответ", "value": ("own", None)}) println() println(" " + c("accent2", "❓ ") + c("text", question, bold=True)) self.watch.stop() # на время опроса Esc означает «пропустить» try: v = select("Выбери вариант", items, searchable=False, hint="↑↓ выбор · Enter ок · Esc пропустить") if options else ("own", None) if v is None: println(" " + c("muted", "пропущено")) println() return None if v[0] == "opt": ans = options[v[1]] else: typed = self.ask(c("text", "Твой вариант", bold=True)) if not typed or not typed.strip(): println(" " + c("muted", "пропущено")) println() return None ans = typed.strip() finally: self.watch.start() println(" " + c("ok", "✓ ") + c("text", ans)) println() return ans def on_paste(self): """Ctrl+V / Ctrl+G в строке ввода: картинка из буфера прикрепляется, текст вставляется в строку.""" path, _ = clipboard_image() if path: self.attach_paths([path]) return "" return clipboard_text() def cmd_attach(self, arg=""): a = (arg or "").strip() if not a: if self.pending_images: println(" " + c("text", "Прикреплены: ") + c("muted", ", ".join(i["name"] for i in self.pending_images))) println(" " + c("muted", "/attach clear — убрать · /paste — картинка из буфера · или перетащи файл в окно")) else: println(" " + c("muted", "Фото не прикреплено. Перетащи файл в окно, /attach <путь> или /paste (картинка из буфера).")) println() return if a.lower() in ("clear", "очистить", "сброс"): self.pending_images = [] println(" " + c("ok", "✓ Прикреплённые фото убраны")) println() return if a.lower() == "paste": return self.cmd_paste() p = a.strip("\"'").replace("\\ ", " ") full = os.path.normpath(os.path.join(self.cwd, os.path.expanduser(p))) if not os.path.isfile(full): println(" " + c("err", "✗ Нет такого файла: ") + c("muted", p)) println() return self.attach_paths([full]) def cmd_paste(self): path, why = clipboard_image() if not path: println(" " + c("err", "✗ " + why)) println() return self.attach_paths([path]) def cmd_sidebar(self): SIDEBAR["on"] = not SIDEBAR["on"] self.cfg["settings"]["sidebar"] = SIDEBAR["on"] save_cfg(self.cfg) println(" " + c("ok", "✓ Правая панель: ") + c("text", "вкл" if SIDEBAR["on"] else "выкл")) println() def cmd_help(self): println() for nm, ds in COMMANDS: println(" " + c("accent", padv(nm, 12), bold=True) + c("muted", ds)) println() println(" " + c("muted", "Примеры: ") + c("text", "«найди баг в app.py и исправь»") + c("muted", " — агент сам прочитает файл и попросит разрешение на правку.")) println(" " + c("text", "@Код напиши тесты") + c("muted", " — вызвать агента. ") + c("text", "/debate кэшировать ли ответы") + c("muted", " — дискуссия агентов с итогом.")) println(" " + c("text", "@Код > @Ревью задача") + c("muted", " — несколько агентов по очереди, «+» — параллельно.")) println(" " + c("muted", "Как не запутаться с агентами и скиллами — ") + c("text", "/tips")) println() def cmd_model(self, arg): if not self.models or not any(m["id"] == self.model for m in self.models): try: with_spinner("Обновляю список моделей…", self.refresh_models) except ApiError as e: println(" " + c("err", "✗ Не удалось обновить: " + str(e))) println() return println() println(" " + c("accent", "Доступные модели · %s" % self.p["name"], bold=True)) println(" " + c("dim", "──────────────────────────────────────────────────────────────────")) W = term_w() id_w = min(40, max(28, W // 3)) info_w = W - id_w - 8 header = "%s %s" % (padv(c("text", "Модель", bold=True), id_w), c("muted", "контекст ◆размышления vision reasoning цена")) println(" " + header) println(" " + c("dim", "──────────────────────────────────────────────────────────────────")) for m in self.models: cur = "●" if m["id"] == self.model else " " id_cell = "%s %s" % (cur, clip(m["id"], id_w - 2)) parts = [] if m.get("ctx"): parts.append(fmt_tok(m["ctx"]) + " ctx") lv = m.get("levels") or ["auto"] if len(lv) > 1: parts.append("◆" + "/".join(x for x in lv if x != "auto")) if m.get("vision"): parts.append("👁vision") if m.get("think"): parts.append("💭reasoning") if m.get("price") and any(m["price"]): parts.append("$%.2f/$%.2f" % tuple(m["price"])) info = clip(" ".join(parts), info_w) is_current = m["id"] == self.model println(" " + (c("ok", id_cell) if is_current else c("text", id_cell)) + " " + (c("ok", info) if is_current else c("muted", info))) println(" " + c("dim", "──────────────────────────────────────────────────────────────────")) println(" " + c("muted", "Выбрать: /model <поиск> · Текущая: ") + c("text", self.model)) println() ch = select("Сменить модель?", [ {"label": "Да, выбрать другую", "desc": "", "value": "yes"}, {"label": "Оставить текущую", "desc": "", "value": "no"}, ], searchable=False) if ch == "yes": if self.pick_model(query=arg): self.save_last() println(" " + c("ok", "✓ Модель: ") + c("text", self.model)) else: println(" " + c("muted", "Модель не изменена")) println() def cmd_think(self, arg): lv = self.levels() if arg: a = arg.strip().lower() if a in lv: self.effort = a self.save_last() println(" " + c("ok", "✓ Размышления: ") + c("text", a)) else: println(" " + c("err", "✗ Нет такого уровня. Доступно: ") + ", ".join(lv)) elif len(lv) > 1: self.pick_effort() self.save_last() println(" " + c("ok", "✓ Размышления: ") + c("text", self.effort)) else: println(" " + c("muted", "У этой модели нет настройки размышлений.")) println() def cmd_provider(self): prev = (self.p, self.key, self.base, self.models, self.model, self.effort) p = self.pick_provider() if p is None: return if self.connect(p) and self.pick_after_connect(): println(" " + c("ok", "✓ Теперь: ") + c("text", "%s · %s" % (p["name"], self.model)) + c("muted", " (контекст чата сохранён)")) else: self.p, self.key, self.base, self.models, self.model, self.effort = prev println() def cmd_perm(self): self.perm_menu() println() def cmd_cd(self, arg): if not arg: println(" " + c("muted", "Использование: /cd <папка> · сейчас: " + self.cwd)) else: path = os.path.abspath(os.path.expanduser(arg)) if os.path.isdir(path): self.cwd = path println(" " + c("ok", "✓ Рабочая папка: ") + c("text", path)) else: println(" " + c("err", "✗ Нет такой папки: ") + path) println() def cmd_settings(self): S = self.cfg["settings"] yn = lambda b: "Вкл" if b else "Выкл" while True: n_ask = sum(1 for k in ASK if S["perms"].get(k) == "ask") n_ag = len(self.cfg.get("agents", [])) n_sk = len(list_skills(self.cwd)) items = [ {"label": "Агенты", "desc": "%d шт. · @имя в чате, /debate тема" % n_ag, "value": "agents"}, {"label": "Раундов в дебатах", "desc": str(S.get("rounds", 2)), "value": "rounds"}, {"label": "Показывать размышления", "desc": yn(S["show_thinking"]), "value": "think"}, {"label": "Инструменты (файлы, bash, интернет)", "desc": yn(S["tools"]), "value": "tools"}, {"label": "Разрешения инструментов", "desc": "спрашивать: %d из %d" % (n_ask, len(ASK)), "value": "perms"}, {"label": "Скиллы", "desc": "%d шт. · нажми, чтобы открыть папку" % n_sk, "value": "skills"}, {"label": "Самопроверка ответа", "desc": yn(S.get("verify", True)) + " · сверяет факты перед показом", "value": "verify"}, {"label": "Автопроверка правок", "desc": yn(S.get("autocheck", True)) + " · синтаксис и вёрстка после изменений", "value": "autocheck"}, {"label": "Исправлять до успеха", "desc": "до %d попыток · потом остановлюсь" % S.get("autocheck_rounds", 10), "value": "autocheck_rounds"}, {"label": "Скриншоты для зрения", "desc": VISION_LABEL[S.get("vision", "auto")], "value": "vision"}, {"label": "Звуки: конец и ошибка", "desc": yn(S.get("sound", True)), "value": "sound"}, {"label": "Уведомления на экран", "desc": yn(S.get("notify", True)) + " · «ответ готов» и «ошибка» поверх окон", "value": "notify"}, {"label": "Сжатие памяти", "desc": ("при ~%s токенов" % fmt_tok(S.get("compact_at", 60000))) if S.get("compact_at", 60000) else "выкл", "value": "compact"}, {"label": "Экономия токенов", "desc": S.get("token_saving", "auto") + " · короткие правила, обрезка длинных результатов", "value": "token_saving"}, {"label": "Ставить Bun сам", "desc": yn(S.get("bun_auto", True)) + " · на C:\\FlynCode\\bun", "value": "bun_auto"}, {"label": "Тема", "desc": THEMES[S["theme"]][0], "value": "theme"}, {"label": "Команда", "desc": "отдел разработчиков FlynCode", "value": "team"}, {"label": "Забыть API-ключ этого провайдера", "desc": self.p["name"], "value": "forget"}, ] v = select("Настройки FlynCode", items, searchable=False, hint="Enter — изменить, Esc — закрыть") if v is None: save_cfg(self.cfg) return if v == "agents": self.agents_menu() elif v == "rounds": S["rounds"] = S.get("rounds", 2) % 4 + 1 elif v == "think": S["show_thinking"] = not S["show_thinking"] elif v == "tools": S["tools"] = not S["tools"] elif v == "perms": self.perm_menu() elif v == "skills": path = os.path.join(CFG_DIR, "skills") if open_folder(path): println(" " + c("ok", "✓ Открыл папку скиллов: ") + c("text", path)) println(" " + c("muted", "Положи туда папку со SKILL.md, например: ") + c("text", "design-pro/SKILL.md")) else: println(" " + c("warn", "⚠ Не удалось открыть папку. Она здесь: ") + c("text", path)) elif v == "bun_auto": S["bun_auto"] = not S.get("bun_auto", True) if S["bun_auto"]: start_bun_autoinstall(True) elif v == "compact": lv = [30000, 60000, 120000, 0] cur = S.get("compact_at", 60000) S["compact_at"] = lv[(lv.index(cur) + 1) % len(lv)] if cur in lv else 60000 elif v == "token_saving": lv = ["off", "auto", "always"] cur = S.get("token_saving", "auto") S["token_saving"] = lv[(lv.index(cur) + 1) % len(lv)] if cur in lv else "auto" elif v == "verify": S["verify"] = not S.get("verify", True) elif v == "autocheck": S["autocheck"] = not S.get("autocheck", True) elif v == "autocheck_rounds": cur = S.get("autocheck_rounds", 10) S["autocheck_rounds"] = AUTOCHECK_LEVELS[(AUTOCHECK_LEVELS.index(cur) + 1) % len(AUTOCHECK_LEVELS)] \ if cur in AUTOCHECK_LEVELS else 10 elif v == "vision": S["vision"] = {"auto": "on", "on": "off", "off": "auto"}.get(S.get("vision", "auto"), "auto") elif v == "sound": S["sound"] = not S.get("sound", True) SOUND["on"] = S["sound"] if SOUND["on"]: play_sound("done") elif v == "notify": S["notify"] = not S.get("notify", True) NOTIFY["on"] = S["notify"] if NOTIFY["on"]: desktop_notify("done") elif v == "theme": t = select("Тема", [{"label": val[0], "desc": "", "value": k} for k, val in THEMES.items()], current=S["theme"], searchable=False) if t: S["theme"] = t set_theme(t) elif v == "team": self.show_team() elif v == "forget": self.cfg["keys"].pop(self.p["id"], None) self.key = "" println(" " + c("ok", "✓ Ключ удалён из конфига")) save_cfg(self.cfg) def show_team(self): """Экран «Команда»: отдел разработчиков FlynCode.""" println() println(" " + c("accent", "Отдел «Разработчики»", bold=True)) println() println(" " + c("text", "Главные разработчики", bold=True)) for n in ("Buldak_Skillofich", "vakuru"): println(" " + c("accent2", "● ") + c("text", n)) println() println(" " + c("text", "Помощники", bold=True)) println(" " + c("accent2", "● ") + c("text", "Ryzik")) println() println(" " + c("text", "Монтажер", bold=True)) println(" " + c("accent2", "● ") + c("text", "Waykaste")) println() select("Команда FlynCode", [{"label": "Назад", "desc": "", "value": "back"}], searchable=False, hint="Enter или Esc — назад") println() def cmd_usage(self): t = self.tot println() rows = [("Сообщений", str(t["turns"])), ("Токенов на входе", fmt_tok(t["in"])), ("Токенов на выходе", fmt_tok(t["out"])), ("из них размышления", fmt_tok(t["think"])), ("Время генерации", fmt_dur(t["time"]))] if t["cost"]: rows.append(("Стоимость", "$%.4f" % t["cost"])) for k, v in rows: println(" " + c("muted", padv(k, 22)) + c("text", v, bold=True)) println() def cmd_models(self): old = {m["id"] for m in self.models} try: with_spinner("Обновляю список моделей…", self.refresh_models) new = [m["id"] for m in self.models if m["id"] not in old] tail = c("accent", " · новых: " + ", ".join(new[:6])) if new else c("muted", " · новых нет") println(" " + c("ok", "✓ Моделей: %d" % len(self.models)) + tail) except ApiError as e: println(" " + c("err", "✗ " + str(e))) println() def cmd_key(self): if self.p.get("nokey"): println(" " + c("muted", "Этому провайдеру ключ не нужен.")) return if self.p.get("url"): println(" " + c("muted", "Получить ключ: ") + link(self.p["url"])) v = self.ask(c("text", "Новый API-ключ для %s" % self.p["name"], bold=True), mask=True) if v and v.strip(): self.key = v.strip() self.cfg["keys"][self.p["id"]] = self.key save_cfg(self.cfg) self.cmd_models() def cmd_clear(self): self.new_chat() self.banner() println(" " + c("ok", "✓ Новый чат") + c("muted", " (рабочая папка и разрешения сохранены)")) println() def show_history(self, cid, msgs): """Открытый чат: экран очищается, видно его последние реплики и можно продолжить с места остановки.""" clear_screen() SCREEN.center = False println(" " + c("accent", "История", bold=True) + c("muted", " %s · %d реплик · пиши, чтобы продолжить" % (cid, len(msgs)))) println() shown = [m for m in msgs if m.get("role") in ("user", "assistant") and not m.get("tool_calls") and not m.get("summary") and isinstance(m.get("content"), str) and m["content"].strip()] for m in shown[-30:]: text = m["content"].strip() if m["role"] == "user": self.print_user(text if len(text) <= 600 else text[:600] + "…") else: for ln in text[:2000].split("\n"): println(" " + c("text", ln)) println() def cmd_chats(self, arg=""): chats = list_chats() if not chats: println(" " + c("muted", "Чатов пока нет — начни писать, они сохранятся автоматически")) println() return items = [] for ch in chats: when = datetime.fromtimestamp(ch["ts"]).strftime("%d.%m %H:%M") if ch["ts"] else "" items.append({"label": "%s %s" % (when, ch["first"] or "(пусто)"), "desc": "%s · %d реплик · %s" % (ch["id"], ch["n"], ch["cwd"] or "—"), "value": ch["id"]}) items.append({"label": "⬢ Удалить чат…", "desc": "выбрать для удаления", "value": "__del"}) while True: v = select("История чатов (Enter открыть · Esc закрыть)", [{"label": "+ Новый чат", "desc": "начать пустой чат", "value": "__new"}] + items, current=None, searchable=True) if v is None: return if v == "__new": self.cmd_clear() return if v == "__del": items2 = [{"label": "%s %s" % (datetime.fromtimestamp(ch["ts"]).strftime("%d.%m %H:%M"), ch["first"][:50] or "—"), "desc": ch["id"], "value": ch["id"]} for ch in chats] d = select("Какой чат удалить?", items2, searchable=True) if d: try: os.remove(_hist_path(d)) println(" " + c("ok", "✓ Удалён чат %s" % d)) except Exception as e: println(" " + c("err", "✗ %s" % e)) if d == self.chat_id: self.new_chat(keep_old=False) chats = list_chats() else: println() return if not chats: println(" " + c("muted", "Больше чатов нет")) println() return items = [{"label": "%s %s" % (datetime.fromtimestamp(ch["ts"]).strftime("%d.%m %H:%M"), ch["first"] or "(пусто)"), "desc": "%s · %d реплик · %s" % (ch["id"], ch["n"], ch["cwd"] or "—"), "value": ch["id"]} for ch in chats] items.append({"label": "⬢ Удалить чат…", "desc": "выбрать для удаления", "value": "__del"}) continue msgs = load_chat(v) if not msgs: println(" " + c("err", "✗ Не удалось загрузить чат")) println() return self.persist_chat() self.messages = msgs self.plan = load_plan(v) self.chat_id = v self.cfg["chats_current"] = v save_cfg(self.cfg) self.show_history(v, msgs) return def cmd_export(self, arg=""): if not self.messages: println(" " + c("muted", "Нечего экспортировать — чат пустой")) println() return name = arg.strip() if arg else "chat_%s.md" % self.chat_id path = os.path.join(os.getcwd(), name if name.endswith(".md") else name + ".md") lines = ["# FlynCode — %s" % self.chat_id, "", "_%s_" % datetime.now().strftime("%Y-%m-%d %H:%M"), "Папка: %s" % self.cwd, ""] for m in self.messages: role = m.get("role", "?") content = m.get("content", "") if not content: continue who = {"user": "## Пользователь", "assistant": "## FlynCode", "tool": "## Инструмент %s" % (m.get("name") or "")}.get(role, "## %s" % role) lines.append(who) lines.append(str(content)) lines.append("") try: with open(path, "w", encoding="utf-8") as f: f.write("\n".join(lines)) println(" " + c("ok", "✓ Экспортировано: ") + c("text", path)) except Exception as e: println(" " + c("err", "✗ %s" % e)) println() def cmd_suggest_skill(self, arg=None): # Авто-скилл: модель предлагает завести скилл для текущей задачи. if not arg: arg = "Повторяющаяся задача пользователя (см. историю)" prompt = ("Проанализируй задачу пользователя и предложи, стоит ли завести FlynCode-скилл для неё.\n\n" "Задача: %s\n\n" "Если да — верни ровно JSON: {\"name\":\"kebab-name\",\"description\":\"короткое описание\",\"body\":\"полный текст SKILL.md (инструкция, 300-1500 знаков)\"}\n" "Если нет — верни: {\"none\": true}" % arg) text = self.quiet_generate(prompt, "wait") if not text: println(" " + c("muted", "Модель не предложила скилл")) println() return try: j = json.loads(text[text.find("{"):text.rfind("}") + 1]) except Exception: println(" " + c("err", "✗ Модель вернула не JSON, скилл не создан")) println() return if j.get("none"): println(" " + c("muted", "Авто-скилл не нужен для этой задачи")) println() return name = (j.get("name") or "auto-skill").lower().replace(" ", "-") desc = j.get("description") or "" body = j.get("body") or "" name = "".join(ch if ch.isalnum() or ch in "-_" else "-" for ch in name)[:60].strip("-") or "auto-skill" base = os.path.join(CFG_DIR, "skills") folder = os.path.join(base, name) sk_path = os.path.join(folder, "SKILL.md") try: os.makedirs(folder, exist_ok=True) with open(sk_path, "w", encoding="utf-8") as f: f.write("---\nname: %s\ndescription: %s\n---\n\n%s\n" % (name, desc, body)) println(" " + c("ok", "✓ Создан авто-скилл @%s" % name) + c("muted", " %s" % sk_path)) println(" " + c("muted", "Использование: @%s задача" % name)) except Exception as e: println(" " + c("err", "✗ %s" % e)) println() def handle(self, line): parts = line.split(None, 1) cmd = parts[0].lower() arg = parts[1].strip() if len(parts) > 1 else "" if cmd in ("/exit", "/quit", "/q"): return False table = { "/help": self.cmd_help, "/model": lambda: self.cmd_model(arg), "/think": lambda: self.cmd_think(arg), "/effort": lambda: self.cmd_think(arg), "/provider": self.cmd_provider, "/perm": self.cmd_perm, "/cd": lambda: self.cmd_cd(arg), "/settings": self.cmd_settings, "/agents": self.agents_menu, "/debate": lambda: self.run_task(lambda: self.cmd_debate(arg)), "/usage": self.cmd_usage, "/models": self.cmd_models, "/skills": self.cmd_skills, "/key": self.cmd_key, "/clear": self.cmd_clear, "/new": self.cmd_clear, "/tips": self.cmd_tips, "/compact": self.cmd_compact, "/verify": lambda: self.cmd_verify(arg), "/runtime": self.cmd_runtime, "/chats": self.cmd_chats, "/chat": self.cmd_chats, "/history": self.cmd_chats, "/export": lambda: self.cmd_export(arg), "/skill": self.cmd_suggest_skill, "/mode": lambda: self.cmd_mode(arg), "/plan": lambda: self.cmd_plan(arg), "/sidebar": self.cmd_sidebar, "/attach": lambda: self.cmd_attach(arg), "/img": lambda: self.cmd_attach(arg), "/paste": self.cmd_paste, "/tray": cmd_tray, } fn = table.get(cmd) if fn: fn() else: println(" " + c("err", "Неизвестная команда: " + cmd) + c("muted", " (/help — список)")) println() return True def status_and_prompt(self): self.status_line() return read_line(self.mode_bar, history=self.history, commands=True, placeholder="Спроси что угодно или введи /команду (@скилл)…", footer=self.composer_footer, on_tab=self.cycle_mode, on_paste=self.on_paste) def run(self): self.onboarding() self.banner() for note in self.notes: println(" " + c("warn", "⚠ " + note)) if self.notes: println() start_bun_autoinstall(self.cfg["settings"].get("bun_auto", True)) self.start_autosave() if self.messages: println(" " + c("muted", "Продолжаем прошлый чат: %d реплик · /clear — начать новый" % len(self.messages))) println() if not self.cfg.get("tips_seen"): println(" " + c("accent2", "Совет: ") + c("text", "/tips — как работать с агентами и скиллами, чтобы не запутаться")) println() self.cfg["tips_seen"] = True save_cfg(self.cfg) try: while True: try: line = self.status_and_prompt() except EOFError: break line = (line or "").strip() CANCEL.clear() if not line: continue try: if line.startswith("/"): if not self.handle(line): break elif line.startswith("@"): self.run_task(lambda: self.mention(line)) else: rest, paths = split_image_paths(line, self.cwd) if paths: self.attach_paths(paths) if not rest: continue # только фото: отправим вместе с текстом self.run_task(lambda: self.turn(rest)) except KeyboardInterrupt: SCREEN.set_dyn([]) println() self.persist_chat() finally: self.persist_chat() println(" " + c("muted", "Пока! 👋")) def ensure_bun_before_start(): """Перед открытием чата: если Bun ещё не установлен — сначала скачиваем его с прогрессом.""" try: enabled = load_cfg()["settings"].get("bun_auto", True) except Exception: enabled = True if not enabled or bun_exe(): return println(" " + c("accent", "Скачиваю Bun — среду для скриптов JavaScript и TypeScript")) println(" " + c("muted", "Один раз, в %s · github.com/oven-sh/bun · проверка SHA256" % os.path.dirname(bun_dirs()[0]))) println() t0 = time.time() def progress(done, total): if total: frac = min(1.0, done / total) filled = int(30 * frac) bar = "█" * filled + "░" * (30 - filled) txt = "%3d%% %s / %s" % (int(frac * 100), fmt_bytes(done), fmt_bytes(total)) else: bar, txt = "…", fmt_bytes(done) speed = done / max(0.1, time.time() - t0) SCREEN.set_dyn([" [%s] %s %s/с" % (bar, txt, fmt_bytes(speed))]) try: path = install_bun(progress=progress) RUNTIMES.clear() SCREEN.set_dyn([]) println(" " + c("ok", "✓ Bun установлен: ") + c("text", path)) BUN_STATE.update(state="ok", msg=path, shown=True) time.sleep(1.0) except KeyboardInterrupt: SCREEN.set_dyn([]) println(" " + c("warn", "⚠ Установка Bun прервана — запускаюсь без него")) BUN_STATE.update(state="error", msg="прервано", shown=True) time.sleep(1.0) except Exception as e: SCREEN.set_dyn([]) println(" " + c("err", "✗ Bun не скачался: " + str(e))) println(" " + c("muted", "Запускаюсь без него, попробую снова в фоне.")) BUN_STATE.update(state="error", msg=str(e), shown=True) time.sleep(2.0) # ════════════════════════════════════════════════════════════════════════════ # ТРЕЙ (Windows): /tray прячет окно терминала, иконка в трее возвращает его # Программа продолжает работать, пока её не закроют пунктом «Выйти». # ════════════════════════════════════════════════════════════════════════════ TRAY = {"started": False, "ev": None, "hwnd": 0, "nid": None, "keep": []} _TRAY_MSG = 0x8001 # WM_APP+1: сообщения от иконки в трее _TRAY_ID = 1 def _tray_console(): import ctypes return ctypes.windll.kernel32.GetConsoleWindow() def _tray_thread(ready): import ctypes from ctypes import wintypes u32 = ctypes.windll.user32 sh = ctypes.windll.shell32 k32 = ctypes.windll.kernel32 LRESULT = ctypes.c_ssize_t WNDPROC = ctypes.WINFUNCTYPE(LRESULT, wintypes.HWND, wintypes.UINT, wintypes.WPARAM, wintypes.LPARAM) u32.DefWindowProcW.restype = LRESULT u32.DefWindowProcW.argtypes = [wintypes.HWND, wintypes.UINT, wintypes.WPARAM, wintypes.LPARAM] class WNDCLASSW(ctypes.Structure): _fields_ = [("style", wintypes.UINT), ("lpfnWndProc", WNDPROC), ("cbClsExtra", ctypes.c_int), ("cbWndExtra", ctypes.c_int), ("hInstance", wintypes.HINSTANCE), ("hIcon", wintypes.HICON), ("hCursor", wintypes.HANDLE), ("hbrBackground", wintypes.HANDLE), ("lpszMenuName", wintypes.LPCWSTR), ("lpszClassName", wintypes.LPCWSTR)] class NOTIFYICONDATAW(ctypes.Structure): _fields_ = [("cbSize", wintypes.DWORD), ("hWnd", wintypes.HWND), ("uID", wintypes.UINT), ("uFlags", wintypes.UINT), ("uCallbackMessage", wintypes.UINT), ("hIcon", wintypes.HICON), ("szTip", ctypes.c_wchar * 128), ("dwState", wintypes.DWORD), ("dwStateMask", wintypes.DWORD), ("szInfo", ctypes.c_wchar * 256), ("uVersion", wintypes.UINT), ("szInfoTitle", ctypes.c_wchar * 64), ("dwInfoFlags", wintypes.DWORD), ("guidItem", ctypes.c_byte * 16), ("hBalloonIcon", wintypes.HICON)] def _menu(hwnd): menu = u32.CreatePopupMenu() u32.AppendMenuW(menu, 0, 1, "Открыть FlynCode") u32.AppendMenuW(menu, 0, 2, "Выйти") pt = wintypes.POINT() u32.GetCursorPos(ctypes.byref(pt)) u32.SetForegroundWindow(hwnd) cmd = u32.TrackPopupMenu(menu, 0x0002 | 0x0100, pt.x, pt.y, 0, hwnd, None) # RIGHTBUTTON | RETURNCMD u32.DestroyMenu(menu) if cmd == 1: tray_show() elif cmd == 2: _tray_quit() def _wndproc(hwnd, msg, wp, lp): if msg == _TRAY_MSG: ev = lp & 0xFFFF if ev in (0x0202, 0x0203): # WM_LBUTTONUP / WM_LBUTTONDBLCLK tray_show() elif ev == 0x0205: # WM_RBUTTONUP _menu(hwnd) return 0 if msg == 0x0002: # WM_DESTROY u32.PostQuitMessage(0) return 0 return u32.DefWindowProcW(hwnd, msg, wp, lp) hinst = k32.GetModuleHandleW(None) wc = WNDCLASSW() wc.lpfnWndProc = WNDPROC(_wndproc) wc.hInstance = hinst wc.lpszClassName = "FlynCodeTray" TRAY["keep"].append((wc, _wndproc)) # держим ссылки, иначе сборщик мусора их уничтожит u32.RegisterClassW(ctypes.byref(wc)) hwnd = u32.CreateWindowExW(0, "FlynCodeTray", "FlynCode", 0, 0, 0, 0, 0, ctypes.c_void_p(-3), None, hinst, None) # HWND_MESSAGE: невидимое окно nid = NOTIFYICONDATAW() nid.cbSize = ctypes.sizeof(nid) nid.hWnd = hwnd nid.uID = _TRAY_ID nid.uFlags = 0x1 | 0x2 | 0x4 # NIF_MESSAGE | NIF_ICON | NIF_TIP nid.uCallbackMessage = _TRAY_MSG nid.hIcon = u32.LoadIconW(None, ctypes.c_void_p(32512)) # стандартная иконка приложения nid.szTip = "FlynCode · клик — открыть" sh.Shell_NotifyIconW(0, ctypes.byref(nid)) # NIM_ADD TRAY.update(hwnd=hwnd, nid=nid) ready.set() msg = wintypes.MSG() while u32.GetMessageW(ctypes.byref(msg), None, 0, 0) != 0: u32.TranslateMessage(ctypes.byref(msg)) u32.DispatchMessageW(ctypes.byref(msg)) def tray_show(): """Вернуть окно терминала на экран и поднять его.""" import ctypes hwnd = _tray_console() if hwnd: ctypes.windll.user32.ShowWindow(hwnd, 5) # SW_SHOW ctypes.windll.user32.SetForegroundWindow(hwnd) def _tray_quit(): import ctypes tray_show() try: SCREEN.stop() except Exception: pass nid = TRAY.get("nid") if nid is not None: ctypes.windll.shell32.Shell_NotifyIconW(2, ctypes.byref(nid)) # NIM_DELETE os._exit(0) def tray_hide(): """Спрятать окно терминала в трей. True — получилось.""" import ctypes hwnd = _tray_console() if not hwnd: return False if not TRAY["started"]: TRAY["started"] = True TRAY["ev"] = threading.Event() threading.Thread(target=_tray_thread, args=(TRAY["ev"],), daemon=True).start() TRAY["ev"].wait(5) nid = TRAY.get("nid") if nid is not None: nid.uFlags |= 0x10 # NIF_INFO: всплывающая подсказка nid.szInfoTitle = "FlynCode" nid.szInfo = "Работает в трее. Клик по иконке — открыть." nid.dwInfoFlags = 1 # NIIF_INFO ctypes.windll.shell32.Shell_NotifyIconW(1, ctypes.byref(nid)) # NIM_MODIFY ctypes.windll.user32.ShowWindow(hwnd, 0) # SW_HIDE return True def cmd_tray(): if not IS_WIN: println(" " + c("warn", "Трей доступен только в Windows.")) return if os.environ.get("WT_SESSION"): println(" " + c("warn", "В Windows Terminal окно не спрятать. Запусти FlynCode в обычной консоли " "(cmd или Windows PowerShell).")) return if not tray_hide(): println(" " + c("warn", "Не удалось спрятать окно в трей.")) def main(): setup_console() if not sys.stdin.isatty() or not sys.stdout.isatty(): print("FlynCode нужно запускать в настоящем терминале.") sys.exit(1) SCREEN.start() for _name in ("SIGTERM", "SIGHUP"): if hasattr(signal, _name): try: signal.signal(getattr(signal, _name), lambda *a: sys.exit(0)) except Exception: pass try: ensure_bun_before_start() App().run() except (KeyboardInterrupt, EOFError): pass finally: SCREEN.stop() if __name__ == "__main__": main()