#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""MCP-сервер MARFOR: доступ ассистента (Claude и любого MCP-клиента) к MARFOR по HTTP API.

Работает от имени пользователя и видит ровно то, что видит он сам. Основной вход —
персональный токен доступа: заголовок `Authorization: Bearer …` уходит с каждым
запросом к своему стенду, cookie-сессии нет, отзыв токена на сайте действует сразу.
Токен берётся из переменной MARFOR_TOKEN либо из файла ~/.marfor_mcp/token.json
(ключ файла — адрес стенда, чтобы токен не ушёл на другой адрес). Получить его
проще всего подтверждением в браузере — секрет не проходит ни через чат, ни через
конфигурацию клиента, ни через историю оболочки:

    python3 server.py --login

Остальные команды — `python3 server.py --help`, инструкция — https://marfor.pro/mcp.

Запасной путь — вход паролем (MARFOR_EMAIL + пароль) для стендов без токенов: сервер сам
делает POST /login и держит cookie в памяти. Работает ТОЛЬКО с явно заданным
MARFOR_BASE_URL: стенд по умолчанию — marfor.pro, и без этого условия пароль от другого
стенда ушёл бы туда. Пароль сервер НЕ хранит и НЕ пишет на диск: берёт из Keychain macOS
(`security find-generic-password -s marfor-mcp -a <email> -w`; на других системах Keychain
нет) либо из переменной окружения MARFOR_PASSWORD. Запись в Keychain заводит сам
пользователь:

    security add-generic-password -s marfor-mcp -a <email> -w

Зависимости: только стандартная библиотека Python 3.8+.

Протокол: MCP поверх stdio (JSON-RPC 2.0, по одному сообщению на строку).
В stdout уходит ТОЛЬКО протокол, все логи — в stderr.
"""

import argparse
import copy
import hashlib
import http.client
import json
import os
import re
import shlex
import socket
import subprocess
import sys
import tempfile
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
import uuid
from http.cookiejar import CookieJar

SERVER_NAME = 'marfor'
SERVER_VERSION = '0.4.0'
SUPPORTED_PROTOCOLS = ('2025-06-18', '2025-03-26', '2024-11-05')

# Публичный стенд. По нему же решается, какие наборы инструментов видны по умолчанию
# и есть ли смысл спрашивать сайт о новой версии сервера.
PUBLIC_BASE = 'https://marfor.pro'
PUBLIC_HOSTS = ('marfor.pro', 'www.marfor.pro')

BASE_URL = (os.environ.get('MARFOR_BASE_URL') or PUBLIC_BASE).strip().rstrip('/')
# Задан ли стенд ЯВНО. Парольный вход без этого запрещён: умолчание — публичный marfor.pro,
# и пароль от другого стенда ушёл бы туда POST-ом на /login. Так вышло бы у любого сценария,
# который импортирует этот файл и держался на прежнем умолчании (до 0.4.0 оно было другим).
BASE_URL_EXPLICIT = bool((os.environ.get('MARFOR_BASE_URL') or '').strip())
EMAIL = os.environ.get('MARFOR_EMAIL', '')
KEYCHAIN_SERVICE = os.environ.get('MARFOR_KEYCHAIN_SERVICE', 'marfor-mcp')
HTTP_TIMEOUT = int(os.environ.get('MARFOR_HTTP_TIMEOUT', '180'))
# Генерация BI-страницы синхронная и ДОЛГАЯ: план запросов + сборка HTML
# чанками по ~1500 токенов (до 20 чанков по ~45с) — минуты, в пределе ~20.
# nginx на mar держит запрос 1800с, обычного MARFOR_HTTP_TIMEOUT не хватит.
BI_GENERATE_TIMEOUT = int(os.environ.get('MARFOR_BI_GENERATE_TIMEOUT', '1500'))
JOB_DIR = os.path.expanduser('~/.marfor_mcp/jobs')
PAGE_DIR = os.path.expanduser('~/.marfor_mcp/pages')
CONFIG_DIR = os.path.expanduser('~/.marfor_mcp')
TOKEN_FILE = os.path.join(CONFIG_DIR, 'token.json')        # {"<адрес стенда>": "<токен>"}
PENDING_FILE = os.path.join(CONFIG_DIR, 'pending.json')    # начатый, но не завершённый вход

# Форма токена. Проверяется ДО того, как он попадёт в заголовок: http.client на мусоре
# в заголовке бросает ValueError с самим значением в тексте — токен утёк бы в чат.
# fullmatch, а не match с «$»: у «$» в Python есть лазейка — он совпадает и перед
# завершающим переводом строки, то есть «токен\n» прошёл бы проверку.
TOKEN_RE = re.compile(r'marfor_pat_[A-Za-z0-9_-]{20,200}')

# Сколько секунд помнить токен, который стенд отверг (401 или 429 too_many_attempts). Пока
# помним — с тем же значением в сеть не ходим: у стенда лимит неудач на IP, и отозванный
# токен, повторяемый на каждый вызов инструмента, выбрал бы его за минуты. Другое значение
# токена идёт в сеть сразу.
REFUSED_TOKEN_TTL = 60

# Батч длиной в часы не влезает в один вызов инструмента, поэтому идёт фоном.
_JOBS = {}
_JOBS_LOCK = threading.Lock()


def log(*a):
    print(*a, file=sys.stderr, flush=True)


# ─────────────────────── стенд: свой ли адрес, публичный ли ───────────────────

def _base_host():
    try:
        u = urllib.parse.urlsplit(BASE_URL)
        return u.scheme, (u.hostname or '').lower()
    except ValueError:              # кривой адрес вроде https://[::1 — urlsplit бросает
        return '', ''


def _is_public_stand():
    return _base_host()[1] in PUBLIC_HOSTS


def _is_base_url(url):
    """Адрес принадлежит СВОЕМУ стенду? Голого startswith(BASE_URL) мало: под него
    подходят и https://marfor.pro.evil.example, и https://marfor.pro@evil.example, и
    соседний порт 127.0.0.1:50001 при стенде 127.0.0.1:5000. После адреса стенда
    обязан идти путь или строка запроса."""
    return url == BASE_URL or url.startswith(BASE_URL + '/') or url.startswith(BASE_URL + '?')


def _base_url_problem():
    """Текст отказа, если стенд задан небезопасно; None — всё в порядке.
    По http токен и пароль ушли бы открытым текстом, поэтому http — только для
    localhost/127.0.0.1 (подставной стенд тестов, локальная разработка)."""
    scheme, host = _base_host()
    if scheme == 'https' and host:
        return None
    if scheme == 'http' and host in ('localhost', '127.0.0.1'):
        return None
    return ('MARFOR_BASE_URL=%s — небезопасный адрес: нужен https://… (http разрешён только '
            'для localhost и 127.0.0.1). Сервер не запускается, чтобы токен не ушёл по сети '
            'открытым текстом. Адрес по умолчанию — %s.' % (BASE_URL or '(пусто)', PUBLIC_BASE))


# ───────────────────────── токен доступа: где лежит и как пишется ─────────────

def _secure_dir():
    """Каталог ~/.marfor_mcp с правами 0700. На Windows chmod прав не ограничивает —
    там каталог и так лежит в профиле пользователя, шаг молча пропускаем."""
    os.makedirs(CONFIG_DIR, exist_ok=True)
    if os.name != 'nt':
        try:
            os.chmod(CONFIG_DIR, 0o700)
        except OSError:
            pass


def _write_private_json(path, obj):
    """Атомарная запись JSON с правами 0600: временный файл рядом + os.replace.
    Обрыв на середине не оставит ни полфайла, ни секрета, открытого на чтение всем."""
    _secure_dir()
    fd, tmp = tempfile.mkstemp(prefix='.tmp_', suffix='.json', dir=CONFIG_DIR)
    try:
        with os.fdopen(fd, 'w', encoding='utf-8') as f:
            json.dump(obj, f, ensure_ascii=False, indent=1)
        if os.name != 'nt':
            os.chmod(tmp, 0o600)
        os.replace(tmp, path)
    except BaseException:
        try:
            os.unlink(tmp)
        except OSError:
            pass
        raise


def _read_json(path):
    try:
        with open(path, encoding='utf-8') as f:
            js = json.load(f)
        return js if isinstance(js, dict) else {}
    except (OSError, ValueError):
        return {}


def _find_token():
    """(токен, источник). Порядок: MARFOR_TOKEN → token.json с ключом по адресу стенда.
    Ключ по стенду нужен, чтобы токен marfor.pro не ушёл на другой адрес, когда
    MARFOR_BASE_URL поменяли, а файл остался прежним."""
    tok = (os.environ.get('MARFOR_TOKEN') or '').strip()
    if tok:
        return tok, 'env'
    tok = _read_json(TOKEN_FILE).get(BASE_URL)
    if isinstance(tok, str) and tok.strip():
        return tok.strip(), 'file'
    return '', None


def _save_token(token):
    data = _read_json(TOKEN_FILE)
    data[BASE_URL] = token
    _write_private_json(TOKEN_FILE, data)


def _forget_token():
    """Убрать из файла токен ЭТОГО стенда (токены других стендов остаются). True — был."""
    data = _read_json(TOKEN_FILE)
    if BASE_URL not in data:
        return False
    del data[BASE_URL]
    if data:
        _write_private_json(TOKEN_FILE, data)
    else:
        try:
            os.unlink(TOKEN_FILE)
        except OSError:
            pass
    return True


_NT_PLAIN_ARG = re.compile(r'[A-Za-z0-9_.:/=@%+,-]+')


def _shq(s, path=False):
    """Аргумент команды, которую человек или ассистент скопирует и выполнит.
    POSIX — shlex.quote. Windows: путь ВСЕГДА в двойных кавычках и с прямыми слэшами
    ("C:/Users/…/server.py"). Такую запись одинаково читают Git Bash (им исполняет команды
    Claude Code на Windows; путь без кавычек он лишал обратных слэшей, и сервер
    регистрировался с несуществующим адресом C:Usersivan.marfor_mcpserver.py), PowerShell
    и cmd. Остальные аргументы берутся в кавычки, только когда в них есть что-то кроме
    букв, цифр и безобидных знаков: «claude» в кавычках PowerShell командой не считает."""
    if os.name != 'nt':
        return shlex.quote(s)
    if path:
        s = s.replace('\\', '/')
    if path or not _NT_PLAIN_ARG.fullmatch(s):
        return '"%s"' % s.replace('"', '\\"')
    return s


def _self_cmd(*extra):
    """Команда запуска ЭТОГО файла ЭТИМ интерпретатором, с абсолютными путями: подсказка
    «python3 server.py --login» не сработает, если человек стоит в другом каталоге.
    Стенд не по умолчанию едет вместе с командой: без переменной вход по подсказке начался
    бы на marfor.pro, токен лёг бы в файл под чужим ключом, а сервер снова ответил бы
    «токен не найден» — по кругу. Записи «переменная перед командой», общей для оболочек
    Windows, нет, поэтому там то же самое говорит словами _win_notes()."""
    parts = [_shq(sys.executable or 'python3', path=True),
             _shq(os.path.abspath(__file__), path=True)] + [_shq(x) for x in extra]
    if BASE_URL != PUBLIC_BASE and os.name != 'nt':
        parts.insert(0, 'MARFOR_BASE_URL=' + shlex.quote(BASE_URL))
    return ' '.join(parts)


def _win_notes():
    """Пояснение к напечатанным командам для Windows; на остальных системах — пусто.
    Команда там начинается со строки в кавычках: cmd и Git Bash исполняют её как есть, а
    PowerShell считает строку в начале выражением и без «&» отвечает ошибкой разбора."""
    if os.name != 'nt':
        return ''
    out = ('\nWindows: в cmd и Git Bash команды верны как напечатаны; в PowerShell поставьте '
           'перед командой знак & и пробел.')
    if BASE_URL != PUBLIC_BASE:
        out += (' Перед запуском задайте переменную окружения MARFOR_BASE_URL=%s (стенд не по '
                'умолчанию; без неё вход уйдёт на %s).' % (BASE_URL, PUBLIC_BASE))
    return out


def _no_credentials_text():
    return ('Вход в MARFOR не настроен: токен доступа для %s не найден (нет ни переменной '
            'MARFOR_TOKEN, ни записи в %s).\n'
            'Войти подтверждением в браузере — в терминале:\n'
            '  %s\n'
            'Если команды запускает ассистент (терминала с вводом нет): %s — показать '
            'пользователю ссылку и код и сразу запустить %s (она сама ждёт «Разрешить»; код '
            'возврата 2 — повторить).\n'
            'Перезапускать клиент после входа не нужно. Инструкция: %s/mcp%s'
            % (BASE_URL, TOKEN_FILE, _self_cmd('--login'), _self_cmd('--login-start'),
               _self_cmd('--login-finish'), PUBLIC_BASE, _win_notes()))


def _password_needs_base_text():
    return ('Парольный вход (MARFOR_EMAIL) работает только с явно заданным MARFOR_BASE_URL. '
            'Сейчас переменная не задана, стенд по умолчанию — %s, и пароль от другого стенда '
            'ушёл бы туда. Пароль НЕ отправлен.\n'
            'Нужен другой стенд — задайте рядом с MARFOR_EMAIL переменную '
            'MARFOR_BASE_URL=https://<адрес вашего стенда>.\n'
            'Для %s войдите токеном, подтверждением в браузере:\n  %s%s'
            % (PUBLIC_BASE, PUBLIC_BASE, _self_cmd('--login'), _win_notes()))


def _token_refusal(code, js):
    """Код отказа подлинности токена (401 или 429) либо None. 401 — всегда отказ. 429 с
    code=too_many_attempts хук Bearer отдаёт ТОЛЬКО неверному токену — после 30 неудач с адреса
    за 10 минут; верный токен проходит всегда. Это тот же отказ, и повторять его так же
    бессмысленно. Прочие 429 (лимиты самих ручек) к токену отношения не имеют."""
    if code == 401:
        return 401
    if code == 429 and isinstance(js, dict) and js.get('code') == 'too_many_attempts':
        return 429
    return None


def _token_rejected_text(src=None, http=401):
    # Токен из переменной важнее файла: без этой оговорки человек вошёл бы заново, токен
    # лёг бы в файл, а сервер продолжал бы слать старый из конфигурации клиента — по кругу.
    env_note = ('Этот токен задан переменной MARFOR_TOKEN в конфигурации MCP-клиента, и она '
                'важнее файла: уберите её оттуда (или замените в ней токен), иначе новый вход '
                'не подействует.\n') if src == 'env' else ''
    head = 'Токен недействителен: он отозван или истёк (HTTP 401 от %s).\n' % BASE_URL
    if http == 429:
        head = ('Токен недействителен: он отозван или истёк. Стенд %s ответил HTTP 429 '
                '(too_many_attempts): с этого адреса пришло слишком много запросов с неверным '
                'токеном, и проверять его стенд перестал. Ждать не нужно: новый токен заработает '
                'сразу.\n' % BASE_URL)
    return (head + env_note +
            ('Войдите заново — в терминале:\n'
             '  %s\n'
             'Если команды запускает ассистент: %s, показать пользователю ссылку и код и '
             'сразу запустить %s (она сама ждёт «Разрешить»; код возврата 2 — повторить).\n'
             'Перезапускать клиент не нужно. Токены видны и отзываются на %s/account, '
             'раздел «Подключение Claude (MCP)».%s'
             % (_self_cmd('--login'), _self_cmd('--login-start'),
                _self_cmd('--login-finish'), PUBLIC_BASE, _win_notes())))


# ──────────────── ответы стенда: «нет сессии» и ошибки сети словами ───────────

_UNAUTH_MESSAGES = ('Требуется авторизация', 'Не авторизован', 'Необходима авторизация')


def _json_or_none(raw):
    try:
        return json.loads(raw.decode('utf-8', 'replace'))
    except Exception:
        return None


def _looks_unauthorized(js):
    """Отказ «нет сессии» — по РАЗОБРАННОМУ ответу. Искать фразу в сыром тексте нельзя:
    стенд отдаёт кириллицу \\u-экранированной, и такая проверка не срабатывала ни разу."""
    if not isinstance(js, dict) or js.get('success') is not False:
        return False
    return str(js.get('message') or '').strip().startswith(_UNAUTH_MESSAGES)


def _session_stale(code, hdrs, js):
    """Протухла ли парольная сессия. 403 сюда НЕ входит: это законный отказ («Нет
    доступа к проекту»), повторный вход его не вылечит — только лишний раз сходит на /login."""
    if code == 401:
        return True
    loc = (hdrs.get('Location') or '') if hdrs else ''
    if code in (301, 302, 303, 307, 308) and '/login' in loc:
        return True
    return _looks_unauthorized(js)


def _net_error_text(e, url, timeout):
    """Сбой сети человеческим языком. Пользователь здесь один на один с ассистентом,
    и строка «<urlopen error [Errno 61] Connection refused>» ему ничего не скажет."""
    reason = getattr(e, 'reason', None) or e
    txt = '%s: %s' % (type(reason).__name__, reason)
    host = urllib.parse.urlsplit(url).netloc or url
    if isinstance(e, http.client.InvalidURL):
        return ('Недопустимый адрес запроса: в пути есть пробел или управляющий символ. '
                'Проверьте идентификаторы и путь, которые передаются инструменту.')
    if 'CERTIFICATE_VERIFY_FAILED' in txt:
        return ('Python не смог проверить сертификат %s (CERTIFICATE_VERIFY_FAILED). Обычная '
                'причина на macOS — Python, установленный с python.org, без корневых '
                'сертификатов: откройте папку «Программы → Python 3.x» и запустите «Install '
                'Certificates.command», затем повторите. Другая возможная причина — '
                'корпоративный прокси или антивирус, подменяющий сертификаты.' % host)
    if isinstance(reason, (socket.timeout, TimeoutError)) or 'timed out' in txt:
        return ('%s не ответил за %s с. Проверьте интернет и повторите; для заведомо долгой '
                'операции увеличьте MARFOR_HTTP_TIMEOUT.' % (host, timeout))
    if isinstance(reason, socket.gaierror):
        return ('Не удалось найти адрес %s: проверьте интернет, VPN и значение '
                'MARFOR_BASE_URL.' % host)
    if isinstance(reason, ConnectionRefusedError):
        return ('%s не принимает соединения: сервис недоступен либо адрес в MARFOR_BASE_URL '
                'неверный. Повторите позже.' % host)
    return 'Нет связи с %s (%s). Проверьте интернет, VPN или прокси и повторите.' % (host, txt)


# ─────────────────────────── HTTP-клиент: токен или сессия ───────────────────

class TokenRefused(RuntimeError):
    """Стенд отверг токен: 401 token_invalid либо 429 too_many_attempts (см. _token_refusal) —
    он отозван или истёк. Отдельный класс нужен фоновому батчу: для него это конец задания,
    а не «ошибка очередной правки»."""


class Marfor:
    """Тонкий клиент, два режима. Токен: Bearer в каждом запросе к своему стенду, без
    /login и cookie, отказ — сразу понятным текстом. Пароль: вход через /login, cookie в
    памяти, после протухания сессии — один повторный вход."""

    def __init__(self):
        self.jar = CookieJar()
        self.opener = urllib.request.build_opener(
            urllib.request.HTTPCookieProcessor(self.jar),
            _NoRedirect(),
        )
        # Режим токена и публичные ручки ходят БЕЗ cookie: сессии там нет по построению.
        self.opener_plain = urllib.request.build_opener(_NoRedirect())
        self.logged_in = False
        self.account = None
        self.token = ''
        self.token_source = None
        self.refused = None         # (sha256 отвергнутого токена, когда отвергнут) — см. _is_refused
        self.lock = threading.Lock()

    # -- пароль --
    @staticmethod
    def _password():
        pw = os.environ.get('MARFOR_PASSWORD')
        if pw:
            return pw
        if not EMAIL:
            raise RuntimeError('не задан MARFOR_EMAIL')
        if sys.platform != 'darwin':
            # Keychain и команда security есть только на macOS. Раньше на Linux и Windows здесь
            # получалось «[Errno 2] No such file or directory: 'security'» — и ни слова о том,
            # что делать дальше.
            raise RuntimeError(
                'пароль взять неоткуда: Keychain есть только на macOS, а переменная '
                'MARFOR_PASSWORD не задана. Обычный путь — вход токеном, подтверждением в '
                'браузере:\n  %s\nДля CI и стендов без токенов задайте MARFOR_PASSWORD рядом с '
                'MARFOR_EMAIL.%s' % (_self_cmd('--login'), _win_notes()))
        try:
            out = subprocess.run(
                ['security', 'find-generic-password', '-s', KEYCHAIN_SERVICE,
                 '-a', EMAIL, '-w'],
                capture_output=True, text=True, timeout=20)
        except Exception as e:
            raise RuntimeError(f'не удалось спросить Keychain: {e}')
        if out.returncode != 0:
            raise RuntimeError(
                f'пароль не найден. Заведите запись командой:\n'
                f'  security add-generic-password -s {KEYCHAIN_SERVICE} '
                f'-a {EMAIL} -w')
        return out.stdout.strip('\n')

    # -- низкий уровень --
    def _raw(self, method, path, *, data=None, form=None, headers=None,
             timeout=None, auth=True):
        """auth=False — публичные ручки (вход подтверждением в браузере, version.json):
        токен к ним прикладывать нельзя, стенд ответит на него 401/403 вместо дела."""
        url = path if path.startswith('http') else BASE_URL + path
        body, hdr = None, {'User-Agent': 'marfor-mcp/%s' % SERVER_VERSION}
        if form is not None:
            body = urllib.parse.urlencode(form).encode()
            hdr['Content-Type'] = 'application/x-www-form-urlencoded'
        elif data is not None:
            body = json.dumps(data, ensure_ascii=False).encode()
            hdr['Content-Type'] = 'application/json'
        hdr.update(headers or {})
        # Токен уходит ТОЛЬКО на свой стенд: метод принимает и абсолютные адреса.
        # Редиректы отключены (_NoRedirect), так что и переадресацией его не увести.
        if auth and self.token and _is_base_url(url):
            hdr['Authorization'] = 'Bearer ' + self.token
        opener = self.opener if (auth and not self.token) else self.opener_plain
        req = urllib.request.Request(url, data=body, headers=hdr, method=method)
        t = timeout or HTTP_TIMEOUT
        try:
            resp = opener.open(req, timeout=t)
            return resp.getcode(), resp.headers, resp.read()
        except urllib.error.HTTPError as e:
            try:
                err_body = e.read()
            except Exception:
                err_body = b''
            return e.code, e.headers, err_body
        except (OSError, http.client.HTTPException) as e:
            # URLError, таймауты, обрывы и ошибки TLS — все наследники OSError
            raise RuntimeError(_net_error_text(e, url, t))

    def login(self):
        with self.lock:
            if not BASE_URL_EXPLICIT:
                # Стоит ДО чтения пароля: без явного стенда не трогаем ни Keychain, ни сеть.
                raise RuntimeError(_password_needs_base_text())
            pw = self._password()
            code, hdrs, _ = self._raw('POST', '/login',
                                      form={'email': EMAIL, 'password': pw})
            del pw
            loc = hdrs.get('Location', '') if hdrs else ''
            if code not in (200, 302, 303) or (code == 200):
                # 200 на POST /login = страница входа с ошибкой (редиректа не было)
                if code == 200:
                    raise RuntimeError('вход отклонён: неверный логин или пароль')
            if '/login' in loc:
                raise RuntimeError('вход отклонён: сервер вернул обратно на /login')
            # Проверка входа идёт по /api/account/balance: он отдаёт user_id и email,
            # то есть сразу видно, ПОД КАКИМ аккаунтом мы вошли. /api/projects для
            # этого не годится — 05.08 он падал с «string indices must be integers»
            # и молча выглядел как отказ авторизации.
            code, _, raw = self._raw('GET', '/api/account/balance')
            if code != 200:
                raise RuntimeError(f'после входа /api/account/balance отдал HTTP {code}')
            try:
                js = json.loads(raw.decode('utf-8', 'replace'))
            except Exception:
                raise RuntimeError('после входа /api/account/balance вернул не JSON')
            bal = (js or {}).get('balance') or {}
            if not bal.get('user_id'):
                raise RuntimeError(f'после входа не удалось определить аккаунт: '
                                   f'{str(js)[:200]}')
            self.logged_in = True
            self.account = {'email': bal.get('email') or EMAIL, 'base_url': BASE_URL,
                            'user_id': bal.get('user_id'), 'tariff': bal.get('tariff')}
            log(f'[marfor] вход выполнен: {self.account["email"]} '
                f'(user_id={self.account["user_id"]}) @ {BASE_URL}')
            return self.account

    def token_login(self, tok, src):
        """Режим токена: ни /login, ни cookie. Один GET /api/account/balance проверяет
        токен и заодно говорит, ПОД КАКИМ аккаунтом идёт работа."""
        with self.lock:
            if not TOKEN_RE.fullmatch(tok):
                raise RuntimeError(
                    'Токен из %s не похож на токен MARFOR: он начинается с marfor_pat_ и '
                    'состоит из латинских букв, цифр, «-» и «_». Проверьте, что он скопирован '
                    'целиком и без кавычек, либо войдите заново: %s%s'
                    % ('переменной MARFOR_TOKEN' if src == 'env' else TOKEN_FILE,
                       _self_cmd('--login'), _win_notes()))
            self.token, self.token_source = tok, src
            try:
                # Первое обращение: на «чёрной дыре» вместо сети ждать три минуты незачем
                code, _, raw = self._raw('GET', '/api/account/balance',
                                         timeout=min(HTTP_TIMEOUT, 30))
                js = _json_or_none(raw)
                refusal = _token_refusal(code, js)
                if refusal:
                    self._remember_refused(tok, refusal)
                    raise TokenRefused(_token_rejected_text(src, refusal))
                if _looks_unauthorized(js):
                    raise RuntimeError(
                        'Стенд %s не узнал токен: похоже, вход по токену на нём не включён. '
                        'Проверьте MARFOR_BASE_URL (по умолчанию %s).' % (BASE_URL, PUBLIC_BASE))
                bal = (js or {}).get('balance') if isinstance(js, dict) else None
                if code != 200 or not isinstance(bal, dict) or not bal.get('user_id'):
                    msg = (js or {}).get('message') if isinstance(js, dict) else None
                    raise RuntimeError('проверка токена не удалась: /api/account/balance '
                                       'отдал HTTP %s%s' % (code, (' — %s' % msg) if msg else ''))
            except RuntimeError:
                self.token, self.token_source, self.logged_in = '', None, False
                raise
            self.logged_in = True
            self.account = {'email': bal.get('email'), 'base_url': BASE_URL,
                            'user_id': bal.get('user_id'), 'tariff': bal.get('tariff')}
            log(f'[marfor] токен принят: {self.account["email"]} '
                f'(user_id={self.account["user_id"]}) @ {BASE_URL}')
            return self.account

    def _token_refused(self, http=401):
        # Токен забываем: следующий вызов перечитает переменную и файл — человек мог
        # успеть войти заново, и тогда всё заработает без перезапуска клиента.
        src = self.token_source
        self._remember_refused(self.token, http)
        self.token, self.token_source, self.logged_in = '', None, False
        return TokenRefused(_token_rejected_text(src, http))

    @staticmethod
    def _digest(tok):
        return hashlib.sha256(tok.encode('utf-8', 'replace')).hexdigest()

    def _remember_refused(self, tok, http=401):
        # В памяти остаётся хеш, а не сам токен: отвергнутое значение хранить незачем.
        self.refused = (self._digest(tok), time.monotonic(), http) if tok else None

    def _is_refused(self, tok):
        """Код прежнего отказа, если этот самый токен стенд только что отверг, иначе None.
        С отвергнутым токеном в сеть не идём: отказ повтором не лечится, а у стенда лимит
        неудач на IP (30 за 10 минут). Раньше каждый вызов инструмента заново читал тот же
        token.json и снова получал отказ. Память короткая (REFUSED_TOKEN_TTL) и только на это
        значение: после нового входа токен другой и идёт в сеть сразу."""
        ref = self.refused
        if (ref and ref[0] == self._digest(tok)
                and time.monotonic() - ref[1] < REFUSED_TOKEN_TTL):
            return ref[2] if len(ref) > 2 else 401
        return None

    def ensure(self):
        """Вход, если его ещё не было. Токен ищется КАЖДЫЙ раз заново: сервер обычно уже
        запущен клиентом, когда человек делает --login, и перезапуска это требовать не должно."""
        if self.logged_in:
            return
        tok, src = _find_token()
        if tok:
            refused = self._is_refused(tok)
            if refused:
                raise TokenRefused(_token_rejected_text(src, refused))
            self.token_login(tok, src)
        elif EMAIL:
            self.token, self.token_source = '', None
            self.login()
        else:
            raise RuntimeError(_no_credentials_text())

    def call(self, method, path, *, data=None, params=None, _retry=True,
             timeout=None):
        """GET/POST к /api/*. Пароль: при протухшей сессии — один повторный вход.
        Токен: повторять нечем, отказ сразу превращается в понятный текст."""
        self.ensure()
        if params:
            path = path + ('&' if '?' in path else '?') + urllib.parse.urlencode(params)
        code, hdrs, raw = self._raw(method, path, data=data, timeout=timeout)
        text = raw.decode('utf-8', 'replace')
        try:
            js = json.loads(text)
        except Exception:
            js = {'_raw': text[:4000], '_not_json': True}
        if self.token:
            refusal = _token_refusal(code, js)
            if refusal:
                raise self._token_refused(refusal)
            return code, js
        if _retry and _session_stale(code, hdrs, js):
            log('[marfor] сессия протухла — повторный вход')
            self.logged_in = False
            self.login()
            return self.call(method, path, data=data, _retry=False,
                             timeout=timeout)
        return code, js

    def fetch_raw(self, path, *, timeout=None, _retry=True):
        """GET, отдающий тело КАК ЕСТЬ (HTML AI-страниц — это не JSON)."""
        self.ensure()
        code, hdrs, raw = self._raw('GET', path, timeout=timeout)
        if self.token:
            refusal = _token_refusal(code, _json_or_none(raw) if code == 429 else None)
            if refusal:
                raise self._token_refused(refusal)
            return code, raw
        js = _json_or_none(raw) if raw[:1] == b'{' else None
        if _retry and _session_stale(code, hdrs, js):
            log('[marfor] сессия протухла — повторный вход')
            self.logged_in = False
            self.login()
            return self.fetch_raw(path, timeout=timeout, _retry=False)
        return code, raw


class _NoRedirect(urllib.request.HTTPRedirectHandler):
    """Редирект на /login должен быть виден как признак «не авторизован»."""

    def redirect_request(self, req, fp, code, msg, headers, newurl):
        return None


def _count_projects(js):
    if isinstance(js, list):
        return len(js)
    if isinstance(js, dict):
        for k in ('projects', 'data', 'items'):
            if isinstance(js.get(k), list):
                return len(js[k])
    return None


MF = Marfor()


# ───────────────────────────── прикладные операции ────────────────────────────

DRIVER_DIMS = ['region_to', 'channel_new', 'segment_new', 'category']


def apply_edit(session_id, metric, filters, new_value, base_value=None,
               wait=True, poll_sec=2, timeout_sec=1800, pivot_dimensions=None):
    """Одна правка моделирования — тем же путём, что и правка ячейки в UI."""
    payload = {
        'metric': metric,
        'mode': 'slices',
        'filters': filters,
        'filter_fields': list(filters.keys()),
        'filter_values': [],
        'time_period': [],
        'is_total': False,
        'base_value': base_value,      # None → сервер посчитает базу сам
        'new_value': new_value,
        'recursive': True,
        'apply_ui_filters': True,
        'is_chain_constant': False,
        'mmm_slice_key': None,         # ε берётся построчно из колонки elasticity
        'pivot_dimensions': pivot_dimensions or DRIVER_DIMS,
    }
    code, js = MF.call('POST', f'/api/modeling_task/{session_id}',
                       data={'action': 'apply', 'payload': payload})
    if not isinstance(js, dict):
        return {'ok': False, 'message': f'HTTP {code}: неожиданный ответ'}
    if js.get('busy'):
        return {'ok': False, 'busy': True,
                'message': js.get('message') or js.get('reason') or 'сервер занят'}
    if not js.get('success') or not js.get('task_id'):
        return {'ok': False, 'message': js.get('message') or js.get('error') or str(js)[:300]}
    task_id = js['task_id']
    if not wait:
        return {'ok': True, 'task_id': task_id, 'pending': True}
    t0 = time.time()
    while time.time() - t0 < timeout_sec:
        time.sleep(poll_sec)
        _, st = MF.call('GET', f'/api/modeling_task_status/{task_id}')
        if not isinstance(st, dict):
            continue
        status = st.get('task_status')
        if status == 'completed':
            return {'ok': st.get('success') is not False, 'task_id': task_id,
                    'seconds': round(time.time() - t0, 1),
                    'message': st.get('message') or 'ok'}
        if status in ('error', 'failed'):
            return {'ok': False, 'task_id': task_id,
                    'seconds': round(time.time() - t0, 1),
                    'message': st.get('error') or st.get('message') or 'ошибка задачи'}
    return {'ok': False, 'task_id': task_id, 'message': f'таймаут {timeout_sec}с'}


def _query_filters(filters):
    """У /api/query формат фильтра — {'колонка': {'values': [...], 'mode': 'include'}}.
    Плоский список он МОЛЧА игнорирует (`if not isinstance(fval, dict): continue`) и
    считает по всей таблице — так 05.08 «russia Q3» превратилось в 4,87 млрд за всю
    историю. У apply_modeling_changes формат обратный (плоские значения), поэтому
    нормализуем здесь, а не в вызывающем коде."""
    out = {}
    for k, v in (filters or {}).items():
        if isinstance(v, dict) and 'values' in v:
            out[k] = {'values': list(v['values']), 'mode': v.get('mode', 'include')}
        elif isinstance(v, (list, tuple, set)):
            out[k] = {'values': list(v), 'mode': 'include'}
        elif v is not None:
            out[k] = {'values': [v], 'mode': 'include'}
    return out


def query(session_id, metrics, dimensions=None, filters=None,
          date_from=None, date_to=None, data_source='scenario'):
    body = {'metrics': metrics, 'dimensions': dimensions or [],
            'filters': _query_filters(filters), 'data_source': data_source}
    if date_from:
        body['date_from'] = date_from
    if date_to:
        body['date_to'] = date_to
    code, js = MF.call('POST', f'/api/query/{session_id}', data=body)
    return js


_DERIVED_CACHE = {}


def _derived_formulas(session_id):
    """{имя вычисляемой метрики: формула} из «Метрик и правил» сессии."""
    if session_id not in _DERIVED_CACHE:
        _, js = MF.call('GET', f'/api/derived_metrics/{session_id}')
        lst = (js or {}).get('derived_metrics') or []
        _DERIVED_CACHE[session_id] = {
            d['name']: d['formula'] for d in lst
            if isinstance(d, dict) and d.get('name') and d.get('formula')}
    return _DERIVED_CACHE[session_id]


def _expand_formula(session_id, metric, depth=0):
    """Развернуть метрику до выражения только из базовых колонок.

    Драйвер сценария сплошь и рядом вычисляемый («reatr. Рекламные расходы без НДС»
    = (ads_cost_reattributed + comission_cost) − nds_cost_reattributed), своей
    колонки у него нет, и /api/query по имени отдаёт пусто. Разворачиваем формулу
    рекурсивно: компонент сам может быть вычисляемым."""
    import re
    forms = _derived_formulas(session_id)
    expr = forms.get(metric)
    if not expr:
        return f'[[{metric}]]'
    if depth > 5:
        return expr
    def sub(m):
        return '(' + _expand_formula(session_id, m.group(1), depth + 1) + ')'
    return re.sub(r'\[\[(.+?)\]\]', sub, expr)


def _metric_sum(session_id, metric, filters):
    """Текущее значение метрики в срезе — для сухого прогона и коэффициентов.

    Компоненты суммируются по отдельности, и только потом считается выражение:
    для метрики-отношения это даёт Σn/Σd, а не сумму построчных отношений."""
    import re
    expr = _expand_formula(session_id, metric)
    names = sorted(set(re.findall(r'\[\[(.+?)\]\]', expr)))
    if not names:
        return None
    js = query(session_id, names, [], filters)
    rows = js.get('data') if isinstance(js, dict) else None
    if not rows:
        return None
    sums, got = {}, False
    for r in rows:
        for k, v in r.items():
            if k in names and isinstance(v, (int, float)):
                sums[k] = sums.get(k, 0.0) + float(v)
                got = True
    if not got:
        return None
    filled = re.sub(r'\[\[(.+?)\]\]',
                    lambda m: repr(sums.get(m.group(1), 0.0)), expr)
    if not re.fullmatch(r'[0-9eE_.+\-*/() ]+', filled):
        return None                      # в формуле что-то кроме арифметики — не считаем
    try:
        return float(eval(filled, {'__builtins__': {}}, {}))
    except Exception:
        return None


# ─────────── массовое применение: правки страницы, одним запросом ─────────────
# Модельная математика живёт в MARFOR, не здесь. Коннектор берёт значения по
# срезам и разворачивает их в payload'ы РОВНО того формата, который строит
# страница в режиме «Массовые изменения», — включая mmm_slice_key и
# pivot_dimensions. Ручка /api/apply_modeling_bulk исполняет каждый payload той
# же задачей, что и одиночную правку (мутация на срез, не одна на всех), а
# батчит только цепочку метрик и пересчёт бэндов — по разу на весь прогон.
#
# Так было не всегда: сначала коннектор считал множители сам, потом ручка
# считала отклик своей копией формулы. Оба раза числа расходились с ручным
# вводом — во второй раз на адстоке (325 370 против 319 505). Отсюда правило:
# один и тот же код МАЛО, нужен ещё и тот же запрос.


DEFAULT_PIVOT_DIMS = ['region_to', 'channel_new', 'segment_new', 'category']
TIME_KEYS = ('year', 'month', 'quarter', 'halfyear')


def build_change(metric, slice_dict, value, pivot_dimensions):
    """Payload ровно в том формате, который страница строит в режиме
    «Массовые изменения» (_queueOnly → {action:'apply', payload}).

    Ключевое — mmm_slice_key и pivot_dimensions: адсток по ним решает, по каким
    срезам копится запас бюджета. Подменять их нельзя, иначе тот же код даст
    другие числа (вердикт шлюза 05.08: 325 370 против 319 505).
    Страница строит ключ как значения фильтров строки через «|».
    """
    filters, dims = {}, list(pivot_dimensions or DEFAULT_PIVOT_DIMS)
    for k, v in (slice_dict or {}).items():
        filters[k] = [str(v)] if k in TIME_KEYS else v
    key_parts = [str(slice_dict[d]) for d in dims if d in (slice_dict or {})]
    return {
        'metric': metric,
        'mode': 'slices',
        'filters': filters,
        'filter_fields': list(filters.keys()),
        'filter_values': [],
        'time_period': [],
        'is_total': False,
        'base_value': None,       # базу считает сервер — клиент её знать не обязан
        'new_value': float(value),
        'recursive': True,
        'apply_ui_filters': True,
        'is_chain_constant': False,
        'mmm_slice_key': '|'.join(key_parts) if key_parts else None,
        'pivot_dimensions': dims,
    }


def apply_values(session_id, changes, scope, dry_run=True):
    payload = {'changes': changes, 'scope': scope or {}, 'dry_run': bool(dry_run)}
    return MF.call('POST', f'/api/apply_modeling_bulk/{session_id}', data=payload)


# ────────────────────────────── фоновые батчи ─────────────────────────────────

def _job_path(job_id):
    os.makedirs(JOB_DIR, exist_ok=True)
    return os.path.join(JOB_DIR, f'{job_id}.jsonl')


def _job_write(job_id, rec):
    with open(_job_path(job_id), 'a', encoding='utf-8') as f:
        f.write(json.dumps(rec, ensure_ascii=False) + '\n')


def batch_worker(job_id, session_id, metric, edits, stop_on_error):
    job = _JOBS[job_id]
    t0 = time.time()
    for i, e in enumerate(edits):
        with _JOBS_LOCK:
            if job['stop']:
                job['status'] = 'stopped'
                break
            job['current'] = i
        ts = time.time()
        fatal = None
        try:
            res = apply_edit(session_id, e.get('metric') or metric,
                             e['filters'], e['new_value'])
        except TokenRefused as ex:
            # Токен отозван или истёк: остальные правки получили бы тот же отказ. Это конец
            # задания при ЛЮБОМ stop_on_error — иначе отозванный токен уходил бы на стенд с
            # каждой оставшейся правкой (60 правок → 61 отказ за 0,1 с) и выбирал бы лимит
            # неудач стенда на IP. Отзыв токена — ещё и способ пользователя остановить запись.
            fatal = str(ex)
            res = {'ok': False, 'message': 'Токен недействителен: отозван или истёк. Задание '
                                           'остановлено, оставшиеся правки не отправлены.'}
        except Exception as ex:
            # Сбой сети посреди батча: без перехвата поток умер бы молча, а задача навсегда
            # осталась бы в статусе running.
            res = {'ok': False, 'message': str(ex)}
        rec = {'i': i, 'label': e.get('label', ''), 'new_value': e['new_value'],
               'seconds': round(time.time() - ts, 1), 'ok': res.get('ok'),
               'message': res.get('message', '')}
        _job_write(job_id, rec)
        with _JOBS_LOCK:
            job['done'] = i + 1
            job['ok'] += 1 if res.get('ok') else 0
            job['fail'] += 0 if res.get('ok') else 1
            job['last'] = rec
        if fatal:
            with _JOBS_LOCK:
                job['status'] = 'error'
                job['error'] = fatal
            break
        if not res.get('ok') and stop_on_error:
            with _JOBS_LOCK:
                job['status'] = 'failed'
            break
    with _JOBS_LOCK:
        if job['status'] == 'running':
            job['status'] = 'finished'
        job['elapsed_min'] = round((time.time() - t0) / 60, 1)


# ──────────────────────────── BI: дашборды и AI-страницы ─────────────────────
# Кастомные HTML-дашборды (План/Факт и другие) руками агента: посмотреть
# датасеты → сгенерировать или положить свой HTML → опубликовать по адресу
# /r/<slug>. Слаги и PUT-ручка появляются на стенде патчем
# deploy_gate/patches/260809_bi_slug_public_links.py; без него инструменты
# печатают отказ сервера как есть.


def _bi_dataset_key(d):
    """Ключ датасета в формате dataset_keys у /api/bi/ai/generate."""
    return '%s||%s' % (d.get('session_id'), d.get('data_source'))


def _bi_html_title(html):
    import re as _re
    m = _re.search(r'<title[^>]*>(.*?)</title>', html, _re.S | _re.I)
    return (m.group(1).strip()[:120] if m else '(без <title>)')


def bi_publish(dashboard_id, slug=None, mode=None, unpublish=False, chrome=None):
    if unpublish:
        code, js = MF.call('DELETE', '/api/bi/dashboards/%s/publish' % dashboard_id)
        if isinstance(js, dict) and js.get('success'):
            return 'Публикация отозвана: публичная ссылка дашборда %s больше не работает.\nСтенд: %s' % (dashboard_id, BASE_URL)
        return 'ОТКАЗ СЕРВЕРА (HTTP %s): %s' % (code, (js or {}).get('message') or str(js)[:300])
    body = {'mode': 'snapshot' if mode == 'snapshot' else 'live'}
    if chrome is not None:
        # Вид публичной обёртки AI-страницы: 'none' — без шапки MARFOR,
        # 'default' — вернуть шапку. Валидирует сервер (патч 260810).
        body['chrome'] = str(chrome)
    if slug:
        # Слаг = осознанный отказ от секретности адреса; подтверждение
        # public_by_name сервер требует явно — передаём его вместе со слагом.
        body['slug'] = str(slug)
        body['public_by_name'] = True
    code, js = MF.call('POST', '/api/bi/dashboards/%s/publish' % dashboard_id, data=body)
    if not isinstance(js, dict) or not js.get('success'):
        return 'ОТКАЗ СЕРВЕРА (HTTP %s): %s' % (code, (js or {}).get('message') or str(js)[:300])
    link = js.get('public_link') or {}
    url = link.get('url') or ''
    out = ['ОПУБЛИКОВАНО. %s' % url,
           'Стенд: %s   режим: %s   дашборд: %s' % (BASE_URL, link.get('mode'), dashboard_id)]
    if chrome == 'none':
        out.append('Обёртка отключена (chrome=none): /r/<адрес> отдаёт страницу '
                   'без шапки MARFOR, как /raw. Требует патча 260810 на стенде: '
                   'БЕЗ патча сервер молча игнорирует ключ и шапка останется — '
                   'проверьте страницу по ссылке выше.')
    if slug and str(link.get('token')) != str(slug).strip().lower():
        out.append('🔴 ВНИМАНИЕ: запрошен slug «%s», но сервер вернул токен «%s» — '
                   'стенд БЕЗ патча слагов молча публикует со случайным токеном. '
                   'Ссылка выше рабочая, но не именная.' % (slug, link.get('token')))
    elif slug:
        out.append('⚠️ Адрес именной, то есть УГАДЫВАЕМЫЙ: страница доступна всем, '
                   'кто знает или подберёт имя.')
    return '\n'.join(out)


# ─────────────────────────────── описание тулов ───────────────────────────────

TOOLS = [
    {
        'name': 'marfor_guide',
        'description': 'НАЧНИТЕ С ЭТОГО ИНСТРУМЕНТА: marfor_guide возвращает путеводитель по '
                       'MARFOR — что это за сервис, в каком порядке вызывать инструменты '
                       '(whoami → projects → datasets → describe → query), правила записи '
                       '(у каких инструментов есть сухой прогон, а какие пишут сразу; явное «да» '
                       'пользователя; обязательный период), форматы '
                       'фильтров, что делать при пустом аккаунте и как объяснять отказы. '
                       'Вызовите один раз в начале беседы, до остальных инструментов MARFOR. '
                       'Только чтение, в сеть не ходит.',
        'inputSchema': {'type': 'object', 'properties': {}},
    },
    {
        'name': 'marfor_whoami',
        'description': 'Проверить вход в MARFOR: адрес стенда, email, user_id, тариф, число '
                       'проектов, версия сервера и включённые наборы инструментов. Первый шаг '
                       'после marfor_guide и первое, что проверять при «нет доступа»: тот ли '
                       'это аккаунт.',
        'inputSchema': {'type': 'object', 'properties': {}},
    },
    {
        'name': 'marfor_projects',
        'description': 'Проекты пользователя — свои и те, к которым ему дали доступ: id, '
                       'название, уровень доступа. project_id отсюда нужен marfor_datasets. '
                       'Пустой список — аккаунт новый: ответ подскажет, что делать. Только чтение.',
        'inputSchema': {'type': 'object', 'properties': {}},
    },
    {
        'name': 'marfor_datasets',
        'description': 'Датасеты проекта — каталог всего, что можно читать и править: '
                       'источники, прогнозы, сценарии. По каждому печатает session_id и '
                       'data_source (их ждут marfor_describe и marfor_query) и ключ '
                       'session_id||data_source (его ждёт dataset_keys у marfor_bi_generate). '
                       'Только чтение.',
        'inputSchema': {
            'type': 'object',
            'properties': {
                'project_id': {'type': 'string', 'description': 'Из marfor_projects'},
                'max_chars': {'type': 'integer',
                              'description': 'Задан — дописать полный JSON ответа, обрезав до '
                                             'этой длины'},
            },
            'required': ['project_id'],
        },
    },
    {
        'name': 'marfor_describe',
        'description': 'Описание датасета: точные имена метрик и измерений (ровно их ждёт '
                       'marfor_query), число строк, диапазон дат, доступные годы и месяцы, '
                       'вычисляемые метрики с формулами. Вызывайте перед первым marfor_query и '
                       'перед любой правкой: имена колонок у каждого проекта свои и не '
                       'угадываются. Только чтение.',
        'inputSchema': {
            'type': 'object',
            'properties': {
                'session_id': {'type': 'string', 'description': 'Из marfor_datasets'},
                'data_source': {'type': 'string',
                                'description': "Из marfor_datasets: processed — источник, "
                                               "forecast — прогноз, scenario — сценарий. "
                                               "По умолчанию 'scenario'"},
                'save_to': {'type': 'string', 'description': 'Куда сохранить полный JSON'},
                'max_values': {'type': 'integer',
                               'description': 'Сколько значений каждого измерения показать, '
                                              'по умолчанию 12'},
            },
            'required': ['session_id'],
        },
    },
    {
        'name': 'marfor_api_get',
        'description': 'GET-запрос к любому /api/... эндпоинту MARFOR под своей сессией. '
                       'Только чтение. Большой ответ можно сохранить в файл через save_to.',
        'inputSchema': {
            'type': 'object',
            'properties': {
                'path': {'type': 'string', 'description': 'Путь, начинается с /api/'},
                'params': {'type': 'object', 'description': 'Query-параметры'},
                'save_to': {'type': 'string', 'description': 'Куда сохранить полный JSON'},
                'max_chars': {'type': 'integer', 'description': 'Обрезка ответа, по умолчанию 8000'},
            },
            'required': ['path'],
        },
    },
    {
        'name': 'marfor_query',
        'description': 'Агрегированные данные сценария: метрики по срезам за период '
                       '(тот же роут, что и сводная таблица). Возвращает плоские строки.',
        'inputSchema': {
            'type': 'object',
            'properties': {
                'session_id': {'type': 'string'},
                'metrics': {'type': 'array', 'items': {'type': 'string'},
                            'description': 'Имена КОЛОНОК-метрик из marfor_describe (не подписи)'},
                'dimensions': {'type': 'array', 'items': {'type': 'string'},
                               'description': 'Имена измерений из marfor_describe; время '
                                              '(year, month, quarter) добавляется само'},
                'filters': {'type': 'object',
                            'description': 'Например {"region_to": ["russia"], "year": [2026]} '
                                           'или {"region_to": {"values": ["russia"], '
                                           '"mode": "exclude"}}. У пишущих инструментов '
                                           'формат фильтров ДРУГОЙ — см. marfor_guide'},
                'date_from': {'type': 'string', 'description': 'ГГГГ-ММ-ДД'},
                'date_to': {'type': 'string', 'description': 'ГГГГ-ММ-ДД'},
                'data_source': {'type': 'string', 'description': "по умолчанию 'scenario'"},
                'save_to': {'type': 'string'},
                'max_chars': {'type': 'integer'},
            },
            'required': ['session_id', 'metrics'],
        },
    },
    {
        'name': 'marfor_apply_edit',
        'description': 'ЗАПИСЬ. Одна правка моделирования: задать метрике новое значение '
                       'в срезе (фильтры) — как правка ячейки в UI, с пересчётом цепочки. '
                       'base_value не передавать: сервер посчитает базу сам. '
                       'Сухого прогона НЕТ, вызов пишет сразу: сначала покажите пользователю, '
                       'что именно изменится (метрика, срез, период, «было → станет»), и '
                       'получите явное «да». Нужна сверка цифрами — marfor_apply_batch с одной '
                       'правкой и dry_run=true.',
        'inputSchema': {
            'type': 'object',
            'properties': {
                'session_id': {'type': 'string'},
                'metric': {'type': 'string', 'description': 'Отображаемое имя метрики'},
                'filters': {'type': 'object', 'description': 'Например {"region_to":"russia","channel_new":"vk_ads","year":["2026"],"month":["7"]}'},
                'new_value': {'type': 'number'},
                'wait': {'type': 'boolean', 'description': 'Ждать завершения, по умолчанию да'},
            },
            'required': ['session_id', 'metric', 'filters', 'new_value'],
        },
    },
    {
        'name': 'marfor_apply_batch',
        'description': 'ЗАПИСЬ. Пачка правок подряд. dry_run=true (по умолчанию) ничего не '
                       'меняет: считает текущую базу по каждому срезу и возвращает сверку '
                       '«база → цель → коэффициент». dry_run=false запускает фоновую задачу '
                       'и сразу возвращает job_id — следить через marfor_job_status.',
        'inputSchema': {
            'type': 'object',
            'properties': {
                'session_id': {'type': 'string'},
                'metric': {'type': 'string'},
                'edits': {'type': 'array', 'description': 'Список {label, filters, new_value}',
                          'items': {'type': 'object'}},
                'edits_file': {'type': 'string', 'description': 'Путь к JSON-файлу со списком правок'},
                'dry_run': {'type': 'boolean'},
                'stop_on_error': {'type': 'boolean'},
            },
            'required': ['session_id', 'metric'],
        },
    },
    {
        'name': 'marfor_apply_values',
        'description': 'ЗАПИСЬ, МАССОВО. Отдать MARFOR значения метрики по многим срезам '
                       'одним запросом. Коннектор разворачивает их в payload’ы РОВНО того '
                       'формата, который строит страница в режиме «Массовые изменения» '
                       '(включая mmm_slice_key и pivot_dimensions), а приложение исполняет '
                       'каждый той же задачей, что и ручной ввод: мутация на срез, числа '
                       'совпадают с интерфейсом до последнего знака. Экономия — на цепочке '
                       'метрик и бэндах: они считаются по разу на весь прогон, а не на каждый '
                       'срез. Порядок величины — десятки секунд на срез, не «минуты на всё». '
                       'dry_run=true (по умолчанию) ничего не меняет и показывает, что уйдёт. '
                       'Запись возвращает task_id — это ЕЩЁ НЕ результат, дождитесь '
                       'task_status=completed и прочитайте, сколько срезов применено. '
                       'Требует ручки /api/apply_modeling_bulk на стенде.',
        'inputSchema': {
            'type': 'object',
            'properties': {
                'session_id': {'type': 'string'},
                'metric': {'type': 'string',
                           'description': 'Отображаемое имя метрики, можно вычисляемую'},
                'scope': {'type': 'object',
                          'description': 'Границы мутации: период и общие фильтры, например '
                                         '{"year":[2026],"quarter":["Q3"],"region_to":["russia"]}'},
                'values': {'type': 'array', 'items': {'type': 'object'},
                           'description': 'Список {"slice": {"channel_new":"vk_ads","month":7}, '
                                          '"value": 5100000}. slice — любые колонки-измерения '
                                          'внутри scope; value — сколько должно СТАТЬ в срезе'},
                'values_file': {'type': 'string', 'description': 'Путь к JSON с тем же списком'},
                'pivot_dimensions': {'type': 'array', 'items': {'type': 'string'},
                                     'description': 'Измерения сводной; из них строится '
                                                    'mmm_slice_key. По умолчанию '
                                                    'region_to|channel_new|segment_new|category'},
                'dry_run': {'type': 'boolean'},
                'max_chars': {'type': 'integer'},
            },
            'required': ['session_id', 'metric', 'scope'],
        },
    },
    {
        'name': 'marfor_job_status',
        'description': 'Ход записи. job_id — фоновый батч marfor_apply_batch: сколько применено, '
                       'последняя правка, ошибки. task_id — задача MARFOR, которую вернули '
                       'marfor_apply_values или marfor_apply_edit(wait=false): ждите '
                       'task_status=completed и читайте, сколько срезов применено.',
        'inputSchema': {
            'type': 'object',
            'properties': {
                'job_id': {'type': 'string', 'description': 'Из ответа marfor_apply_batch'},
                'task_id': {'type': 'string',
                            'description': 'Из ответа marfor_apply_values / marfor_apply_edit'},
                'tail': {'type': 'integer', 'description': 'Сколько последних записей лога показать'},
            },
        },
    },
    {
        'name': 'marfor_job_stop',
        'description': 'Остановить фоновый батч после текущей правки.',
        'inputSchema': {'type': 'object', 'properties': {'job_id': {'type': 'string'}},
                        'required': ['job_id']},
    },
    {
        'name': 'marfor_bi_datasets',
        'description': 'Прежнее имя marfor_datasets, оставлено для совместимости и делает '
                       'то же самое. Датасеты проекта для BI-модуля: источники, прогнозы, '
                       'сценарии. Возвращает ключи вида session_id||data_source — '
                       'ровно их ждёт dataset_keys у marfor_bi_generate.',
        'inputSchema': {
            'type': 'object',
            'properties': {
                'project_id': {'type': 'string'},
                'max_chars': {'type': 'integer'},
            },
            'required': ['project_id'],
        },
    },
    {
        'name': 'marfor_bi_dashboards',
        'description': 'Список BI-дашбордов (свои + расшаренные): id, название, '
                       'тип (canvas — виджетный, ai_page — HTML-страница), роль, '
                       'опубликован ли. project_id сужает до одного проекта.',
        'inputSchema': {
            'type': 'object',
            'properties': {
                'project_id': {'type': 'string'},
                'max_chars': {'type': 'integer'},
            },
        },
    },
    {
        'name': 'marfor_bi_generate',
        'description': 'ЗАПИСЬ. Сгенерировать AI-страницу (интерактивный HTML-'
                       'дашборд, отчёт или презентацию) из датасетов проекта '
                       'встроенным генератором MARFOR. mode=create (по умолчанию) '
                       'создаёт НОВЫЙ дашборд: нужны project_id и dataset_keys '
                       '(из marfor_bi_datasets). mode=revise дорабатывает '
                       'существующий (нужен dashboard_id) и ПЕРЕЗАПИСЫВАЕТ его '
                       'страницу. skill: dashboard | presentation | plan_fact | '
                       'forecast_report | scenario_report | custom. Запрос '
                       'СИНХРОННЫЙ и долгий — минуты, на большие страницы до '
                       '~20 минут; таймаут у инструмента свой (MARFOR_BI_'
                       'GENERATE_TIMEOUT, по умолчанию 1500с), не обрывайте '
                       'вызов раньше времени. Сухого прогона нет: до вызова '
                       'скажите пользователю, какой дашборд будет создан или '
                       'перезаписан, и получите явное «да».',
        'inputSchema': {
            'type': 'object',
            'properties': {
                'project_id': {'type': 'string'},
                'prompt': {'type': 'string',
                           'description': 'Что построить (до 4000 символов)'},
                'skill': {'type': 'string'},
                'dataset_keys': {'type': 'array', 'items': {'type': 'string'},
                                 'description': 'Ключи session_id||data_source'},
                'dashboard_id': {'type': 'string',
                                 'description': 'Для mode=revise'},
                'mode': {'type': 'string', 'description': 'create | revise'},
                'timeout_s': {'type': 'integer'},
            },
            'required': ['prompt'],
        },
    },
    {
        'name': 'marfor_bi_get_page',
        'description': 'Скачать HTML AI-страницы дашборда (внутренняя ручка '
                       '/api/bi/ai_pages/<id>/raw). Страница большая, поэтому '
                       'сохраняется в файл (save_to, по умолчанию '
                       '~/.marfor_mcp/pages/<id>.html), а в диалог — сводка: '
                       'размер, заголовок.',
        'inputSchema': {
            'type': 'object',
            'properties': {
                'dashboard_id': {'type': 'string'},
                'save_to': {'type': 'string'},
            },
            'required': ['dashboard_id'],
        },
    },
    {
        'name': 'marfor_bi_put_page',
        'description': 'ЗАПИСЬ. Положить СВОЙ HTML в AI-страницу дашборда — для '
                       'кастомных дашбордов, собранных вне встроенного '
                       'генератора. html_file — путь к готовому файлу '
                       '(предпочтительно; html строкой — для мелочей). Нужен '
                       'цельный документ <!DOCTYPE html>…</html>, кап 5 МБ. '
                       'Дашборд — AI-страница или пустой (конвертируется); '
                       'виджетный не перезапишет. Публичная ссылка mode=live '
                       'обновится сразу, snapshot — при перепубликации. Требует '
                       'ручки PUT /api/bi/ai_pages/<id> на стенде (патч '
                       '260809_bi_slug_public_links); без неё — отказ сервера. '
                       'Сухого прогона нет: до вызова скажите пользователю, '
                       'страница какого дашборда будет перезаписана, и получите '
                       'явное «да».',
        'inputSchema': {
            'type': 'object',
            'properties': {
                'dashboard_id': {'type': 'string'},
                'html_file': {'type': 'string'},
                'html': {'type': 'string'},
            },
            'required': ['dashboard_id'],
        },
    },
    {
        'name': 'marfor_bi_publish',
        'description': 'ЗАПИСЬ. Опубликовать дашборд публичной ссылкой (или '
                       'отозвать её: unpublish=true). Без slug адрес — секретный '
                       'случайный токен /r/<43 символа>, его не угадать. slug '
                       'даёт человекочитаемый адрес /r/plan_fact, НО адрес '
                       'становится УГАДЫВАЕМЫМ — страница доступна ПО ИМЕНИ '
                       'любому; указывайте slug только по явной просьбе '
                       'пользователя (инструмент сам передаст подтверждение '
                       'public_by_name=true). Смена слага при перепубликации '
                       'освобождает старый адрес. mode: live (публичная '
                       'страница читает текущее состояние) | snapshot '
                       '(заморозка на момент публикации). Возвращает итоговый '
                       'URL. Слаги требуют патча 260809 на стенде; без него '
                       'инструмент явно скажет, что опубликовано со случайным '
                       'токеном. chrome="none" отдаёт /r/<адрес> без шапки '
                       'MARFOR (только AI-страницы, патч 260810 на стенде); '
                       'без параметра перепубликация сохраняет прежний вид. '
                       'Сухого прогона нет: до вызова назовите пользователю '
                       'дашборд и будущий адрес и получите явное «да».',
        'inputSchema': {
            'type': 'object',
            'properties': {
                'dashboard_id': {'type': 'string'},
                'slug': {'type': 'string',
                         'description': '3-64 символа: a-z, 0-9, «-», «_»'},
                'mode': {'type': 'string', 'description': 'live | snapshot'},
                'chrome': {'type': 'string', 'enum': ['none', 'default'],
                           'description': "Обёртка публичной AI-страницы: "
                                          "'none' — без шапки MARFOR (как /raw), "
                                          "'default' — с шапкой. Не указан — "
                                          "прежний выбор сохраняется."},
                'unpublish': {'type': 'boolean'},
            },
            'required': ['dashboard_id'],
        },
    },
    {
        'name': 'marfor_mmm_params',
        'description': 'Калибровка MMM: справочник ручек модели — что можно крутить, в каких '
                       'границах, что стоит за умолчанием. С этого начинают настройку: '
                       'дальше marfor_mmm_calibrate гоняет варианты, marfor_mmm_compare их '
                       'сравнивает. Только чтение.',
        'inputSchema': {
            'type': 'object',
            'properties': {
                'save_to': {'type': 'string'},
                'max_chars': {'type': 'integer'},
            },
        },
    },
    {
        'name': 'marfor_mmm_calibrate',
        'description': 'Калибровка MMM: запустить прогон В ПЕСОЧНИЦЕ с заданными ручками. '
                       'НИЧЕГО не меняет — ни модель, ни проект, ни данные: результат ложится '
                       'отдельным прогоном под своим run_id. Возвращает run_id сразу, счёт идёт '
                       'фоном (минуты–часы). Больше одного идущего прогона на сессию нельзя. '
                       'Ход прогона — marfor_mmm_runs, готовый результат — marfor_mmm_report. '
                       'Имена и границы ручек — marfor_mmm_params. Для быстрой прикидки '
                       'draws=150, tune=100. holdout_periods>0 даёт ПРОВЕРОЧНЫЙ прогон: доли '
                       'каналов из него брать нельзя, он мерит точность.',
        'inputSchema': {
            'type': 'object',
            'properties': {
                'session_id': {'type': 'string'},
                'params': {'type': 'object',
                           'description': 'Ручки модели: prior_strength, baseline_prior, '
                                          'season_per_market, n_harmonics, season_sigma, '
                                          'level_mode, level_rw_sigma, trend_sigma, '
                                          'min_spend_share, holdout_periods, draws, tune, '
                                          'target_accept, market_dims, channel_dims, drivers, '
                                          'dependent. Неуказанное — по умолчанию.'},
                'label': {'type': 'string',
                          'description': 'Пометка «зачем этот прогон» — её видно в списке и в сравнении'},
            },
            'required': ['session_id'],
        },
    },
    {
        'name': 'marfor_mmm_runs',
        'description': 'Калибровка MMM: список прогонов сессии — что считается сейчас, что '
                       'посчитано, с какими ручками и чем кончилось. Им же проверяют ход '
                       'запущенного прогона. Только чтение.',
        'inputSchema': {
            'type': 'object',
            'properties': {
                'session_id': {'type': 'string'},
                'limit': {'type': 'integer', 'description': 'по умолчанию 30'},
                'save_to': {'type': 'string'},
                'max_chars': {'type': 'integer'},
            },
            'required': ['session_id'],
        },
    },
    {
        'name': 'marfor_mmm_report',
        'description': 'Калибровка MMM: полный отчёт по прогону — доли каналов с интервалами, '
                       'база каждого рынка, R²/MAPE, сходимость (rhat, ess, дивергенции), '
                       'формальная приёмка и метрики проверочного прогона. Помесячные ряды сюда '
                       'не попадают, они в marfor_mmm_diagnose. Только чтение.',
        'inputSchema': {
            'type': 'object',
            'properties': {
                'run_id': {'type': 'string'},
                'market': {'type': 'string', 'description': 'только один рынок (иначе все)'},
                'save_to': {'type': 'string'},
                'max_chars': {'type': 'integer'},
            },
            'required': ['run_id'],
        },
    },
    {
        'name': 'marfor_mmm_compare',
        'description': 'Калибровка MMM: сравнить прогоны бок о бок — доли, качество, сходимость, '
                       'приёмка. Отдельно показывает разброс доли платных между вариантами: это '
                       'мера того, насколько вывод держится на допущениях, а не на данных. '
                       'Только чтение.',
        'inputSchema': {
            'type': 'object',
            'properties': {
                'run_ids': {'type': 'array', 'items': {'type': 'string'},
                            'description': 'минимум два run_id'},
                'save_to': {'type': 'string'},
                'max_chars': {'type': 'integer'},
            },
            'required': ['run_ids'],
        },
    },
    {
        'name': 'marfor_mmm_diagnose',
        'description': 'Калибровка MMM: почему прогон получился таким. Раскладывает помесячную '
                       'динамику KPI на сезон, уровень и медиа; показывает, у каких каналов '
                       'интервал эластичности накрывает ноль или вклад почти не меняется '
                       '(доля держится на приоре, а не на данных), и какие каналы спутаны с '
                       'сезоном. Только чтение.',
        'inputSchema': {
            'type': 'object',
            'properties': {
                'run_id': {'type': 'string'},
                'market': {'type': 'string', 'description': 'по умолчанию самый крупный по KPI'},
                'save_to': {'type': 'string'},
                'max_chars': {'type': 'integer'},
            },
            'required': ['run_id'],
        },
    },
    {
        'name': 'marfor_mmm_apply',
        'description': 'ЗАПИСЬ. Применить калибровочный прогон к MMM-модели: эластичности каналов '
                       'из прогона становятся эластичностями модели. Это единственный инструмент '
                       'калибровки, который что-то меняет. Без confirm=true — сухой прогон: '
                       'показывает отчёт и не пишет ничего. На боевом стенде ОТКАЗЫВАЕТ: правки '
                       'в бой идут через шлюз. Проверочные прогоны (holdout_periods>0) применять '
                       'нельзя — сервер откажет.',
        'inputSchema': {
            'type': 'object',
            'properties': {
                'run_id': {'type': 'string'},
                'confirm': {'type': 'boolean',
                            'description': 'true — записать в модель; без него только показ'},
                'model_id': {'type': 'string',
                             'description': 'к какой модели применить; по умолчанию последняя '
                                            'готовая MMM-модель этой сессии'},
            },
            'required': ['run_id'],
        },
    },
]


# ───────────────── путеводитель: instructions и marfor_guide ──────────────────
# Поле instructions из ответа initialize читает Claude Code и режет его на 2 КБ — лимит
# считается в БАЙТАХ, русская буква стоит двух, поэтому текст английский и короткий.
# Claude Desktop и claude.ai это поле до модели не доносят вовсе, поэтому всё существенное
# продублировано инструментом marfor_guide, а первая фраза описания каждого инструмента
# на него ссылается (см. visible_tools).

INSTRUCTIONS = (
    'Start by calling marfor_guide: it returns the full MARFOR playbook (in Russian). '
    'Read it once per conversation and follow it.\n\n'
    'MARFOR (https://marfor.pro) is a forecasting and scenario-planning service for marketers: '
    'metric forecasts, budget scenarios, channel elasticities and BI dashboards. It acts on '
    'behalf of the signed-in user and sees exactly what that user sees in the web app.\n\n'
    'Workflow: marfor_whoami -> marfor_projects -> marfor_datasets(project_id) -> '
    'marfor_describe(session_id, data_source) -> marfor_query. Never invent a session_id, '
    'project_id, metric or dimension name and never ask the user to look them up: take them '
    'from these tools. If the account has no projects or datasets, say that data cannot be '
    'uploaded through MCP and send the user to https://marfor.pro to create a project and '
    'upload data there.\n\n'
    # Сухой прогон обещан ПОИМЁННО: у apply_edit и BI-инструментов параметра dry_run нет, и
    # общее «always run with dry_run=true first» приводило к записи без «да» пользователя.
    'Writing (tools whose description contains the word "ЗАПИСЬ"): write only after an '
    'explicit yes from the user. marfor_apply_values and marfor_apply_batch have a dry run: '
    'call them with dry_run=true first and show the reconciliation. marfor_apply_edit, '
    'marfor_bi_generate, marfor_bi_put_page and marfor_bi_publish have NO dry run and write '
    'at once: first tell the user exactly what will change, then get the yes. Every edit must '
    'be bounded by a year or by a date_from/date_to pair; a month without a year changes that '
    'month in ALL years. Never pass new_value=0 (use 0.01). Filter formats differ between '
    'marfor_query and the apply tools - see marfor_guide. marfor_apply_batch runs in the '
    'background: poll marfor_job_status. Publish under a readable slug only when the user '
    'explicitly asks: it makes the page guessable.\n\n'
    'HTTP 402 / plan_limit means the user\'s plan does not allow the action: explain it, do not '
    'retry. If the token is rejected, the user must sign in again: the error text gives the '
    'exact command. Never ask the user to paste a token or a password into the chat.\n\n'
    'Answer in the user\'s language. Help: https://marfor.pro/mcp, support@marfor.pro'
)

GUIDE_TEXT = """\
MARFOR — ПУТЕВОДИТЕЛЬ ДЛЯ АССИСТЕНТА
Прочитайте один раз в начале беседы. С пользователем говорите на его языке и простыми
словами: он маркетолог, а не разработчик.

1. ЧТО ТАКОЕ MARFOR
MARFOR (https://marfor.pro) — сервис прогнозирования и моделирования сценариев для
маркетологов: прогнозы метрик, сценарии бюджетов («что будет с заказами и выручкой, если
перераспределить расходы по каналам»), эластичности каналов и BI-дашборды.
Данные лежат в проектах. В проекте есть датасеты трёх видов: источник (загруженные
данные), прогноз и сценарий (прогноз плюс правки моделирования).
Этот сервер работает от имени вошедшего пользователя и видит ровно то же, что он сам в
веб-интерфейсе: его проекты и проекты, к которым ему дали доступ.

2. ПОРЯДОК РАБОТЫ — всегда в этой последовательности
  1) marfor_whoami   — проверить вход: стенд, email, тариф, число проектов.
  2) marfor_projects — проекты. Если их несколько — спросите, с каким работать.
  3) marfor_datasets(project_id) — датасеты проекта; отсюда берутся session_id и
     data_source (processed — источник, forecast — прогноз, scenario — сценарий).
  4) marfor_describe(session_id, data_source) — точные имена метрик и измерений,
     доступные годы и месяцы, вычисляемые метрики с формулами.
  5) marfor_query — цифры: метрики по срезам за период.
Идентификаторы и имена НЕ выдумывайте и не просите пользователя искать их руками:
session_id, project_id, имена метрик и измерений берутся только из шагов 2–4.

3. ЧТЕНИЕ: marfor_query
  • metrics — имена КОЛОНОК (левый столбец marfor_describe), а не подписи.
  • dimensions — измерения разреза; время (year, month, quarter) добавляется само.
  • Период: date_from / date_to (ГГГГ-ММ-ДД) либо фильтры year, month, quarter.
  • У сценария рядом с метрикой есть колонка <имя>_modeling — значение после правок;
    колонка без суффикса — исходный прогноз. «Было → стало» = обе колонки в одном запросе.
  • Вычисляемая метрика (CRR, CAC, средний чек, конверсия) своей колонки не имеет:
    запросите компоненты её формулы и посчитайте ОТ СУММ (Σчислитель / Σзнаменатель).
    Складывать или усреднять готовые отношения по строкам нельзя — ответ будет неверным.
  • Слишком подробный разрез сервер отклонит ошибкой (кап ~25 000 комбинаций), а не
    усечёт молча: уберите измерение или сузьте период.
  • Большой ответ сохраняйте в файл (save_to), в диалог берите сводку.

4. ЗАПИСЬ — три правила без исключений
Пишущие инструменты помечены словом «ЗАПИСЬ» в описании. Любая запись — только после
явного «да» пользователя на то, что ему показано.
  1) С сухим прогоном (параметр dry_run): marfor_apply_values, marfor_apply_batch. Сначала
     dry_run=true (так стоит по умолчанию): покажите сверку — какие срезы, сколько строк,
     какие годы и регионы, «база → цель». Пишите (dry_run=false) только после явного «да»
     на эту сверку. У marfor_mmm_apply (набор mmm) сухой прогон — вызов без confirm=true.
  2) БЕЗ сухого прогона: marfor_apply_edit, marfor_bi_generate, marfor_bi_put_page,
     marfor_bi_publish. Параметра dry_run у них нет, вызов пишет СРАЗУ (вызов с dry_run=true
     сервер отклонит, ничего не записав). Сначала скажите пользователю, что именно
     изменится: метрика, срез, период, «было → станет» (текущее значение даёт marfor_query);
     для BI — какой дашборд будет создан, перезаписан или опубликован и по какому адресу.
     Вызывайте только после явного «да».
  3) У каждой правки обязан быть период: год (year) либо пара дат date_from + date_to.
     Месяц без года — это тот же месяц ВО ВСЕХ годах, включая прошлые; массовую правку
     без периода сервер отклонит.
Ещё:
  • new_value=0 не передавайте: чтобы обнулить срез, поставьте 0.01.
  • Значение — АБСОЛЮТНОЕ («сколько должно стать в срезе»), не прирост и не множитель.
  • Правки не накапливаются: каждая считается от базового прогноза, повтор с тем же
    значением ничего не меняет. Откат — «Сброс цепочки» на странице сценария.
  • Срез с нулевой базой умножением не поднять: массовая правка такие срезы пропускает.
  • Ответ «сервер занят» — подождите и повторите; ту же правку параллельно не запускайте.

5. КАКОЙ ИНСТРУМЕНТ ЗАПИСИ ВЫБРАТЬ
  • marfor_apply_values — основной: значения по многим срезам одним запросом. scope —
    общие границы (период и общие фильтры), values — список {"slice": {…}, "value": число}.
    Запись возвращает task_id — это ещё не результат: опрашивайте
    marfor_job_status(task_id=…) до task_status=completed и читайте, сколько применено.
  • marfor_apply_edit — одна правка одного среза, как правка ячейки в интерфейсе (~минута).
    Сухого прогона нет (правило 2); нужна сверка цифрами — marfor_apply_batch с одной
    правкой и dry_run=true.
  • marfor_apply_batch — цепочка одиночных правок ФОНОМ: сразу возвращает job_id и идёт
    от минут до часов. Ход — marfor_job_status(job_id=…), остановка — marfor_job_stop.
    Фоновая задача живёт в процессе этого сервера: перезапуск клиента её обрывает; её же
    останавливает отозванный токен (статус error) — после входа повторите оставшиеся правки.

6. ФИЛЬТРЫ: ДВА РАЗНЫХ ФОРМАТА — не путайте
  • marfor_query принимает любую форму и сам приводит к нужной:
      {"region": "msk"}   {"region": ["msk", "spb"]}
      {"region": {"values": ["msk"], "mode": "exclude"}}
  • marfor_apply_edit и правки marfor_apply_batch — ТОЛЬКО плоские значения и списки,
    время — списками строк; словарь {"values": …} здесь не работает:
      {"region": "msk", "channel": "vk_ads", "year": ["2026"], "month": ["7"]}
  • marfor_apply_values: scope — списки ({"year": [2026], "quarter": ["Q3"]}),
    slice — плоские значения ({"channel": "vk_ads", "month": 7}).
Имена колонок в примерах условные: у каждого проекта свои, берите их из marfor_describe.
pivot_dimensions у marfor_apply_values — измерения сводной таблицы сценария; умолчание
подходит не всякому проекту, сверяйтесь с измерениями из marfor_describe.

7. BI-ДАШБОРДЫ
marfor_bi_dashboards — список; marfor_bi_generate — собрать страницу встроенным генератором
(вызов синхронный и ДОЛГИЙ: минуты, до ~20 — не обрывайте); marfor_bi_get_page и
marfor_bi_put_page — скачать HTML или положить свой; marfor_bi_publish — публичная ссылка.
Без slug адрес секретный. Со slug он УГАДЫВАЕМЫЙ — страницу увидит любой, кто знает имя:
указывайте slug только по прямой просьбе пользователя.

8. ЕСЛИ АККАУНТ ПУСТОЙ (нет проектов или датасетов)
Через MCP нельзя ни создать проект, ни загрузить данные. Скажите это прямо и ведите
пользователя в веб-интерфейс: открыть https://marfor.pro, создать проект, загрузить данные,
построить прогноз и сценарий — после этого они появятся в marfor_datasets.
Демо-данные и цифры «для примера» не придумывайте.

9. ОТКАЗЫ И ОШИБКИ — как объяснять
  • HTTP 402, "error": "plan_limit" — ограничение тарифа: функция недоступна на тарифе или
    исчерпана квота. Перескажите message, назовите нужный тариф (required_plan), дайте
    ссылку https://marfor.pro/account. Вызов не повторяйте и обход не ищите.
  • «Токен недействителен» (HTTP 401) — токен отозван или истёк. Нужен новый вход:
    python3 server.py --login (точная команда есть в тексте ошибки). Токен и пароль в чат
    не просите НИКОГДА; если токен попал в чат — посоветуйте отозвать его на
    https://marfor.pro/account, раздел «Подключение Claude (MCP)».
  • HTTP 403 token_scope — токену это действие не разрешено (оплата, доступы и
    приглашения, удаление проекта, управление токенами): оно делается в веб-интерфейсе.
  • «Проект не найден или нет доступа» — проверьте marfor_whoami (тот ли аккаунт) и
    marfor_projects; доступ к чужому проекту выдаёт его владелец.
  • «Инструмент выключен» — ответ сам называет переменную (MARFOR_TOOLSETS или
    MARFOR_READONLY). Наборы mmm (калибровка MMM) и raw (произвольный GET по API) на
    marfor.pro по умолчанию выключены.
  • Нет связи, таймаут, ошибка сертификата — в тексте ошибки есть подсказка, перескажите её.

10. ПОМОЩЬ
Подключение и частые вопросы: https://marfor.pro/mcp · поддержка: support@marfor.pro
"""

EMPTY_ACCOUNT_TEXT = (
    'В этом аккаунте пока нет проектов.\n'
    'Через MCP нельзя ни создать проект, ни загрузить данные — это делается в веб-интерфейсе:\n'
    '  1. откройте %s и войдите под этим же аккаунтом;\n'
    '  2. создайте проект и загрузите в него данные;\n'
    '  3. постройте прогноз и сценарий.\n'
    'После этого проект появится в marfor_projects, а его данные — в marfor_datasets.'
    % PUBLIC_BASE)


# ─────────────────────────── наборы инструментов ─────────────────────────────
# Токен умеет ровно то, что умеет этот сервер, поэтому на публичном стенде по умолчанию
# видны только наборы, ручки которых там есть и безопасны: core, write, bi. Набор mmm
# требует ручек калибровки (их на marfor.pro нет), raw — произвольный GET по /api/.
# На ЛЮБОМ другом стенде умолчание — всё: прежние настройки и тесты не меняются.

TOOLSET_NAMES = ('core', 'write', 'bi', 'mmm', 'raw')
PUBLIC_TOOLSETS = ('core', 'write', 'bi')
_TOOLSET_OF = {
    'marfor_api_get': 'raw',
    'marfor_apply_edit': 'write', 'marfor_apply_batch': 'write',
    'marfor_apply_values': 'write',
    'marfor_job_status': 'write', 'marfor_job_stop': 'write',
}
# Сами ничего не пишут, но без пишущих инструментов бессмысленны — прячутся вместе с ними.
_WRITE_FLOW = ('marfor_job_status', 'marfor_job_stop')

GUIDE_HINT = 'Правила работы с MARFOR — в marfor_guide: вызовите его один раз в начале беседы. '
_PARAM_HINTS = {
    'session_id': 'Идентификатор датасета — из marfor_datasets, не выдумывать',
    'project_id': 'Идентификатор проекта — из marfor_projects',
    'dashboard_id': 'Идентификатор дашборда — из marfor_bi_dashboards',
}


def _toolset_of(name):
    if name in _TOOLSET_OF:
        return _TOOLSET_OF[name]
    if name.startswith('marfor_mmm_'):
        return 'mmm'
    if name.startswith('marfor_bi_'):
        return 'bi'
    return 'core'


def _is_write_tool(tool):
    """Пишущий инструмент узнаётся по слову «ЗАПИСЬ» в начале описания: так помечены
    все нынешние, и тот же признак подхватит будущие, кто бы их ни добавил."""
    return (tool.get('description') or '').lstrip().startswith('ЗАПИСЬ')


def _readonly():
    return (os.environ.get('MARFOR_READONLY') or '').strip().lower() in ('1', 'true', 'yes', 'on')


def _active_toolsets():
    raw = (os.environ.get('MARFOR_TOOLSETS') or '').strip().lower()
    if not raw:
        return set(PUBLIC_TOOLSETS) if _is_public_stand() else set(TOOLSET_NAMES)
    names = {x.strip() for x in raw.replace(';', ',').split(',') if x.strip()}
    if 'all' in names:
        return set(TOOLSET_NAMES)
    # core включён всегда: в нём marfor_guide, на который ссылается каждое описание
    return (names & set(TOOLSET_NAMES)) | {'core'}


def _toolsets_label():
    active = _active_toolsets()
    return ', '.join(x for x in TOOLSET_NAMES if x in active)


def _hidden_reason(name):
    """None — инструмент доступен (или такого нет вовсе); иначе текст, как его включить."""
    tool = next((t for t in TOOLS if t['name'] == name), None)
    if tool is None:
        return None
    ts, active = _toolset_of(name), _active_toolsets()
    if ts not in active:
        want = ','.join(x for x in TOOLSET_NAMES if x in active or x == ts)
        return ('Инструмент %s выключен: его набор «%s» не входит во включённые (%s). Чтобы '
                'включить, задайте серверу переменную окружения MARFOR_TOOLSETS=%s (или all) в '
                'конфигурации MCP-клиента и перезапустите клиент. Готовую команду регистрации '
                'с этой переменной печатает --print-config, если задать переменную при его '
                'запуске.' % (name, ts, _toolsets_label(), want))
    if _readonly() and (_is_write_tool(tool) or name in _WRITE_FLOW):
        return ('Инструмент %s меняет данные, а сервер запущен только для чтения '
                '(MARFOR_READONLY=1). Чтобы разрешить запись, уберите эту переменную из '
                'конфигурации MCP-клиента и перезапустите клиент.' % name)
    return None


def visible_tools():
    """Список для tools/list: только включённые наборы, без пишущих при MARFOR_READONLY.
    Первая фраза каждого описания отсылает к marfor_guide, а идентификаторы получают
    подсказку, откуда их брать, — исходный TOOLS при этом не меняется."""
    out = []
    for tool in TOOLS:
        if _hidden_reason(tool['name']):
            continue
        t = copy.deepcopy(tool)
        if t['name'] != 'marfor_guide':
            t['description'] = GUIDE_HINT + t['description']
        props = (t.get('inputSchema') or {}).get('properties') or {}
        for key, hint in _PARAM_HINTS.items():
            if isinstance(props.get(key), dict) and not props[key].get('description'):
                props[key]['description'] = hint
        out.append(t)
    return out


def _refusal(code, js, limit=300):
    """Отказ сервера одной строкой. Тарифный отказ разворачиваем отдельно: его надо
    объяснить пользователю, а не повторять вызов."""
    msg = (js.get('message') or js.get('error')) if isinstance(js, dict) else None
    out = 'ОТКАЗ СЕРВЕРА (HTTP %s): %s' % (code, msg or str(js)[:limit])
    if code == 402 or (isinstance(js, dict) and js.get('error') == 'plan_limit'):
        out += ('\nЭто ограничение тарифа%s. Объясните это пользователю и не повторяйте вызов; '
                'тарифы — %s/account.'
                % ((' (нужен тариф: %s)' % js.get('required_plan'))
                   if isinstance(js, dict) and js.get('required_plan') else '', PUBLIC_BASE))
    return out


def _ver_tuple(v):
    return tuple(int(x) for x in re.findall(r'\d+', str(v or ''))[:3])


def _published_version():
    """{version, sha256, size} публичной копии сервера. Только для публичного стенда и без
    токена: ручка открытая, а на токен вне /api/ стенд отвечает отказом. Любая ошибка —
    молча: проверка версии не должна мешать работе."""
    if not _is_public_stand():
        return None
    try:
        code, _, raw = MF._raw('GET', '/mcp/version.json', timeout=5, auth=False)
        js = _json_or_none(raw) if code == 200 else None
        return js if isinstance(js, dict) and js.get('version') else None
    except Exception:
        return None


def _newer_version_note():
    pub = _published_version()
    if not pub or _ver_tuple(pub.get('version')) <= _ver_tuple(SERVER_VERSION):
        return ''
    return ('⚠️ Доступна новая версия сервера: %s (у вас %s). Обновление: скачайте '
            '%s/mcp/server.py поверх файла %s и перезапустите клиент.'
            % (pub.get('version'), SERVER_VERSION, PUBLIC_BASE, os.path.abspath(__file__)))


def _login_mode_label():
    if MF.token:
        return ('токен из переменной MARFOR_TOKEN' if MF.token_source == 'env'
                else 'токен из файла %s' % TOKEN_FILE)
    return 'пароль (MARFOR_EMAIL)'


def _projects_line():
    """«Проектов: N» для whoami. Ошибка списка — не повод ронять проверку входа."""
    try:
        code, js = MF.call('GET', '/api/bi/projects')
    except Exception:
        return ''
    n = _count_projects(js) if isinstance(js, dict) and js.get('success') else None
    if n is None:
        return ''
    if n == 0:
        return 'Проектов: 0 — аккаунт пустой. ' + EMPTY_ACCOUNT_TEXT.split('\n', 1)[1]
    names = ', '.join('«%s»' % p.get('name') for p in js['projects'][:5] if isinstance(p, dict))
    return 'Проектов: %d (%s%s)' % (n, names, ', …' if n > 5 else '')


def _datasets_text(args):
    code, js = MF.call('GET', '/api/bi/datasets/%s' % _q(args['project_id']))
    if not isinstance(js, dict) or not js.get('success'):
        return _refusal(code, js)
    ds = [d for d in (js.get('datasets') or []) if isinstance(d, dict)]
    if not ds:
        return ('В проекте %s пока нет датасетов: данные ещё не загружены. Загрузка данных, '
                'прогноз и сценарий делаются в веб-интерфейсе %s — через MCP загрузки нет.'
                % (args['project_id'], PUBLIC_BASE))
    kinds = (('source', 'ИСТОЧНИКИ (загруженные данные)'), ('forecast', 'ПРОГНОЗЫ'),
             ('scenario', 'СЦЕНАРИИ'))
    out = ['Датасетов: %d   проект: %s   стенд: %s' % (len(ds), args['project_id'], BASE_URL)]
    for kind, title in kinds + ((None, 'ПРОЧЕЕ'),):
        rows = [d for d in ds if (d.get('kind') == kind if kind
                                  else d.get('kind') not in ('source', 'forecast', 'scenario'))]
        if not rows:
            continue
        out += ['', title]
        for d in rows:
            marks = []
            if d.get('snapshot'):
                marks.append('снапшот')
            if d.get('available') is False:
                marks.append('таблица НЕ построена — откройте сценарий в веб-интерфейсе')
            upd = d.get('data_updated_at') or d.get('updated_at')
            if upd:
                marks.append('данные от %s' % str(upd)[:16])
            out.append('  %s%s' % (d.get('name'), (' · ' + ' · '.join(marks)) if marks else ''))
            out.append('      session_id=%s   data_source=%s'
                       % (d.get('session_id'), d.get('data_source')))
            out.append('      ключ для marfor_bi_generate: %s' % _bi_dataset_key(d))
    first = ds[0]
    out += ['', 'Для marfor_query передайте session_id и data_source выбранного датасета — '
                'оба значения, как они напечатаны выше.',
            'Перед первым запросом: marfor_describe(session_id="%s", data_source="%s") — точные '
            'имена метрик и измерений.' % (first.get('session_id'), first.get('data_source'))]
    text = '\n'.join(out)
    return _emit(js, args, prefix=text + '\n\nПолный JSON:\n') if args.get('max_chars') else text


def _describe_text(args):
    sid = args['session_id']
    src = args.get('data_source') or 'scenario'
    code, js = MF.call('GET', '/api/metadata/%s' % _q(sid), params={'data_source': src})
    if not isinstance(js, dict) or not js.get('success'):
        return _refusal(code, js)
    # Формулы — отдельной ручкой: в метаданных вычисляемые метрики есть не всегда.
    # Отказ здесь не мешает отдать остальное.
    derived = []
    try:
        _, dj = MF.call('GET', '/api/derived_metrics/%s' % _q(sid))
        if isinstance(dj, dict) and dj.get('success'):
            derived = [d for d in (dj.get('derived_metrics') or []) if isinstance(d, dict)]
    except Exception:
        pass
    if not derived:
        derived = [d for d in (js.get('derived_metrics') or []) if isinstance(d, dict)]
    metrics = [m for m in (js.get('metrics') or []) if isinstance(m, dict)]
    dims = [d for d in (js.get('dimensions') or []) if isinstance(d, dict)]
    values = js.get('dimensions_values') if isinstance(js.get('dimensions_values'), dict) else {}
    cap = int(args.get('max_values') or 12)
    rng = js.get('date_range') or {}
    out = ['Датасет: session_id=%s   data_source=%s   стенд: %s' % (sid, src, BASE_URL),
           'Строк: %s   даты: %s … %s' % (js.get('total_rows'), rng.get('min'), rng.get('max'))]
    for key, title in (('available_years', 'Годы'), ('available_quarters', 'Кварталы'),
                       ('available_months', 'Месяцы')):
        if js.get(key):
            out.append('%s: %s' % (title, ', '.join(str(x) for x in js[key])))

    def _named(row):
        name, disp = row.get('name'), row.get('display_name')
        return '  %s%s' % (name, (' — %s' % disp) if disp and disp != name else '')

    out += ['', 'МЕТРИКИ (%d) — в marfor_query передаётся имя слева:' % len(metrics)]
    out += [_named(m) for m in metrics] or ['  (нет)']
    out += ['', 'ИЗМЕРЕНИЯ (%d):' % len(dims)]
    for d in dims:
        line = _named(d)
        vals = values.get(d.get('name'))
        if isinstance(vals, list) and vals:
            line += '   [%s%s]' % (', '.join(str(v) for v in vals[:cap]),
                                   ', … всего %d' % len(vals) if len(vals) > cap else '')
        out.append(line)
    if not dims:
        out.append('  (нет)')
    if dims and not values:
        out.append('  Значения измерения: marfor_query с dimensions=["<имя>"] и одной метрикой.')
    if derived:
        out += ['', 'ВЫЧИСЛЯЕМЫЕ МЕТРИКИ (%d) — формулы поверх колонок. В marfor_query своей '
                    'колонки у них нет: запрашивайте компоненты и считайте ОТ СУММ.'
                    % len(derived)]
        out += ['  %s = %s' % (d.get('name'), d.get('formula')) for d in derived]
    if src == 'scenario':
        out += ['', 'Это сценарий: рядом с метрикой есть колонка <имя>_modeling — значение '
                    'после правок; колонка без суффикса — исходный прогноз.']
    if args.get('save_to'):
        # mapping_config в файл не кладём: это служебная разметка колонок на сотни строк
        slim = {k: v for k, v in js.items() if k != 'mapping_config'}
        slim['derived_metrics'] = derived
        p = os.path.expanduser(args['save_to'])
        os.makedirs(os.path.dirname(p) or '.', exist_ok=True)
        with open(p, 'w', encoding='utf-8') as f:
            json.dump(slim, f, ensure_ascii=False, indent=1)
        out += ['', 'Полный ответ сохранён: %s' % p]
    return '\n'.join(out)


def _task_status_text(args):
    tid = args['task_id']
    code, js = MF.call('GET', '/api/modeling_task_status/%s' % _q(tid))
    if not isinstance(js, dict):
        return 'HTTP %s: неожиданный ответ сервера' % code
    st = js.get('task_status')
    out = ['Задача %s: %s   стенд: %s' % (tid, st or 'статус неизвестен', BASE_URL)]
    if st == 'completed' and 'applied' in js:
        out += ['  применено   : %s из %s' % (js.get('applied'), js.get('changes')),
                '  без эффекта : %d' % len(js.get('unchanged') or []),
                '  ошибок      : %d' % len(js.get('failed') or []),
                '  цепочка     : %s' % ('пересчитана' if js.get('chain_propagated')
                                        else 'НЕ пересчитана')]
    elif st not in ('completed', 'error', 'failed') and js.get('success') is not False:
        out.append('  Ещё идёт — повторите вызов чуть позже.')
    if js.get('message') or js.get('error'):
        out.append('  %s' % (js.get('error') or js.get('message')))
    if st in ('completed', 'error', 'failed') or js.get('success') is False:
        return _emit(js, args, prefix='\n'.join(out) + '\n\nПолный ответ:\n')
    return '\n'.join(out)


def _no_dry_run_refusal(name, args):
    """Пишущий инструмент БЕЗ параметра dry_run вызван с dry_run=true → отказ, в сеть не идём.
    Лишний аргумент схема пропускает, а ветка инструмента его не читает: вызов
    marfor_apply_edit(dry_run=true) сразу делал POST и менял сценарий без «да» пользователя."""
    if not args.get('dry_run'):
        return None
    tool = next((t for t in TOOLS if t['name'] == name), None)
    if tool is None or not _is_write_tool(tool):
        return None
    props = (tool.get('inputSchema') or {}).get('properties') or {}
    if 'dry_run' in props:
        return None
    how = ('Сухой прогон у него — вызов без confirm=true.' if 'confirm' in props else
           'Сверку «база → цель» без записи дают marfor_apply_batch и marfor_apply_values с '
           'dry_run=true. У этого инструмента правило другое: покажите пользователю словами, '
           'что именно изменится, получите явное «да» и повторите вызов без dry_run.')
    return ('ОТКАЗ: вызов НЕ выполнен, ничего не записано. У инструмента %s нет сухого прогона '
            '(параметра dry_run): он пишет сразу, а dry_run=true был бы молча пропущен. %s'
            % (name, how))


def call_tool(name, args):
    hidden = _hidden_reason(name)
    if hidden:
        return hidden
    refusal = _no_dry_run_refusal(name, args)
    if refusal:
        return refusal

    if name == 'marfor_guide':
        return GUIDE_TEXT

    if name == 'marfor_whoami':
        MF.logged_in = False
        MF.ensure()
        acc = MF.account
        out = [f"Вход выполнен ({_login_mode_label()}).",
               f"Стенд   : {acc['base_url']}",
               f"Email   : {acc['email']}",
               f"user_id : {acc['user_id']}",
               f"Тариф   : {acc.get('tariff')}",
               f"Сервер  : marfor-mcp {SERVER_VERSION}, наборы инструментов: {_toolsets_label()}"
               + (', только чтение (MARFOR_READONLY)' if _readonly() else '')]
        projects = _projects_line()
        out += [projects] if projects else []
        if _is_public_stand():
            if not projects.startswith('Проектов: 0'):      # пустому аккаунту идти пока некуда
                out.append('Дальше  : marfor_projects → marfor_datasets → marfor_describe → '
                           'marfor_query (подробности — marfor_guide).')
        else:
            out.append("⚠️ Проверьте, что user_id совпадает с владельцем нужных данных: "
                       "таблицы сценариев зовутся user_<id>_<проект>_modeling_<сессия>.")
        out += [x for x in (_newer_version_note(),) if x]
        return '\n'.join(out)

    if name == 'marfor_projects':
        code, js = MF.call('GET', '/api/bi/projects')
        if not isinstance(js, dict) or not js.get('success'):
            return _refusal(code, js)
        rows = [p for p in (js.get('projects') or []) if isinstance(p, dict)]
        if not rows:
            return EMPTY_ACCOUNT_TEXT
        out = ['Проектов: %d   стенд: %s' % (len(rows), BASE_URL), '']
        for p in rows:
            out.append('  %s   «%s»   доступ: %s' % (p.get('id'), p.get('name'), p.get('access')))
        out += ['', 'Дальше: marfor_datasets(project_id="…") — датасеты выбранного проекта. '
                    'Проектов несколько — спросите пользователя, с каким работать.']
        return '\n'.join(out)

    if name in ('marfor_datasets', 'marfor_bi_datasets'):
        return _datasets_text(args)

    if name == 'marfor_describe':
        return _describe_text(args)

    if name == 'marfor_api_get':
        path = args['path']
        if not path.startswith('/api/'):
            return 'Разрешены только пути, начинающиеся с /api/'
        code, js = MF.call('GET', path, params=args.get('params'))
        return _emit(js, args, prefix=f'HTTP {code}\n')

    if name == 'marfor_query':
        js = query(args['session_id'], args['metrics'], args.get('dimensions'),
                   args.get('filters'), args.get('date_from'), args.get('date_to'),
                   args.get('data_source', 'scenario'))
        rows = js.get('data') if isinstance(js, dict) else None
        head = f'строк: {len(rows)}\n' if isinstance(rows, list) else ''
        return _emit(js, args, prefix=head)

    if name == 'marfor_apply_edit':
        res = apply_edit(args['session_id'], args['metric'], args['filters'],
                         float(args['new_value']), wait=args.get('wait', True))
        return json.dumps(res, ensure_ascii=False, indent=1)

    if name == 'marfor_apply_batch':
        edits = args.get('edits')
        if not edits and args.get('edits_file'):
            with open(os.path.expanduser(args['edits_file']), encoding='utf-8') as f:
                edits = json.load(f)
        edits = [_norm_edit(e) for e in (edits or [])]
        if not edits:
            return 'Список правок пуст: передайте edits или edits_file.'
        sid, metric = args['session_id'], args['metric']
        if args.get('dry_run', True):
            out = [f'СУХОЙ ПРОГОН — ничего не изменено. Правок: {len(edits)}',
                   f'Стенд: {BASE_URL}   сессия: {sid}   метрика: {metric}', '']
            tot_base = tot_new = 0.0
            for e in edits[:200]:
                base = _metric_sum(sid, e.get('metric') or metric, e['filters'])
                k = (e['new_value'] / base) if base else None
                tot_base += base or 0
                tot_new += e['new_value']
                out.append('%-52s база %14s → цель %14s  %s' % (
                    (e.get('label') or json.dumps(e['filters'], ensure_ascii=False))[:52],
                    f'{base:,.0f}' if base is not None else 'н/д',
                    f"{e['new_value']:,.0f}",
                    f'×{k:.3f}' if k else ''))
            if len(edits) > 200:
                out.append(f'… и ещё {len(edits) - 200} правок (сверка показана для первых 200)')
            out += ['', f'ИТОГО база {tot_base:,.0f} → цель {tot_new:,.0f}',
                    'Применить: тот же вызов с dry_run=false.']
            return '\n'.join(out)
        job_id = uuid.uuid4().hex[:12]
        with _JOBS_LOCK:
            _JOBS[job_id] = {'status': 'running', 'total': len(edits), 'done': 0,
                             'ok': 0, 'fail': 0, 'stop': False, 'current': -1,
                             'last': None, 'error': None, 'session_id': sid, 'metric': metric}
        threading.Thread(target=batch_worker, daemon=True,
                         args=(job_id, sid, metric, edits,
                               args.get('stop_on_error', True))).start()
        return (f'Батч запущен. job_id={job_id}, правок: {len(edits)}\n'
                f'Стенд: {BASE_URL}, сессия: {sid}, метрика: {metric}\n'
                f'Лог: {_job_path(job_id)}\n'
                f'Прогресс: marfor_job_status(job_id="{job_id}")')

    if name == 'marfor_apply_values':
        vals = args.get('values')
        if not vals and args.get('values_file'):
            with open(os.path.expanduser(args['values_file']), encoding='utf-8') as f:
                vals = json.load(f)
        if not vals:
            return 'Пустой список: передайте values или values_file.'
        dims = args.get('pivot_dimensions') or DEFAULT_PIVOT_DIMS
        changes = [build_change(args['metric'], v.get('slice') or {}, v.get('value'), dims)
                   for v in vals]
        code, js = apply_values(args['session_id'], changes, args.get('scope') or {},
                                dry_run=args.get('dry_run', True))
        if not isinstance(js, dict):
            return f'HTTP {code}: неожиданный ответ сервера'
        # Печатаем ТОЛЬКО то, что вернул сервер. Раньше обёртка писала «ПРИМЕНЕНО»
        # независимо от содержимого — из-за этого 06.08 правка молча не применилась,
        # а в диалоге стояло «применено».
        if not js.get('success'):
            return ('ОТКАЗ СЕРВЕРА — ничего не применено.\n'
                    f'{js.get("message") or js.get("error") or json.dumps(js, ensure_ascii=False)[:600]}')
        if js.get('async') and js.get('task_id'):
            return (f'ЗАДАЧА ПОСТАВЛЕНА (правок отправлено: {len(changes)}). '
                    f'task_id={js["task_id"]}\n'
                    f'Стенд: {BASE_URL}   сессия: {args["session_id"]}\n'
                    f'⚠️ Это ещё НЕ результат. Дождитесь task_status=completed через '
                    f'marfor_job_status(task_id="{js["task_id"]}") '
                    f'и прочитайте, сколько срезов применено.')
        if js.get('dry_run'):
            # Ширина правки — главное, ради чего сухой прогон существует: увидеть
            # «82 193 строки, годы [2023…2026]» ДО записи. Сервер это отдаёт;
            # обёртка обязана показать, иначе предохранитель не доходит до человека.
            prev = js.get('preview') or []
            wide = js.get('multi_year_or_region') or []
            out = [f'СУХОЙ ПРОГОН — ничего не изменено. Правок: {js.get("changes")}',
                   f'Стенд: {BASE_URL}   сессия: {js.get("session_id")}   '
                   f'таблица: {js.get("table")}',
                   f'Строк будет затронуто всего: {js.get("rows_total")}']
            if wide:
                out.append(f'🔴 ШИРОКИХ ПРАВОК: {len(wide)} — выходят за один год или '
                           f'один регион (индексы {wide[:10]}). Проверьте их до записи.')
            nk = js.get('without_mmm_slice_key') or []
            if nk:
                out.append(f'⚠️ Без mmm_slice_key: {len(nk)} правок (индексы {nk[:10]}) — '
                           f'адсток посчитает срезы иначе, чем при ручном вводе')
            out.append('')
            for p in prev[:15]:
                _mark = '🔴' if p.get('i') in wide else '  '
                out.append(f'{_mark} #{p.get("i")}  {p.get("mmm_slice_key") or "—"} '
                           f'→ {p.get("new_value")}')
                out.append(f'      строк {p.get("rows")}   годы {p.get("years")}   '
                           f'регионы {p.get("regions")}')
                out.append(f'      условие: {p.get("where")}')
            if len(prev) > 15:
                out.append(f'   … и ещё {len(prev) - 15} правок (показаны первые 15)')
            out.append('')
            out.append('Сверьте строки/годы/регионы с ожиданием. Если сходится — '
                       'тот же вызов с dry_run=false.')
            return '\n'.join(out)
        return ('РЕЗУЛЬТАТ СЕРВЕРА:\n'
                f'  применено   : {js.get("applied")} из {js.get("changes")}\n'
                f'  без эффекта : {len(js.get("unchanged") or [])}\n'
                f'  ошибок      : {len(js.get("failed") or [])}\n'
                f'  цепочка     : {"пересчитана" if js.get("chain_propagated") else "НЕ пересчитана"}\n'
                f'  {js.get("message") or ""}')

    if name == 'marfor_job_status':
        if args.get('task_id'):
            return _task_status_text(args)
        if not args.get('job_id'):
            return ('Передайте job_id (фоновый батч marfor_apply_batch) или task_id (задача из '
                    'ответа marfor_apply_values / marfor_apply_edit).')
        job_id = args['job_id']
        with _JOBS_LOCK:
            job = dict(_JOBS.get(job_id) or {})
        if not job:
            return f'Задача {job_id} не найдена (сервер перезапускался?). Лог: {_job_path(job_id)}'
        lines = [f"job {job_id}: {job['status']}  {job['done']}/{job['total']}  "
                 f"успешно {job['ok']}, ошибок {job['fail']}"]
        if job.get('error'):
            lines += ['ЗАДАНИЕ ОСТАНОВЛЕНО: стенд отверг токен. Оставшиеся правки не отправлены: '
                      '%d из %d. После нового входа запустите заново только их — применённые '
                      'видны в журнале ниже.' % (job['total'] - job['done'], job['total']),
                      job['error']]
        tail = int(args.get('tail') or 5)
        try:
            with open(_job_path(job_id), encoding='utf-8') as f:
                recs = f.readlines()[-tail:]
            lines += ['  ' + r.strip() for r in recs]
        except OSError:
            pass
        return '\n'.join(lines)

    if name == 'marfor_job_stop':
        with _JOBS_LOCK:
            job = _JOBS.get(args['job_id'])
            if not job:
                return 'Задача не найдена'
            job['stop'] = True
        return 'Остановится после текущей правки.'

    # marfor_bi_datasets — прежнее имя marfor_datasets, обслуживается выше той же функцией

    if name == 'marfor_bi_dashboards':
        params = {'project': args['project_id']} if args.get('project_id') else None
        code, js = MF.call('GET', '/api/bi/dashboards', params=params)
        if not isinstance(js, dict) or not js.get('success'):
            return 'ОТКАЗ СЕРВЕРА (HTTP %s): %s' % (
                code, (js or {}).get('message') or str(js)[:300])
        rows = js.get('dashboards') or []
        out = ['Дашбордов: %d   Стенд: %s' % (len(rows), BASE_URL), '']
        for r in rows:
            pub = 'опубликован' if r.get('is_published') else '—'
            out.append('%s\n    «%s» · %s · роль %s · проект «%s» · публикация: %s'
                       % (r.get('id'), r.get('title'), r.get('kind'),
                          r.get('role'), r.get('project_name'), pub))
        return '\n'.join(out)

    if name == 'marfor_bi_generate':
        mode = 'revise' if args.get('mode') == 'revise' else 'create'
        if mode == 'revise' and not args.get('dashboard_id'):
            return 'Для mode=revise обязателен dashboard_id.'
        if mode == 'create' and not args.get('project_id'):
            return 'Для создания обязателен project_id.'
        if mode == 'create' and not args.get('dataset_keys'):
            return ('Для создания обязательны dataset_keys — возьмите их из '
                    'marfor_bi_datasets(project_id).')
        body = {'prompt': args['prompt'], 'mode': mode}
        for k in ('project_id', 'skill', 'dataset_keys', 'dashboard_id'):
            if args.get(k):
                body[k] = args[k]
        t = int(args.get('timeout_s') or max(BI_GENERATE_TIMEOUT, HTTP_TIMEOUT))
        code, js = MF.call('POST', '/api/bi/ai/generate', data=body, timeout=t)
        if not isinstance(js, dict) or not js.get('success'):
            return 'ОТКАЗ СЕРВЕРА (HTTP %s): %s' % (
                code, (js or {}).get('message') or str(js)[:400])
        return ('ГОТОВО. Дашборд %s — «%s»\n'
                'Стенд: %s   редактор: %s%s\n'
                'Дальше: marfor_bi_get_page — забрать HTML; '
                'marfor_bi_publish — опубликовать.'
                % (js.get('dashboard_id'), js.get('title'), BASE_URL,
                   BASE_URL, js.get('url')))

    if name == 'marfor_bi_get_page':
        did = args['dashboard_id']
        code, raw = MF.fetch_raw('/api/bi/ai_pages/%s/raw' % did)
        text = raw.decode('utf-8', 'replace')
        if code != 200:
            return 'ОТКАЗ СЕРВЕРА (HTTP %s): %s' % (code, text[:300])
        if not text.lstrip()[:200].lower().startswith(('<!doctype', '<html')):
            # сервер отвечает HTML и на «страница не сгенерирована»
            if 'не сгенерирована' in text[:400]:
                return 'Страница дашборда %s ещё не сгенерирована.' % did
        p = os.path.expanduser(args.get('save_to') or
                               os.path.join(PAGE_DIR, '%s.html' % did))
        os.makedirs(os.path.dirname(p) or '.', exist_ok=True)
        with open(p, 'w', encoding='utf-8') as f:
            f.write(text)
        return ('Сохранено: %s\nРазмер: %d байт   <title>: %s\nСтенд: %s'
                % (p, len(raw), _bi_html_title(text), BASE_URL))

    if name == 'marfor_bi_put_page':
        html = args.get('html')
        if not html and args.get('html_file'):
            with open(os.path.expanduser(args['html_file']), encoding='utf-8') as f:
                html = f.read()
        if not html:
            return 'Передайте html_file (путь к файлу) или html (строка).'
        code, js = MF.call('PUT', '/api/bi/ai_pages/%s' % args['dashboard_id'],
                           data={'html': html})
        if not isinstance(js, dict) or not js.get('success'):
            return 'ОТКАЗ СЕРВЕРА (HTTP %s): %s' % (
                code, (js or {}).get('message') or str(js)[:300])
        return ('ЗАПИСАНО: %d байт в AI-страницу дашборда %s\n'
                'Стенд: %s   редактор: %s%s\n'
                'Опубликовать: marfor_bi_publish(dashboard_id="%s")'
                % (js.get('bytes') or len(html.encode('utf-8')),
                   js.get('dashboard_id'), BASE_URL, BASE_URL,
                   js.get('url'), js.get('dashboard_id')))

    if name == 'marfor_bi_publish':
        return bi_publish(args['dashboard_id'], slug=args.get('slug'),
                          mode=args.get('mode'),
                          unpublish=bool(args.get('unpublish')),
                          chrome=args.get('chrome'))

    # ── калибровка MMM ───────────────────────────────────────────────────────
    # Считает всё стенд MARFOR, сервер только носит запросы: прогон MMM идёт часами и на
    # Маке ему делать нечего. Тяжёлое здесь — только форматирование ответа под чтение.

    if name == 'marfor_mmm_params':
        code, js = MF.call('GET', '/api/mmm_calib/params')
        err = _mmm_err(code, js, 'marfor_mmm_params')
        if err:
            return err
        return _emit({k: v for k, v in js.items() if k != 'success'}, args,
                     prefix='Ручки калибровки MMM (стенд %s, env=%s)\n'
                            % (BASE_URL, js.get('env')))

    if name == 'marfor_mmm_calibrate':
        code, js = MF.call('POST', '/api/mmm_calib/run/%s' % _q(args['session_id']),
                           data={'params': args.get('params') or {},
                                 'label': args.get('label') or ''})
        if isinstance(js, dict) and js.get('in_progress'):
            return ('Прогон НЕ запущен: %s\nПосмотрите ход: marfor_mmm_runs(session_id="%s")'
                    % (js.get('message'), args['session_id']))
        err = _mmm_err(code, js, 'marfor_mmm_calibrate')
        if err:
            return err
        p = js.get('params') or {}
        out = ['Прогон запущен: run_id = %s' % js.get('run_id'),
               'Стенд   : %s' % BASE_URL,
               'Сессия  : %s' % args['session_id']]
        if args.get('label'):
            out.append('Пометка : %s' % args['label'])
        out.append('Ручки   : ' + ', '.join('%s=%s' % (k, v) for k, v in sorted(p.items())))
        if js.get('unknown_params'):
            out.append('⚠️ Незнакомые параметры отброшены: %s (список ручек — marfor_mmm_params)'
                       % ', '.join(js['unknown_params']))
        if int(p.get('holdout_periods') or 0):
            out.append('⚠️ Это ПРОВЕРОЧНЫЙ прогон: последние %s периодов скрыты от обучения. '
                       'Доли каналов из него брать нельзя — он мерит точность.'
                       % p['holdout_periods'])
        out += ['', 'Счёт идёт фоном, ничего в модели не меняется.',
                'Ход      : marfor_mmm_runs(session_id="%s")' % args['session_id'],
                'Результат: marfor_mmm_report(run_id="%s")' % js.get('run_id')]
        return '\n'.join(out)

    if name == 'marfor_mmm_runs':
        code, js = MF.call('GET', '/api/mmm_calib/runs/%s' % _q(args['session_id']),
                           params={'limit': args.get('limit') or 30})
        err = _mmm_err(code, js, 'marfor_mmm_runs')
        if err:
            return err
        runs = js.get('runs') or []
        if not runs:
            return ('В сессии %s прогонов калибровки ещё нет.\nЗапустить: '
                    'marfor_mmm_calibrate(session_id="%s")' % (args['session_id'], args['session_id']))
        if args.get('save_to') or args.get('max_chars'):
            return _emit(runs, args, prefix='Прогоны сессии %s\n' % args['session_id'])
        lines = ['Прогоны калибровки MMM, сессия %s (стенд %s)' % (args['session_id'], BASE_URL), '']
        for r in runs:
            h = r.get('headline') or {}
            head = ('платные %s %% (ласт-клик %s %%), rhat %s, дивергенций %s'
                    % (h.get('paid_mmm_pct'), h.get('paid_lastclick_pct'),
                       h.get('rhat_max'), h.get('divergences'))) if h else ''
            lines.append('%-10s %-9s %s' % (r.get('run_id'), r.get('status'),
                                            r.get('label') or ''))
            lines.append('   ручки : ' + ', '.join('%s=%s' % (k, v)
                                                   for k, v in sorted((r.get('params') or {}).items())))
            if r.get('status') == 'running':
                lines.append('   идёт  : %s %% — %s' % (r.get('progress'), r.get('message')))
            elif r.get('status') == 'stalled':
                lines.append('   ⚠️ считался, но перестал подавать признаки жизни (процесс убит?)')
            elif r.get('error'):
                lines.append('   ошибка: %s' % r['error'])
            if head:
                lines.append('   итог  : %s   (%s с)' % (head, r.get('elapsed_sec')))
            if r.get('applied_to_model'):
                lines.append('   ✅ применён к модели %s (%s)'
                             % (r['applied_to_model'].get('model_id'), r['applied_to_model'].get('at')))
            lines.append('')
        return '\n'.join(lines)

    if name == 'marfor_mmm_report':
        code, js = MF.call('GET', '/api/mmm_calib/report/%s' % _q(args['run_id']),
                           params=({'market': args['market']} if args.get('market') else None))
        err = _mmm_err(code, js, 'marfor_mmm_report')
        if err:
            return err
        rep = js.get('report') or {}
        acc = rep.get('acceptance') or {}
        pref = 'Отчёт по прогону %s' % args['run_id']
        if rep.get('label'):
            pref += ' («%s»)' % rep['label']
        pref += '\nСтенд: %s\n' % BASE_URL
        if rep.get('status') != 'done':
            return pref + ('Статус: %s (%s %%) — %s\n%s'
                           % (rep.get('status'), rep.get('progress'), rep.get('message'),
                              rep.get('error') or ''))
        if acc:
            bad = [c['name'] for c in (acc.get('checks') or []) if not c.get('ok')]
            pref += ('Приёмка: %s%s\n' % ('пройдена' if acc.get('passed') else 'НЕ пройдена',
                                          ('; провалено: ' + ', '.join(bad)) if bad else ''))
        if rep.get('ВНИМАНИЕ'):
            pref += '⚠️ %s\n' % rep['ВНИМАНИЕ']
        return _emit(rep, args, prefix=pref)

    if name == 'marfor_mmm_compare':
        ids = [str(x) for x in (args.get('run_ids') or []) if x]
        if len(ids) < 2:
            return 'Нужно минимум два run_id: сравнивать нечего.'
        code, js = MF.call('POST', '/api/mmm_calib/compare', data={'run_ids': ids})
        err = _mmm_err(code, js, 'marfor_mmm_compare')
        if err:
            return err
        spread = js.get('разброс доли платных между прогонами, п.п.')
        pref = 'Сравнение прогонов: %s\nСтенд: %s\n' % (', '.join(ids), BASE_URL)
        if spread is not None:
            pref += ('Разброс доли платных между вариантами: %s п.п. — %s\n'
                     % (spread, 'вывод держится на данных' if spread < 5 else
                        'вывод сильно зависит от допущений, одним прогоном решать нельзя'))
        return _emit({k: v for k, v in js.items() if k != 'success'}, args, prefix=pref)

    if name == 'marfor_mmm_diagnose':
        code, js = MF.call('GET', '/api/mmm_calib/diagnose/%s' % _q(args['run_id']),
                           params=({'market': args['market']} if args.get('market') else None))
        err = _mmm_err(code, js, 'marfor_mmm_diagnose')
        if err:
            return err
        d = js.get('diagnosis') or {}
        pref = 'Диагностика прогона %s, рынок «%s»\nСтенд: %s\n' % (
            args['run_id'], d.get('market'), BASE_URL)
        if d.get('слабо идентифицируемы'):
            pref += ('⚠️ Держатся на приоре, а не на данных: %s\n'
                     % ', '.join(d['слабо идентифицируемы']))
        if d.get('спутаны с сезоном'):
            pref += ('⚠️ Спутаны с сезоном (расход идёт в такт календарю): %s\n'
                     % ', '.join(d['спутаны с сезоном']))
        return _emit(d, args, prefix=pref)

    if name == 'marfor_mmm_apply':
        confirm = bool(args.get('confirm'))
        # Клиентский стоп-кран: на боевом стенде не отправляем запрос вообще. Отказ есть и на
        # сервере, но он появится там только после выкатки — а промахнуться стендом можно уже сейчас.
        if confirm and not _is_dev_stand():
            return ('ОТКАЗ. Стенд %s — боевой, применять калибровочный прогон к модели из чата '
                    'нельзя.\nПоложите заявку в шлюз (deploy_gate/inbox/ГГММДД_имя.md): run_id=%s, '
                    'параметры прогона и к какой модели применить — применение сделает дежурный.\n'
                    'Если это должен был быть dev — запустите сервер с '
                    'MARFOR_BASE_URL=https://dev.marfor.pro.' % (BASE_URL, args['run_id']))
        data = {'confirm': confirm}
        if args.get('model_id'):
            data['model_id'] = str(args['model_id'])
        code, js = MF.call('POST', '/api/mmm_calib/apply/%s' % _q(args['run_id']), data=data)
        if isinstance(js, dict) and js.get('refused') == 'prod':
            return 'ОТКАЗ СТЕНДА: %s' % js.get('message')
        if isinstance(js, dict) and js.get('needs_confirm'):
            rep = js.get('preview') or {}
            acc = rep.get('acceptance') or {}
            out = ['СУХОЙ ПРОГОН — ничего не изменено.',
                   'Прогон : %s%s' % (args['run_id'],
                                      (' («%s»)' % rep['label']) if rep.get('label') else ''),
                   'Стенд  : %s' % BASE_URL,
                   'Приёмка: %s' % ('пройдена' if acc.get('passed') else 'НЕ пройдена')]
            h = rep.get('headline') or {}
            if h:
                out.append('Итог   : платные %s %% против %s %% по ласт-клику, рынков %s'
                           % (h.get('paid_mmm_pct'), h.get('paid_lastclick_pct'), h.get('markets')))
            if rep.get('ВНИМАНИЕ'):
                out.append('⚠️ %s' % rep['ВНИМАНИЕ'])
            out += ['', 'В модель уедут доли каналов из этого прогона (эластичности заменятся).',
                    'Применить: тот же вызов с confirm=true.']
            return '\n'.join(out)
        err = _mmm_err(code, js, 'marfor_mmm_apply')
        if err:
            return err
        out = ['ЗАПИСАНО: прогон %s применён к модели %s' % (args['run_id'], js.get('model_id')),
               'Стенд: %s' % BASE_URL]
        for w in (js.get('warnings') or [])[:6]:
            out.append('⚠️ %s' % w)
        return '\n'.join(out)

    return f'Неизвестный инструмент: {name}'


def _is_dev_stand():
    # Дев узнаём по имени хоста стенда. Нужно ровно для одного: не дать применить
    # калибровочный прогон к боевой модели, промахнувшись стендом.
    host = urllib.parse.urlparse(BASE_URL).hostname or ''
    return host.startswith('dev.') or host in ('127.0.0.1', 'localhost')


def _q(v):
    # Сегмент пути: run_id и session_id приходят из чата, и не-ASCII в них не должен
    # валить urllib на кодировании URL.
    return urllib.parse.quote(str(v), safe='')


def _mmm_err(code, js, tool):
    # У всех ручек калибровки один конверт {success, message}. Отдельно ловим случай
    # «роутов на стенде ещё нет»: это не поломка, а невыкаченный патч, и сказать надо именно так.
    if code == 404:
        return ('На стенде %s нет ручек калибровки MMM (HTTP 404). Они живут на dev; '
                'в бой едут отдельной заявкой через шлюз. Запустите сервер с '
                'MARFOR_BASE_URL=https://dev.marfor.pro.' % BASE_URL)
    if not isinstance(js, dict) or js.get('_not_json'):
        return 'ОТКАЗ СЕРВЕРА (HTTP %s) на %s: %s' % (code, tool, str(js)[:400])
    if not js.get('success'):
        return 'ОТКАЗ СЕРВЕРА (HTTP %s) на %s: %s' % (code, tool, js.get('message') or str(js)[:400])
    return None


def _norm_edit(e):
    if 'filters' in e and 'new_value' in e:
        return e
    # формат сегодняшнего edits.json: {kind, channel, segment, category, filters, new_value}
    return {'label': e.get('label') or '|'.join(
        str(e.get(k, '')) for k in ('channel', 'segment', 'category') if e.get(k)),
        'filters': e['filters'], 'new_value': e['new_value'],
        'metric': e.get('metric')}


def _emit(js, args, prefix=''):
    full = json.dumps(js, ensure_ascii=False, indent=1)
    if args.get('save_to'):
        p = os.path.expanduser(args['save_to'])
        os.makedirs(os.path.dirname(p) or '.', exist_ok=True)
        with open(p, 'w', encoding='utf-8') as f:
            f.write(full)
        prefix += f'полный ответ сохранён: {p} ({len(full)} символов)\n'
    cap = int(args.get('max_chars') or 8000)
    if len(full) > cap:
        return prefix + full[:cap] + f'\n… обрезано, всего {len(full)} символов'
    return prefix + full


# ─────────────────────── служебные команды (не MCP-режим) ─────────────────────
# Разбираются ДО цикла stdio; stdout здесь — обычный текст для человека или ассистента.
# Вход разбит на --login-start / --login-finish не из любви к флагам: у оболочки, которой
# пользуется ассистент, нет терминала с вводом, а время одной команды ограничено (120 с) —
# один блокирующий --login там невозможен. Это предел КОМАНДЫ, а не входа: подтверждённый
# вход стенд держит до исходного срока кода (10 минут от --login-start), --login-finish
# можно повторять. Секрет при этом не проходит ни через чат, ни через конфигурацию клиента,
# ни через историю оболочки: он сразу ложится в token.json.

LOGIN_WAIT_MAX = 110        # потолок --wait: запас под предел времени команды у оболочки ассистента (120 с)
LOGIN_WAIT_DEFAULT = 100
# Таймаут одного опроса; у последнего — короче, см. _login_finish. Стенд выпускает токен
# ВНУТРИ ответа на опрос (запись в базу): оборви клиент такой запрос раньше времени — токен
# создан, а получить его уже некому. Поэтому запас щедрый; обычный опрос отвечает за миллисекунды,
# а потолок всей команды держит не эта константа, а срок ожидания (--wait).
POLL_HTTP_TIMEOUT = 15


class _CliError(Exception):
    """Отказ служебной команды: текст для человека, код возврата 1."""


def _unlink(path):
    try:
        os.unlink(path)
    except OSError:
        pass


def _client_label():
    """Как подключение назовётся на странице подтверждения и в списке токенов на сайте:
    по имени компьютера человек отличит свой ноутбук от чужого запроса."""
    try:
        host = socket.gethostname().split('.')[0]
    except Exception:
        host = ''
    return ('Claude @ %s' % host if host else 'Claude')[:60]


def _manual_token_hint():
    return ('Запасной путь — токен вручную: создайте его на %s/account (раздел «Подключение '
            'Claude (MCP)») и сохраните командой\n  %s\nОна спросит токен скрытым вводом, '
            'поэтому ей нужен настоящий терминал; в чат и в командную строку токен не '
            'вставляйте.%s' % (PUBLIC_BASE, _self_cmd('--set-token'), _win_notes()))


def _device_start():
    code, _, raw = MF._raw('POST', '/api/mcp/auth/start', data={'client': _client_label()},
                           timeout=30, auth=False)
    js = _json_or_none(raw)
    js = js if isinstance(js, dict) else {}
    if code == 503 or js.get('code') == 'device_flow_unavailable':
        # Код в скобках — устойчивый признак для инструкции ассистенту (install.md):
        # русскую фразу можно переписать, а по коду случай узнаётся всегда.
        raise _CliError('Вход подтверждением в браузере сейчас недоступен на %s '
                        '(device_flow_unavailable, HTTP %s).\n%s'
                        % (BASE_URL, code, _manual_token_hint()))
    if code == 404:
        raise _CliError('На стенде %s нет входа подтверждением в браузере (HTTP 404).\n%s'
                        % (BASE_URL, _manual_token_hint()))
    if code == 429:
        raise _CliError('Слишком много попыток входа подряд. Подождите 10 минут и повторите.')
    if code != 200 or not js.get('success') or not js.get('device_code') \
            or not js.get('user_code'):
        raise _CliError('Стенд %s не начал вход: HTTP %s %s\n%s'
                        % (BASE_URL, code, js.get('message') or '', _manual_token_hint()))
    return js


def _pending_save(js):
    """Начатый вход → pending.json (0600). device_code — почти секрет: им забирают токен,
    поэтому он живёт только в этом файле и на экран не печатается."""
    code = str(js['user_code'])
    url = js.get('verification_uri_complete') or (
        (js.get('verification_uri') or BASE_URL + '/mcp/connect')
        + '?code=' + urllib.parse.quote(code))
    rec = {'base_url': BASE_URL, 'device_code': str(js['device_code']), 'user_code': code,
           'url': str(url),
           'interval': max(1, min(int(js.get('interval') or 3), 30)),
           'expires_at': time.time() + max(30, min(int(js.get('expires_in') or 600), 3600))}
    _write_private_json(PENDING_FILE, rec)
    return rec


def _print_login_invite(rec):
    mins = max(1, int(round((rec['expires_at'] - time.time()) / 60.0)))
    print('Откройте ссылку в браузере, сверьте код и нажмите «Разрешить»:')
    print('URL:  %s' % rec['url'])
    print('CODE: %s' % rec['user_code'])
    print('Ссылка действует %d мин. Код на странице обязан совпасть с кодом выше; не совпал — '
          'нажмите «Отклонить».' % mins)


def _open_browser(url):
    if os.environ.get('MARFOR_NO_BROWSER'):
        return False
    u = urllib.parse.urlsplit(url)
    local = (u.hostname or '').lower() in ('localhost', '127.0.0.1')
    if not (u.scheme == 'https' or (u.scheme == 'http' and local)):
        return False
    if sys.platform.startswith('linux') and not (os.environ.get('DISPLAY')
                                                 or os.environ.get('WAYLAND_DISPLAY')):
        return False        # без графики поднялся бы консольный браузер и занял терминал
    try:
        import webbrowser
        return bool(webbrowser.open(url))
    except Exception:
        return False


def _device_poll(device_code, timeout=POLL_HTTP_TIMEOUT):
    """→ (статус, токен, пояснение). Статус: pending | approved | denied | expired —
    ответ стенда; retry — временный сбой, стоит спросить ещё раз; error — продолжать незачем.
    Пояснение у denied/expired — поле message стенда, если оно есть: токен выпускается в
    момент опроса, и отказ может прийти уже после «Разрешить» (например, исчерпан предел
    в 10 активных токенов) — причину человеку надо показать, а не заменять общей фразой."""
    try:
        code, _, raw = MF._raw('POST', '/api/mcp/auth/poll', data={'device_code': device_code},
                               timeout=timeout, auth=False)
    except RuntimeError as e:
        return 'retry', None, str(e)
    js = _json_or_none(raw)
    js = js if isinstance(js, dict) else {}
    st = js.get('status')
    if st in ('pending', 'approved', 'denied', 'expired'):
        msg = js.get('message')
        return st, js.get('token'), (str(msg).strip()[:500] or None) if msg else None
    if code == 429 or code >= 500:
        return 'retry', None, 'стенд ответил HTTP %s' % code
    return 'error', None, 'HTTP %s %s' % (code, js.get('message') or '')


def _report_saved(tok):
    print('Готово: токен сохранён в %s' % TOKEN_FILE)
    if (os.environ.get('MARFOR_TOKEN') or '').strip():
        print('⚠️ Задана переменная MARFOR_TOKEN — она важнее файла. Уберите её, чтобы работал '
              'сохранённый токен.')
    try:
        acc = MF.token_login(tok, 'file')
        print('Вы вошли как %s (user_id=%s), стенд %s' % (acc['email'], acc['user_id'], BASE_URL))
    except RuntimeError as e:
        print('⚠️ Токен сохранён, но пробный вызов не прошёл: %s' % e)
    print('Если сервер уже подключён к клиенту — продолжайте работу, перезапуск не нужен.\n'
          'Если ещё нет — команду регистрации печатает:\n  %s%s'
          % (_self_cmd('--print-config', 'claude-code'), _win_notes()))
    return 0


def _login_finish(wait):
    """0 — токен сохранён; 2 — ещё не подтверждено (подождать и повторить); 1 — отказ,
    истекло или начинать заново."""
    rec = _read_json(PENDING_FILE)
    if not rec.get('device_code') or rec.get('base_url') != BASE_URL:
        print('Начатого входа для %s нет. Начните: %s%s'
              % (BASE_URL, _self_cmd('--login-start'), _win_notes()), file=sys.stderr)
        return 1
    again = 'Начните заново: %s%s' % (_self_cmd('--login-start'), _win_notes())
    deadline, note = time.time() + wait, None
    while True:
        if time.time() >= float(rec.get('expires_at') or 0):
            _unlink(PENDING_FILE)
            print('Время на подтверждение вышло. ' + again, file=sys.stderr)
            return 1
        # Чем ближе потолок ожидания, тем короче таймаут опроса: вся команда обязана
        # уложиться в предел времени команды (120 с), иначе оболочка ассистента убьёт её
        # посреди запроса.
        st, tok, note = _device_poll(
            rec['device_code'],
            timeout=min(POLL_HTTP_TIMEOUT, max(5, int(deadline - time.time()))))
        if st == 'approved':
            if not isinstance(tok, str) or not TOKEN_RE.fullmatch(tok):
                _unlink(PENDING_FILE)
                print('Вход подтверждён, но токен не получен: он выдаётся один раз и, видимо, '
                      'уже был забран. ' + again, file=sys.stderr)
                return 1
            _save_token(tok)            # сначала сохранить: токен стенд отдаёт ОДИН раз
            _unlink(PENDING_FILE)
            return _report_saved(tok)
        if st in ('denied', 'expired', 'error'):
            _unlink(PENDING_FILE)
            # Пояснение стенда важнее общей фразы: отказ может прийти и после «Разрешить».
            said = (note.rstrip('. ') + '. ') if note else ''
            print({'denied': ('Вход не состоялся: ' + said) if said
                             else 'Вход отклонён на странице подтверждения. ',
                   'expired': 'Код входа истёк или уже использован. ' + said,
                   'error': 'Стенд не принял опрос входа (%s). ' % note}[st] + again,
                  file=sys.stderr)
            return 1
        pause = int(rec.get('interval') or 3)
        if time.time() + pause > deadline:
            break
        time.sleep(pause)
    print('Вход ещё не подтверждён%s.' % ((' (%s)' % note) if note else ''))
    print('URL:  %s' % rec.get('url'))
    print('CODE: %s' % rec.get('user_code'))
    # Подтверждённый вход стенд держит до исходного срока кода, а токен выпускает в момент
    # опроса: нажать «Разрешить» и вернуться к этой команде можно в любой момент до конца срока.
    left = max(1, int(round((float(rec.get('expires_at') or 0) - time.time()) / 60.0)))
    print('Код действует ещё %d мин.: нажать «Разрешить» и повторить эту команду можно в любой '
          'момент до конца срока.' % left)
    print('Попросите пользователя нажать «Разрешить» и повторите: %s%s'
          % (_self_cmd('--login-finish'), _win_notes()))
    return 2


def _cli_login_start():
    rec = _pending_save(_device_start())
    _print_login_invite(rec)
    # «Сразу», а не «после подтверждения»: команда сама ждёт нажатия «Разрешить». Пока её не
    # запустили, стенд никто не опрашивает — ассистент, ждущий ответа в чате, терял на этом вход.
    print('Сразу запустите завершение входа — команда сама дождётся нажатия «Разрешить» '
          '(код возврата 2 — ещё не подтверждено, повторите её):\n  %s%s'
          % (_self_cmd('--login-finish'), _win_notes()))
    return 0


def _cli_login():
    rec = _pending_save(_device_start())
    _print_login_invite(rec)
    if _open_browser(rec['url']):
        print('Ссылка открыта в браузере.')
    print('Жду подтверждения… (Ctrl+C — прервать; завершить позже: %s)%s'
          % (_self_cmd('--login-finish'), _win_notes()))
    sys.stdout.flush()      # ждать предстоит минуты: приглашение должно быть на экране сразу
    try:
        return _login_finish(max(0.0, rec['expires_at'] - time.time()))
    except KeyboardInterrupt:
        print('\nПрервано. Вход можно завершить позже: %s' % _self_cmd('--login-finish'))
        return 130


def _stdin_is_tty():
    try:
        return bool(sys.stdin.isatty())
    except (AttributeError, ValueError):        # поток закрыт или подменён
        return False


def _cli_set_token(from_stdin=False):
    if sys.stdin is None:
        raise _CliError('Нет потока ввода: запустите команду в терминале.')
    if _stdin_is_tty():
        import getpass
        tok = getpass.getpass('Вставьте токен MARFOR (ввод не отображается): ')
    elif from_stdin:
        # Явный --stdin: токен пришёл по конвейеру или из файла. Эха у конвейера нет,
        # приглашения не печатаем — читать его некому.
        tok = sys.stdin.readline()
    else:
        # Без терминала скрытого ввода не бывает. Раньше команда здесь молча читала stdin: в
        # Git Bash (mintty) на Windows родной Python видит вместо терминала канал, человек
        # получал «зависшую» команду без приглашения, а вставленный токен — открытым на экране.
        raise _CliError(
            '--set-token спрашивает токен скрытым вводом, а у этой оболочки нет терминала '
            '(stdin — не TTY): так бывает у ассистента и в Git Bash на Windows. Ничего не '
            'прочитано и не сохранено.\n'
            'Запустите команду в обычном терминале (Windows: PowerShell или cmd; в Git Bash — '
            'через winpty). Осознанно передать токен через стандартный ввод, минуя экран и '
            'историю оболочки, позволяет флаг --stdin:\n  %s\n'
            'Токен подаётся ей на стандартный ввод из буфера обмена или из файла (macOS: '
            '«pbpaste | команда», PowerShell: «Get-Clipboard | команда», bash: «команда < '
            'файл»). В чат и в аргументы команды токен не вставляйте.%s'
            % (_self_cmd('--set-token', '--stdin'), _win_notes()))
    tok = (tok or '').strip()
    if not TOKEN_RE.fullmatch(tok):
        raise _CliError('Это не похоже на токен MARFOR: он начинается с marfor_pat_ и состоит '
                        'из латинских букв, цифр, «-» и «_» (всего 54 символа). Проверьте, что '
                        'скопировали его целиком. Ничего не сохранено.')
    acc = MF.token_login(tok, 'file')           # пробный вызов; отказ → RuntimeError
    _save_token(tok)
    print('Токен принят: %s (user_id=%s), стенд %s' % (acc['email'], acc['user_id'], BASE_URL))
    print('Сохранён в %s' % TOKEN_FILE)
    if (os.environ.get('MARFOR_TOKEN') or '').strip():
        print('⚠️ Задана переменная MARFOR_TOKEN — она важнее файла.')
    return 0


def _cli_logout():
    had = _forget_token()
    _unlink(PENDING_FILE)
    print(('Токен для %s удалён из %s.' % (BASE_URL, TOKEN_FILE)) if had
          else ('Сохранённого токена для %s не было.' % BASE_URL))
    if (os.environ.get('MARFOR_TOKEN') or '').strip():
        print('⚠️ Токен задан ещё и переменной MARFOR_TOKEN — уберите её из конфигурации клиента.')
    if had:
        print('На стенде токен остаётся действующим, пока вы его не отзовёте: %s/account, '
              'раздел «Подключение Claude (MCP)».' % PUBLIC_BASE)
    return 0


def _self_sha256():
    try:
        with open(os.path.abspath(__file__), 'rb') as f:
            return hashlib.sha256(f.read()).hexdigest()
    except OSError:
        return None


def _cli_check():
    """Диагностика для человека и для ассистента. Секрет НЕ печатается: от токена видны
    только последние 4 символа — те же, что показывает список токенов на сайте."""
    import platform
    sha = _self_sha256()
    print('marfor-mcp %s' % SERVER_VERSION)
    print('Файл    : %s' % os.path.abspath(__file__))
    print('sha256  : %s' % (sha or 'не удалось прочитать файл'))
    print('Python  : %s (%s), %s' % (platform.python_version(), sys.executable, sys.platform))
    print('Стенд   : %s%s' % (BASE_URL, ' (публичный)' if _is_public_stand() else ''))
    bad = _base_url_problem()
    if bad:
        print('ИТОГ    : ОТКАЗ — ' + bad)
        return 1
    print('Наборы  : %s%s%s' % (
        _toolsets_label(),
        '' if (os.environ.get('MARFOR_TOOLSETS') or '').strip() else ' (умолчание для стенда)',
        '; только чтение (MARFOR_READONLY)' if _readonly() else ''))
    tok, src = _find_token()
    if tok:
        where = 'переменная MARFOR_TOKEN' if src == 'env' else 'файл %s' % TOKEN_FILE
        # хвост показываем только у настоящего токена: в переменной может лежать что угодно
        print('Токен   : найден — %s, %s' % (where, ('оканчивается на …%s' % tok[-4:])
                                             if TOKEN_RE.fullmatch(tok)
                                             else 'но на токен MARFOR не похож'))
        if src == 'file' and os.name != 'nt':
            try:
                mode = os.stat(TOKEN_FILE).st_mode & 0o777
                if mode != 0o600:
                    print('          ⚠️ права файла %o, нужно 600: chmod 600 %s'
                          % (mode, _shq(TOKEN_FILE)))
            except OSError:
                pass
    elif EMAIL:
        print('Токен   : не найден; вход паролем под %s%s'
              % (EMAIL, '' if BASE_URL_EXPLICIT else ' — но MARFOR_BASE_URL не задан'))
    else:
        print('Токен   : не найден')
        print('ИТОГ    : вход не настроен. Войти: %s%s' % (_self_cmd('--login'), _win_notes()))
        return 1
    try:
        MF.logged_in = False
        MF.ensure()
    except Exception as e:
        print('Вход    : ОТКАЗ — %s' % e)
        print('ИТОГ    : вход не работает (см. выше)')
        return 1
    acc = MF.account
    print('Вход    : OK — %s (user_id=%s, тариф %s)'
          % (acc.get('email'), acc.get('user_id'), acc.get('tariff')))
    line = _projects_line()
    if line:
        print(line.split('\n', 1)[0])
    pub = _published_version()
    if pub:
        theirs, ours = _ver_tuple(pub.get('version')), _ver_tuple(SERVER_VERSION)
        if theirs > ours:
            print('Версия  : на сайте новее — %s. Обновление: скачать %s/mcp/server.py поверх '
                  'этого файла.' % (pub.get('version'), PUBLIC_BASE))
        elif theirs < ours:
            print('Версия  : у вас новее опубликованной (%s)' % pub.get('version'))
        elif sha and pub.get('sha256') and pub['sha256'] != sha:
            print('Версия  : ⚠️ файл отличается от опубликованного %s (sha256 не совпал): он '
                  'изменён или скачан не полностью. Скачайте заново: %s/mcp/server.py'
                  % (SERVER_VERSION, PUBLIC_BASE))
        elif sha and pub.get('sha256'):
            print('Версия  : актуальная, sha256 совпадает с опубликованным')
        else:
            print('Версия  : актуальная')
    print('ИТОГ    : всё в порядке')
    return 0


def _config_env():
    """Несекретные настройки из текущего окружения — их переносим в конфигурацию клиента.
    Токен, пароль и email сюда не попадают НИКОГДА: секрет живёт в token.json."""
    env = {}
    if BASE_URL != PUBLIC_BASE:
        env['MARFOR_BASE_URL'] = BASE_URL
    for key in ('MARFOR_TOOLSETS', 'MARFOR_READONLY'):
        val = (os.environ.get(key) or '').strip()
        if val:
            env[key] = val
    return env


def _cli_print_config(client):
    """Готовая регистрация сервера с АБСОЛЮТНЫМИ путями: графический клиент не наследует
    PATH оболочки, и голое «python3» у него может оказаться другим Python или не найтись."""
    exe, path, env = sys.executable or 'python3', os.path.abspath(__file__), _config_env()
    if client == 'claude-desktop':
        if os.name == 'nt':
            # Прямые слэши Windows понимает, а в JSON они не требуют экранирования: человек,
            # вливающий блок в свой конфиг руками, не споткнётся об удвоенные «\\».
            exe, path = exe.replace('\\', '/'), path.replace('\\', '/')
        entry = {'command': exe, 'args': [path]}
        if env:
            entry['env'] = env
        # ensure_ascii: путь с кириллицей останется верным JSON в любой кодировке консоли
        print(json.dumps({'mcpServers': {SERVER_NAME: entry}}, indent=2))
        sys.stdout.flush()      # пояснение идёт в stderr; в общем потоке оно обязано быть ПОСЛЕ
        log('Это блок для claude_desktop_config.json (Settings → Developer → Edit Config). '
            'Если в файле уже есть "mcpServers" — добавьте в него запись "%s", остальные не '
            'трогайте; затем полностью перезапустите Claude Desktop.' % SERVER_NAME)
        return 0
    parts = ['claude', 'mcp', 'add', SERVER_NAME, '--scope', 'user']
    for key, val in env.items():
        parts += ['--env', '%s=%s' % (key, val)]
    print(' '.join([_shq(p) for p in parts + ['--']]
                   + [_shq(exe, path=True), _shq(path, path=True)]))
    # Ассистент читает stdout и stderr одним потоком, и велено выполнить «напечатанное»: без
    # сброса буфера русская строка пояснения оказывалась ПЕРЕД командой.
    sys.stdout.flush()
    log('Проверка после регистрации: claude mcp list — в строке %s должно быть Connected.'
        % SERVER_NAME)
    return 0


def _build_parser():
    p = argparse.ArgumentParser(
        prog='server.py', allow_abbrev=False,
        formatter_class=argparse.RawDescriptionHelpFormatter,
        description='MCP-сервер MARFOR %s.\nБез аргументов — режим MCP (stdio): так его запускает '
                    'клиент. С аргументом — служебная команда.' % SERVER_VERSION,
        epilog='Коды возврата --login-finish: 0 — токен сохранён, 2 — ещё не подтверждено '
               '(подождать и повторить), 1 — отказ или время вышло.\n'
               'Стенд задаёт MARFOR_BASE_URL (по умолчанию %s).\nИнструкция: %s/mcp'
               % (PUBLIC_BASE, PUBLIC_BASE))
    g = p.add_mutually_exclusive_group()
    g.add_argument('--login', action='store_true',
                   help='войти подтверждением в браузере (для человека в терминале)')
    g.add_argument('--login-start', action='store_true',
                   help='начать вход: напечатать ссылку и код и сразу выйти')
    g.add_argument('--login-finish', action='store_true',
                   help='завершить вход, начатый --login-start')
    g.add_argument('--set-token', action='store_true',
                   help='сохранить токен, созданный вручную на сайте (скрытый ввод; нужен '
                        'терминал, без терминала — только вместе с --stdin)')
    g.add_argument('--logout', action='store_true',
                   help='удалить сохранённый токен этого стенда')
    g.add_argument('--check', action='store_true',
                   help='диагностика: версия, стенд, откуда токен, ответ стенда')
    g.add_argument('--print-config', choices=('claude-code', 'claude-desktop'),
                   metavar='КЛИЕНТ',
                   help='готовая регистрация сервера: claude-code — команда, '
                        'claude-desktop — JSON')
    g.add_argument('--version', action='store_true', help='напечатать версию')
    p.add_argument('--wait', type=int, metavar='N',
                   help='для --login-finish: сколько секунд ждать подтверждения '
                        '(по умолчанию %d, не больше %d)' % (LOGIN_WAIT_DEFAULT, LOGIN_WAIT_MAX))
    p.add_argument('--stdin', action='store_true',
                   help='для --set-token: прочитать токен из стандартного ввода (конвейер, '
                        'файл) без приглашения — когда терминала нет')
    return p


def _run_cli(args):
    """Код возврата служебной команды; None — команды нет, работаем MCP-сервером."""
    if args.version:
        print('marfor-mcp %s' % SERVER_VERSION)
        return 0
    if args.logout:
        return _cli_logout()
    if args.check:
        return _cli_check()
    if not (args.login or args.login_start or args.login_finish or args.set_token
            or args.print_config):
        return None
    bad = _base_url_problem()
    if bad:
        raise _CliError(bad)
    if args.print_config:
        return _cli_print_config(args.print_config)
    if args.login:
        return _cli_login()
    if args.login_start:
        return _cli_login_start()
    if args.login_finish:
        wait = LOGIN_WAIT_DEFAULT if args.wait is None else args.wait
        return _login_finish(max(0, min(wait, LOGIN_WAIT_MAX)))
    return _cli_set_token(from_stdin=args.stdin)


# ──────────────────────────────── MCP поверх stdio ────────────────────────────

def _utf8_stdio():
    """UTF-8 на всех трёх потоках. На Windows поток по умолчанию в cp1251/cp1252, и
    tools/list падал на первой же «→» в описании: клиент не видел ни одного инструмента."""
    # errors: битый байт на входе не должен ронять цикл чтения, а одинокий суррогат в
    # названии из базы — запись ответа (внутри JSON-строки «\udXXX» остаётся верным JSON).
    for stream, extra in ((sys.stdin, {'errors': 'replace'}),
                          (sys.stdout, {'newline': '\n', 'errors': 'backslashreplace'}),
                          (sys.stderr, {'errors': 'backslashreplace'})):
        try:
            stream.reconfigure(encoding='utf-8', **extra)
        except Exception:
            pass        # поток подменён или закрыт — работаем с тем, что есть


def respond(msg_id, result=None, error=None):
    out = {'jsonrpc': '2.0', 'id': msg_id}
    if error is not None:
        out['error'] = error
    else:
        out['result'] = result
    sys.stdout.write(json.dumps(out, ensure_ascii=False) + '\n')
    sys.stdout.flush()


def main(argv=None):
    _utf8_stdio()
    parser = _build_parser()
    args = parser.parse_args(argv)          # незнакомый аргумент — ошибка argparse, код 2
    if args.wait is not None and not args.login_finish:
        parser.error('--wait имеет смысл только вместе с --login-finish')
    if args.stdin and not args.set_token:
        parser.error('--stdin имеет смысл только вместе с --set-token')
    try:
        rc = _run_cli(args)
    except (_CliError, RuntimeError) as e:
        print(str(e), file=sys.stderr)
        return 1
    except OSError as e:        # не удалось записать token.json / pending.json
        print('Не удалось записать файл в %s: %s' % (CONFIG_DIR, e), file=sys.stderr)
        return 1
    except KeyboardInterrupt:
        print('\nПрервано.', file=sys.stderr)
        return 130
    if rc is not None:
        return rc
    bad = _base_url_problem()
    if bad:
        log('[marfor] ' + bad)
        return 1
    serve()
    return 0


def serve():
    tok, src = _find_token()
    mode = (('токен из MARFOR_TOKEN' if src == 'env' else 'токен из файла') if tok
            else ('пароль, %s' % EMAIL if EMAIL else 'НЕ НАСТРОЕН — нужен --login'))
    log(f'[marfor] MCP-сервер {SERVER_VERSION} запущен, стенд {BASE_URL}, вход: {mode}, '
        f'наборы: {_toolsets_label()}' + (', только чтение' if _readonly() else ''))
    asked = {x.strip() for x in (os.environ.get('MARFOR_TOOLSETS') or '').lower()
             .replace(';', ',').split(',') if x.strip()}
    unknown = sorted(asked - set(TOOLSET_NAMES) - {'all'})
    if unknown:
        log('[marfor] MARFOR_TOOLSETS: незнакомые наборы пропущены: %s (есть: %s, all)'
            % (', '.join(unknown), ', '.join(TOOLSET_NAMES)))
    if not tok and EMAIL and not BASE_URL_EXPLICIT:
        log('[marfor] MARFOR_EMAIL задан, а MARFOR_BASE_URL — нет: парольный вход отключён, '
            'пароль на стенд по умолчанию (%s) не отправляется. Задайте MARFOR_BASE_URL '
            'или войдите токеном (--login).' % PUBLIC_BASE)
    try:
        if sys.stdin.isatty():
            log('[marfor] Сервер ждёт сообщений MCP-клиента на stdin — так и задумано. '
                'Служебные команды для человека: --help')
    except Exception:
        pass
    for line in sys.stdin:
        line = line.strip()
        if not line:
            continue
        try:
            msg = json.loads(line)
        except Exception:
            continue
        method, msg_id = msg.get('method'), msg.get('id')
        try:
            if method == 'initialize':
                want = (msg.get('params') or {}).get('protocolVersion')
                proto = want if want in SUPPORTED_PROTOCOLS else SUPPORTED_PROTOCOLS[0]
                respond(msg_id, {
                    'protocolVersion': proto,
                    'capabilities': {'tools': {}},
                    'serverInfo': {'name': SERVER_NAME, 'version': SERVER_VERSION},
                    'instructions': INSTRUCTIONS,
                })
            elif method == 'notifications/initialized':
                pass
            elif method == 'ping':
                respond(msg_id, {})
            elif method == 'tools/list':
                respond(msg_id, {'tools': visible_tools()})
            elif method == 'tools/call':
                p = msg.get('params') or {}
                try:
                    text = call_tool(p.get('name'), p.get('arguments') or {})
                    respond(msg_id, {'content': [{'type': 'text', 'text': str(text)}]})
                except Exception as e:
                    log(f'[marfor] ошибка инструмента {p.get("name")}: {e}')
                    respond(msg_id, {'content': [{'type': 'text', 'text': f'Ошибка: {e}'}],
                                     'isError': True})
            elif msg_id is not None:
                respond(msg_id, error={'code': -32601, 'message': f'нет метода {method}'})
        except Exception as e:
            log(f'[marfor] сбой обработки {method}: {e}')
            if msg_id is not None:
                respond(msg_id, error={'code': -32603, 'message': str(e)})


if __name__ == '__main__':
    sys.exit(main())