#!/usr/bin/env python3
"""eva-agent — клиент модуля wiki хаба (sdlc-hub) для виртуалки с доступом к Eva банка.

Опрашивает хаб, выполняет команды wiki_sync и wiki_fetch (только чтение Eva), ведёт журнал на диске. Только
стандартная библиотека, Python 3.9. Общие части (HTTP, папка, журнал, цикл, самообновление, запускатель) — копия
jira-agent v14. Спека: jira-hub/openspec/changes/wiki-module/.
Связь с хабом — HTTP (`hub_url = http://…`) или общая папка (`hub_url = folder:<путь>`), если сеть закрыта VPN.

    python3 agent.py check     # доступ к хабу и Eva, без записи
    python3 agent.py run       # основной цикл (по умолчанию); run --once — выполнить выданное и выйти (сквозной прогон)
    python3 agent.py selftest  # самопроверка новой версии перед обновлением: версия + check; код 0 или 3
    python3 agent.py prune     # почистить journal.jsonl от подтверждённых записей старше 30 дней

Самообновление — через запускатель run.py (спека agent, «Самообновление агента»).
"""
import argparse
import base64
import fcntl
import hashlib
import http.client
import json
import os
import re
import signal
import ssl
import subprocess
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
import uuid
from html.parser import HTMLParser
from datetime import datetime, timezone
from typing import Any, Callable, Dict, List, Optional, Tuple

AGENT_VERSION = 1
EXIT_LOCKED = 73              # «журнал занят другим процессом»: для run.py это не сбой версии (design 8)
EXIT_UPDATE = 75              # «обновление готово»: run.py подменит agent.py на agent.py.new
HERE = os.path.dirname(os.path.abspath(__file__))


def now_iso() -> str:
    return datetime.now(timezone.utc).replace(microsecond=0).isoformat()


def log(msg: str) -> None:
    print("[%s] %s" % (datetime.now().strftime("%H:%M:%S"), msg), flush=True)


def load_config(path: str) -> dict:
    try:
        with open(path, encoding="utf-8") as f:
            cfg = json.load(f)
    except FileNotFoundError:
        sys.exit("нет файла настроек %s — скопируйте agent.example.json в agent.json и заполните" % path)
    for key in ("hub_url", "hub_token", "eva"):
        if not cfg.get(key):
            sys.exit("в %s нет обязательного поля %s" % (path, key))
    if not isinstance(cfg["eva"], dict):
        sys.exit("в %s поле eva должно быть объектом {base_url, token}" % path)
    for key in ("base_url", "token"):
        if not cfg["eva"].get(key):
            sys.exit("в %s нет eva.%s" % (path, key))
    if cfg["hub_url"].startswith("folder:") and not os.path.isabs(cfg["hub_url"][len("folder:"):]):
        sys.exit("hub_url: после folder: нужен абсолютный путь общей папки, например "
                 "folder:/Volumes/My Shared Files/RSHB/jira-bridge")
    cfg.setdefault("timeout", 60)
    cfg.setdefault("poll_seconds", 5)
    cfg.setdefault("verify_ssl", True)
    return cfg


# ---------------------------------------------------------------------------
# Самообновление: update.json рядом с agent.py (спека agent «Самообновление агента», design 7)
# ---------------------------------------------------------------------------
def read_update_state(path: str) -> Optional[dict]:
    """Состояние обновления; нет файла или мусор — None."""
    try:
        with open(path, encoding="utf-8") as f:
            data = json.load(f)
    except FileNotFoundError:
        return None
    except (OSError, ValueError):
        log("update.json не читается — считаю, что состояния нет")
        return None
    return data if isinstance(data, dict) else None


def write_update_state(path: str, state: str, from_version, to_version, sha256, error=None) -> dict:
    data = {"state": state, "from_version": from_version, "to_version": to_version, "sha256": sha256,
            "error": error, "at": now_iso()}
    tmp = path + ".tmp"
    with open(tmp, "w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False)
    os.replace(tmp, path)
    return data


# ---------------------------------------------------------------------------
# HTTP
# ---------------------------------------------------------------------------
class HttpError(Exception):
    def __init__(self, code: int, text: str, headers=None):
        super().__init__("HTTP %d: %s" % (code, text[:300]))
        self.code, self.text, self.headers = code, text, headers or {}


class RedirectAway(Exception):
    """Сервер перенаправил запрос на другой адрес (схема, хост или порт): учётные данные туда не отправляются."""
    def __init__(self, host: str):
        super().__init__("перенаправление на %s" % host)
        self.host = host


MAX_ATTACHMENT = 10 * 1024 * 1024        # предел скачивания вложения, как у хранилища хаба (байт)


class TooBig(Exception):
    """Ответ длиннее предела: чтение оборвано."""


def _origin(url: str) -> Tuple[str, str, Optional[int]]:
    u = urllib.parse.urlsplit(url)
    scheme = (u.scheme or "").lower()
    return scheme, (u.hostname or "").lower(), u.port or {"http": 80, "https": 443}.get(scheme)


class _SameOriginRedirect(urllib.request.HTTPRedirectHandler):
    """Перенаправления — только в пределах адреса клиента: urllib переносит Authorization и на чужой хост."""
    def __init__(self, base: str):
        self.origin = _origin(base)

    def redirect_request(self, req, fp, code, msg, headers, newurl):
        if _origin(newurl) != self.origin:
            fp.close()
            raise RedirectAway(urllib.parse.urlsplit(newurl).hostname or newurl)
        return super().redirect_request(req, fp, code, msg, headers, newurl)


class Http:
    def __init__(self, base: str, headers: Dict[str, str], timeout: float = 60, ctx=None):
        self.base, self.headers, self.timeout, self.ctx = base.rstrip("/"), headers, timeout, ctx

    def __repr__(self):
        return "<%s %s>" % (self.__class__.__name__, self.base)

    def call(self, method: str, path: str, body=None, params=None, timeout=None) -> Any:
        data = json.dumps(body, ensure_ascii=False).encode("utf-8") if body is not None else None
        return self.send(method, path, data, "application/json; charset=utf-8" if data is not None else None,
                         params, timeout)

    def send(self, method: str, path: str, data: Optional[bytes], content_type: Optional[str], params=None,
             timeout=None) -> Any:
        """Запрос с готовым телом (JSON или multipart); ответ — JSON. Ошибки — HttpError и ConnectionError."""
        url = self.base + path
        if params:
            url += "?" + urllib.parse.urlencode({k: v for k, v in params.items() if v is not None})
        req = urllib.request.Request(url, data=data, method=method)
        for k, v in self.headers.items():
            req.add_header(k, v)
        req.add_header("Accept", "application/json")
        if content_type is not None:
            req.add_header("Content-Type", content_type)
        try:
            with self._opener().open(req, timeout=timeout or self.timeout) as resp:
                raw = resp.read()
                status = resp.status
        except RedirectAway as e:
            self._touch()
            raise HttpError(302, "перенаправление на %s отклонено: учётные данные туда не отправляются" % e.host) from None
        except urllib.error.HTTPError as e:
            self._touch()
            try:
                headers = {k.lower(): v for k, v in e.headers.items()}
                text = e.read().decode("utf-8", "replace")
            except (OSError, ConnectionError, TimeoutError, http.client.HTTPException) as err:
                raise ConnectionError("%s: обрыв при чтении ответа: %s" % (self.base, err)) from None
            finally:
                e.close()
            raise HttpError(e.code, text, headers) from None
        except (urllib.error.URLError, ConnectionError, TimeoutError, OSError, http.client.HTTPException) as e:
            raise ConnectionError("%s недоступен: %s" % (self.base, e)) from None
        self._touch()
        try:
            return json.loads(raw.decode("utf-8")) if raw else {}
        except ValueError as e:   # включая UnicodeDecodeError и JSONDecodeError
            raise HttpError(status, "ответ не JSON: %s; начало: %s" % (e, raw[:200].decode("utf-8", "replace"))) from None

    def _opener(self):
        """Открыватель запросов: перенаправления — только на адрес клиента (urllib переносит Authorization и на
        чужой хост); контекст TLS — из настроек."""
        handlers = [_SameOriginRedirect(self.base)]       # type: List[Any]
        if self.ctx is not None:
            handlers.append(urllib.request.HTTPSHandler(context=self.ctx))
        return urllib.request.build_opener(*handlers)

    def fetch_bytes(self, path: str, limit: int, timeout=None) -> bytes:
        """GET с ответом-байтами (содержимое вложения): без разбора JSON, читает не больше limit + 1 байт — длиннее
        limit — TooBig. Перенаправление на другой адрес — RedirectAway без перехода. Ошибки — как у send."""
        req = urllib.request.Request(self.base + path, method="GET")
        for k, v in self.headers.items():
            req.add_header(k, v)
        req.add_header("Accept", "*/*")
        try:
            with self._opener().open(req, timeout=timeout or self.timeout) as resp:
                if resp.length is not None and resp.length > limit:
                    raise TooBig("ответ длиннее %d байт (Content-Length %d)" % (limit, resp.length))   # тело не читаем
                data = resp.read(limit + 1)
                rest = resp.length                 # остаток по Content-Length: read(n) при обрыве молча отдаёт меньше
            if rest and len(data) <= limit:
                raise ConnectionError("%s: обрыв при чтении ответа: получено %d байт, не дошло %d" % (self.base, len(data), rest))
        except (RedirectAway, TooBig):
            self._touch()
            raise
        except urllib.error.HTTPError as e:
            self._touch()
            try:
                headers = {k.lower(): v for k, v in e.headers.items()}
                text = e.read().decode("utf-8", "replace")
            except (OSError, ConnectionError, TimeoutError, http.client.HTTPException) as err:
                raise ConnectionError("%s: обрыв при чтении ответа: %s" % (self.base, err)) from None
            finally:
                e.close()
            raise HttpError(e.code, text, headers) from None
        except (urllib.error.URLError, ConnectionError, TimeoutError, OSError, http.client.HTTPException) as e:
            raise ConnectionError("%s недоступен: %s" % (self.base, e)) from None
        self._touch()
        if len(data) > limit:
            raise TooBig("ответ длиннее %d байт" % limit)
        return data

    def _touch(self) -> None:
        """Вызывается на каждый полученный HTTP-ответ (любой код)."""


class EvaError(Exception):
    """JSON-RPC Eva вернула `error`: текст «<метод>: <error>»."""


class ForeignHost(Exception):
    """Адрес вложения не на хосте Eva (или не разбирается): токен туда не отправляется, запроса нет."""


_URL_SAFE = "/%:@!$&'()*+,;=-._~"
_STRAY_PERCENT = re.compile(r"%(?![0-9A-Fa-f]{2})")


def _quote_url_part(part: str, extra_safe: str = "") -> str:
    """Процентная кодировка пути/запроса для HTTP: пробелы и не-ASCII кодируются, готовые %XX остаются как есть,
    одиночный `%` без двух шестнадцатеричных цифр становится %25 (I-2)."""
    return urllib.parse.quote(_STRAY_PERCENT.sub("%25", part), safe=_URL_SAFE + extra_safe)


class EvaClient(Http):
    """Клиент Eva (JSON-RPC 2.2): повторы, пауза между вызовами, счётчики вызовов и 429, время последнего вызова."""
    PAGE_SIZE = 200
    RETRY_CODES = (429, 502, 503, 504)
    ATTEMPTS = 4
    MAX_RETRY_AFTER = 60

    def __init__(self, cfg: dict):
        e = cfg["eva"]
        ctx = None
        if not cfg.get("verify_ssl", True):
            ctx = ssl.create_default_context()
            ctx.check_hostname = False
            ctx.verify_mode = ssl.CERT_NONE
        elif cfg.get("ca_bundle"):
            ctx = ssl.create_default_context(cafile=cfg["ca_bundle"])
        super().__init__(e["base_url"], {"Authorization": "Bearer " + e["token"]}, cfg.get("timeout", 60), ctx)
        self._token = e["token"]
        self.calls = 0
        self.rate_limited = 0
        self.pacing = 0.0            # с между вызовами; ставит Runner из wiki_pacing
        self.last_call_at = None     # type: Optional[float]  (time.monotonic)
        self.last_call_iso = None    # type: Optional[str]
        self.sleep = time.sleep

    def _touch(self) -> None:
        self.calls += 1
        self.last_call_at, self.last_call_iso = time.monotonic(), now_iso()

    def reset_counters(self) -> None:
        self.calls, self.rate_limited = 0, 0

    def _scrub(self, text: str) -> str:
        return text.replace(self._token, "***") if self._token else text

    def _retrying(self, fn: Callable[[], Any]) -> Any:
        """Вызов с паузой между вызовами и повторами: 429/502/503/504 и сетевые сбои, 4 попытки, ожидание 1, 2, 4 с
        (при 429 — Retry-After, не больше 60 с). Тексты ошибок очищаются от токена."""
        for attempt in range(self.ATTEMPTS):
            if self.pacing > 0 and self.last_call_at is not None:
                wait = self.pacing - (time.monotonic() - self.last_call_at)
                if wait > 0:
                    self.sleep(wait)
            wait = float(2 ** attempt)
            try:
                return fn()
            except HttpError as e:
                if e.code not in self.RETRY_CODES:
                    raise HttpError(e.code, self._scrub(e.text), e.headers) from None
                if e.code == 429:
                    self.rate_limited += 1
                    try:
                        wait = min(max(float(e.headers.get("retry-after", wait)), 0.0), self.MAX_RETRY_AFTER)
                    except ValueError:
                        pass
                if attempt == self.ATTEMPTS - 1:
                    raise HttpError(e.code, self._scrub(e.text), e.headers) from None
            except ConnectionError as e:
                if attempt == self.ATTEMPTS - 1:
                    raise ConnectionError(self._scrub(str(e))) from None
            self.sleep(wait)
        raise AssertionError("недостижимо")

    def call(self, method: str, **kwargs) -> Any:     # type: ignore[override]
        """Метод JSON-RPC: POST <base>/api/?m=<метод>. `error` в ответе — EvaError."""
        body = {"jsonrpc": "2.2", "method": method, "callid": str(uuid.uuid4()), "kwargs": kwargs}
        data = self._retrying(lambda: Http.call(self, "POST", "/api/", body, {"m": method}))
        if isinstance(data, dict) and data.get("error"):
            raise EvaError(self._scrub("%s: %s" % (method, data["error"])))
        return data.get("result") if isinstance(data, dict) else None

    def get_doc(self, code: str, fields: List[str]) -> Optional[dict]:
        return self.call("CmfDocument.get", filter=[["code", "==", code]], fields=fields)

    def list_docs(self, flt: list, fields: List[str]) -> List[dict]:
        """Весь список страницами по PAGE_SIZE. Если Eva не соблюдает slice (ответ без новых записей), обход
        останавливается на собранном, а не крутится бесконечно (M-4)."""
        out, offset = [], 0     # type: List[dict], int
        seen = set()            # type: set
        while True:
            chunk = self.call("CmfDocument.list", filter=flt, fields=fields,
                              slice=[offset, offset + self.PAGE_SIZE]) or []
            if isinstance(chunk, dict):                   # обёрнутый результат
                chunk = chunk.get("items") or chunk.get("result") or []
            fresh = 0
            for d in chunk:
                key = json.dumps(d, sort_keys=True, ensure_ascii=False)
                if key not in seen:
                    seen.add(key)
                    out.append(d)
                    fresh += 1
            if len(chunk) < self.PAGE_SIZE:
                return out
            if not fresh:
                log("  Eva: CmfDocument.list повторяет записи (slice не соблюдается?) — останов на %d" % len(out))
                return out
            offset += self.PAGE_SIZE

    def download(self, url: str, limit: int) -> bytes:
        """Файл Eva потоком: читается не больше limit + 1 байта, длиннее limit — TooBig. Адрес — только своего хоста
        (иначе ForeignHost); путь и запрос перед отправкой кодируются (адрес из HTML бывает с пробелами и кириллицей)."""
        try:
            same = _origin(url) == _origin(self.base)
        except ValueError:                         # `http://[::1`, порт не числом
            raise ForeignHost("некорректный адрес") from None
        rest = url[len(self.base):] if url.startswith(self.base) else None
        if not same or rest is None or rest[:1] not in ("", "/", "?"):
            raise ForeignHost("чужой хост")
        path, _, query = rest.split("#", 1)[0].partition("?")
        path = _quote_url_part(path) + ("?" + _quote_url_part(query, "?") if query else "")
        return self._retrying(lambda: self.fetch_bytes(path, limit))


class _ParseError(Exception):
    """Разбор HTML споткнулся (ParserBase.error в 3.9): см. _LinkParser.error."""


class _LinkParser(HTMLParser):
    """Собирает img[src] и a[href] в порядке появления: (тег, значение атрибута после unescape, alt/title, текст ссылки).

    На 3.9 `_markupbase` зовёт error() на `<![foo]>` и подобном, а базовый error() бросает NotImplementedError (I-1).
    error() бросает _ParseError, а parse_marked_section пропускает такую конструкцию до `>`, как мусорный комментарий."""

    def __init__(self):
        HTMLParser.__init__(self, convert_charrefs=True)
        self.items = []         # [tag, ref, hint, text]
        self._open = None       # запись открытой <a>, в неё копится текст

    def error(self, message):
        raise _ParseError(message)

    def parse_marked_section(self, i, report=1):
        try:
            return HTMLParser.parse_marked_section(self, i, report)
        except _ParseError:
            j = self.rawdata.find(">", i + 3)
            return -1 if j < 0 else j + 1          # -1: ждать данных; при close() goahead сам пропустит до `>`

    def handle_starttag(self, tag, attrs):
        a = dict(attrs)
        if tag == "img":
            ref = a.get("src")
            if ref:
                self.items.append(["img", ref, a.get("alt") or a.get("title") or "", ""])
        elif tag == "a":
            ref = a.get("href")
            self._open = None
            if ref:
                self._open = ["a", ref, a.get("title") or "", ""]
                self.items.append(self._open)

    def handle_startendtag(self, tag, attrs):
        if tag == "img":
            self.handle_starttag(tag, attrs)

    def handle_endtag(self, tag):
        if tag == "a":
            self._open = None

    def handle_data(self, data):
        if self._open is not None:
            self._open[3] += data


_PAGE_CODE = re.compile(r"DOC-\d+", re.IGNORECASE)
_THUMB_TAIL = ".meta/thumbnail.jpg"


def classify_links(html: str, base_url: str) -> Tuple[list, list]:
    """Вложения и ссылки на страницы из HTML страницы Eva (design 6–7). Порядок появления, повторы не склеиваются.

    ref — значение атрибута после unescape (design 12). Чужой хост и static/ не передаются."""
    parser = _LinkParser()
    try:
        parser.feed(html or "")
        parser.close()
    except (_ParseError, NotImplementedError, AssertionError) as e:   # запасной путь: собранное до сбоя сохраняется
        log("  разбор HTML прерван (%s: %s), ссылок до сбоя: %d" % (type(e).__name__, e, len(parser.items)))
    base = urllib.parse.urlsplit(base_url)
    attachments, links = [], []
    for _tag, ref, hint, text in parser.items:
        try:
            parts = urllib.parse.urlsplit(urllib.parse.urljoin(base_url.rstrip("/") + "/", ref.strip()))
        except ValueError:                 # `http://[::1` — ссылка битая, пропускаем (I-1)
            continue
        if (parts.scheme, parts.netloc) != (base.scheme, base.netloc):
            continue
        path = parts.path
        if path.startswith("/files/"):
            if path.endswith(_THUMB_TAIL):
                path = path[:-len(_THUMB_TAIL)]
            name = " ".join(hint.split()) or " ".join(text.split()) or urllib.parse.unquote(path.rsplit("/", 1)[-1])
            url = urllib.parse.urlunsplit((parts.scheme, parts.netloc, path, "", ""))
            attachments.append({"ref": ref, "url": url, "name": name})
            continue
        m = _PAGE_CODE.search(path)
        if m:
            links.append({"ref": ref, "code": m.group(0).upper()})
    return attachments, links


class FolderTransport:
    """Обмен с хабом через общую папку (hub_url = folder:<путь>), когда сеть до хаба закрыта (VPN на виртуалке).
    Запрос — файл req/<id>.json, ответ кладёт реле хаба в resp/<id>.json. Ошибки те же, что у Http.call:
    нет ответа к сроку или нет папки — ConnectionError, код не 2xx — HttpError. Спека: specs/agent, folder-transport."""
    POLL = 0.1
    BAD_JSON_WAIT = 2.0

    def __init__(self, root: str, authorization: str):
        self.root, self.authorization = root, authorization

    def __repr__(self):
        return "<FolderTransport %s>" % self.root

    def call(self, method: str, path: str, body=None, timeout: float = 60) -> Any:
        if not os.path.isdir(self.root):
            raise ConnectionError("нет папки %s" % self.root)
        req_dir, resp_dir = os.path.join(self.root, "req"), os.path.join(self.root, "resp")
        rid = uuid.uuid4().hex
        try:
            os.makedirs(req_dir, exist_ok=True)
            os.makedirs(resp_dir, exist_ok=True)
            deadline = time.time() + timeout
            self._write(req_dir, rid + ".json", {
                "id": rid, "method": method, "path": path, "body": body, "authorization": self.authorization,
                "deadline": datetime.fromtimestamp(deadline, timezone.utc).isoformat()})
        except OSError as e:
            raise ConnectionError("папка %s недоступна: %s" % (self.root, e)) from None
        resp_path = os.path.join(resp_dir, rid + ".json")
        end = time.monotonic() + timeout
        while time.monotonic() < end:
            if os.path.exists(resp_path):
                return self._take(resp_path)
            time.sleep(self.POLL)
        try:
            os.remove(os.path.join(req_dir, rid + ".json"))
        except OSError:
            pass
        raise ConnectionError("не ответил через папку %s за %.0f с" % (self.root, timeout))

    @staticmethod
    def _write(directory: str, name: str, payload: dict) -> None:
        tmp = os.path.join(directory, "." + name + ".tmp")
        with open(tmp, "w", encoding="utf-8") as f:
            json.dump(payload, f, ensure_ascii=False)
            f.flush()
            os.fsync(f.fileno())
        os.replace(tmp, os.path.join(directory, name))

    def _take(self, path: str) -> Any:
        end = time.monotonic() + self.BAD_JSON_WAIT
        while True:
            try:
                with open(path, encoding="utf-8") as f:
                    resp = json.load(f)
                if not isinstance(resp, dict):
                    raise ValueError("ответ не объект")
                status, body = resp.get("status"), resp.get("body")
                if type(status) is not int or not (body is None or isinstance(body, str)):
                    raise ValueError("неверные типы status/body")   # bool — подкласс int, поэтому type(), не isinstance (M-2 agent#8)
                break
            except (ValueError, OSError):
                if time.monotonic() >= end:
                    raise ConnectionError("нечитаемый ответ хаба в %s" % path) from None
                time.sleep(self.POLL)
        try:
            os.remove(path)
        except OSError:
            pass
        text = body or ""                      # status и body проверены выше (M-5 #18)
        if not 200 <= status < 300:
            raise HttpError(status, text)
        try:
            return json.loads(text) if text else {}
        except ValueError as e:
            raise HttpError(status, "ответ не JSON: %s; начало: %s" % (e, text[:200])) from None


class HubClient(Http):
    def __init__(self, cfg: dict):
        auth = "Bearer " + cfg["hub_token"]
        super().__init__(cfg["hub_url"], {"Authorization": auth}, cfg.get("timeout", 60))
        url = cfg["hub_url"]
        self.folder = FolderTransport(url[len("folder:"):], auth) if url.startswith("folder:") else None

    def call(self, method: str, path: str, body=None, params=None, timeout=None) -> Any:
        if self.folder is None:
            return super().call(method, path, body, params, timeout)
        if params:
            path += "?" + urllib.parse.urlencode({k: v for k, v in params.items() if v is not None})
        return self.folder.call(method, path, body, timeout or self.timeout)

    def claim(self, agent_version, settings_version, last_call_at, update_state=None, launcher_version=None) -> dict:
        return self.call("POST", "/agent/claim", {"agent_version": agent_version, "settings_version": settings_version,
                                                  "last_call_at": last_call_at, "update_state": update_state,
                                                  "launcher_version": launcher_version}, timeout=max(self.timeout, 45))

    def release(self) -> dict:
        return self.call("GET", "/agent/release")

    def settings(self) -> dict:
        return self.call("GET", "/agent/settings")

    def command(self, command_id) -> dict:
        return self.call("GET", "/agent/commands/%d" % int(command_id))

    def result(self, command_id, payload: dict) -> dict:
        return self.call("POST", "/agent/commands/%d/result" % int(command_id), payload)

    def upload_file(self, data: bytes) -> dict:
        """Файл, скачанный из Eva, — в хранилище хаба: {file (sha256), size}."""
        return self.call("POST", "/agent/files", {"content": base64.b64encode(data).decode("ascii")})

    def wiki_pages(self, command_id, pages: list) -> dict:
        """Пачка страниц wiki_sync в стадию хаба (до 50 страниц, тело до 16 МБ)."""
        return self.call("POST", "/agent/wiki/pages", {"command_id": command_id, "pages": pages})


# ---------------------------------------------------------------------------
# Журнал: только дозапись; started → finished → acked по id команды
# ---------------------------------------------------------------------------
class Journal:
    def __init__(self, path: str):
        self.path = path

    def _torn_tail(self) -> bool:
        """Файл не пуст и не заканчивается переводом строки (оборванная запись)."""
        try:
            if os.path.getsize(self.path) == 0:
                return False
        except FileNotFoundError:
            return False
        with open(self.path, "rb") as f:
            f.seek(-1, os.SEEK_END)
            return f.read(1) != b"\n"

    def append(self, record: dict) -> None:
        record = dict(record)
        record.setdefault("at", now_iso())
        data = (json.dumps(record, ensure_ascii=False) + "\n").encode("utf-8")
        if self._torn_tail():
            data = b"\n" + data
        fd = os.open(self.path, os.O_WRONLY | os.O_APPEND | os.O_CREAT, 0o600)
        with os.fdopen(fd, "ab") as f:
            f.write(data)
            f.flush()
            os.fsync(f.fileno())

    def _scan(self):
        """Возвращает (годные записи, номера битых строк). Годная — dict с command_id."""
        good, bad = [], []
        if not os.path.exists(self.path):
            return good, bad
        with open(self.path, "rb") as f:
            for n, raw in enumerate(f, 1):
                line = raw.strip()
                if not line:
                    continue
                try:
                    r = json.loads(line.decode("utf-8"))
                except ValueError:
                    r = None
                if isinstance(r, dict) and "command_id" in r:
                    good.append(r)
                else:
                    bad.append(n)
        return good, bad

    def records(self) -> List[dict]:
        good, bad = self._scan()
        for n in bad:
            log("журнал: строка %d не читается, пропускаю" % n)
        return good

    def last_at(self) -> Optional[datetime]:
        """Самое позднее время записи журнала (UTC). Записи без времени или с непонятным временем пропускаются."""
        latest = None
        for r in self.records():
            try:
                t = datetime.fromisoformat(r["at"])
            except (KeyError, TypeError, ValueError):
                continue
            if t.tzinfo is None:
                continue
            if latest is None or t > latest:
                latest = t
        return latest

    def claimed(self, command_id) -> None:
        """Команда получена из claim (до паузы): с этого момента хаб считает её выданной агенту."""
        self.append({"kind": "claimed", "command_id": int(command_id)})

    def started(self, command_id) -> None:
        self.append({"kind": "started", "command_id": int(command_id)})

    def finished(self, command_id, status, result, error, jira_calls, rate_limited) -> None:
        self.append({"kind": "finished", "command_id": int(command_id), "status": status, "result": result,
                     "error": error, "jira_calls": int(jira_calls or 0), "rate_limited": int(rate_limited or 0)})

    def acked(self, command_id) -> None:
        self.append({"kind": "acked", "command_id": int(command_id)})

    def unacked_finished(self) -> List[dict]:
        last, acked = {}, set()
        for r in self.records():
            if r.get("kind") == "finished":
                last[r["command_id"]] = r
                acked.discard(r["command_id"])
            elif r.get("kind") == "acked":
                acked.add(r["command_id"])
        return [r for cid, r in last.items() if cid not in acked]

    def unfinished_started(self) -> List[int]:
        """Порядковый учёт: claimed/started открывают id, finished/acked закрывает; порядок — первое открытие."""
        opened = {}  # command_id -> открыта ли сейчас; dict сохраняет порядок первого открытия
        for r in self.records():
            kind, cid = r.get("kind"), r["command_id"]
            if kind in ("claimed", "started"):
                opened[cid] = True
            elif kind in ("finished", "acked") and cid in opened:
                opened[cid] = False
        return [cid for cid, is_open in opened.items() if is_open]

    def prune(self, days: int = 30) -> int:
        """Удалить записи id до его последнего acked, если тот старше days дней (возраст — по acked)
        и после него у id нет started/finished. Возвращает число удалённых годных записей; битые строки
        отбрасываются при перезаписи и учитываются в логе."""
        good, bad = self._scan()
        if bad:
            log("журнал: удаляю битых строк %d" % len(bad))
        cutoff = datetime.now(timezone.utc).timestamp() - days * 86400
        last_ack, last_open = {}, {}
        for i, r in enumerate(good):
            cid = r["command_id"]
            if r.get("kind") == "acked":
                last_ack[cid] = i
            elif r.get("kind") in ("claimed", "started", "finished"):
                last_open[cid] = i
        drop_upto = {}  # command_id -> индекс последнего acked: всё до него включительно удаляется
        for cid, i in last_ack.items():
            if last_open.get(cid, -1) > i:
                continue
            try:
                ts = _parse_at(good[i].get("at"))
            except (ValueError, TypeError):
                continue
            if ts < cutoff:
                drop_upto[cid] = i
        keep = [r for i, r in enumerate(good)
                if not (r["command_id"] in drop_upto and i <= drop_upto[r["command_id"]])]
        tmp = self.path + ".tmp"
        try:
            os.remove(tmp)
        except FileNotFoundError:
            pass
        fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
        with os.fdopen(fd, "wb") as f:
            for r in keep:
                f.write((json.dumps(r, ensure_ascii=False) + "\n").encode("utf-8"))
            f.flush()
            os.fsync(f.fileno())
        os.replace(tmp, self.path)
        return len(good) - len(keep)


def _parse_at(value) -> float:
    """Метка времени журнала в эпоху; наивное время трактуется как UTC."""
    dt = datetime.fromisoformat(value)
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    return dt.timestamp()


# ---------------------------------------------------------------------------
# Дерево страниц Eva
# ---------------------------------------------------------------------------
MAX_PAGES = 50                       # страниц в пачке (предел хаба, stage.MAX_PAGES)
MAX_BATCH_BYTES = 4 * 1024 * 1024    # JSON пачки; предел тела хаба — 16 МБ
TREE_FIELDS = ["id", "code", "name", "parent", "tree_parent", "project", "orderno"]


def _ref_code(ref) -> Optional[str]:
    """Код из ссылки Eva с полем code; страница без родителя — None."""
    code = ref.get("code") if isinstance(ref, dict) else None
    return code if isinstance(code, str) and code else None


def _ref_id(ref) -> Optional[str]:
    """Ссылочные поля Eva приходят то объектом {id, code, ...}, то строкой-id (как в eva_wiki_export.ref_id)."""
    if not ref:
        return None
    if isinstance(ref, dict):
        v = ref.get("id")
        return str(v) if v not in (None, "") else None
    return str(ref)


def _orderno(value) -> Optional[int]:
    if isinstance(value, bool):
        return None
    if isinstance(value, float) and value.is_integer():
        return int(value)
    return value if isinstance(value, int) else None


def build_children(eva: "EvaClient", root: dict) -> Dict[str, List[dict]]:
    """{код родителя: [дети]} для поддерева корня. С проектом — один список документов проекта и связи parent/tree_parent
    (tree_parent в Eva бывает не заполнен); без проекта — обход в ширину через фильтр tree_parent.code."""
    children = {}    # type: Dict[str, List[dict]]
    project = _ref_code(root.get("project"))
    if project:
        docs = eva.list_docs([["project.code", "==", project]], TREE_FIELDS)
        id_to_code = {str(d["id"]): d["code"] for d in docs + [root] if d.get("id") is not None and d.get("code")}

        def parent_code(ref) -> Optional[str]:
            # dict с code — код как есть; иначе (dict только с id, голая строка-id) — через карту id → код
            return _ref_code(ref) or id_to_code.get(_ref_id(ref) or "")

        for d in docs:
            for pc in {parent_code(d.get("parent")), parent_code(d.get("tree_parent"))} - {None}:
                children.setdefault(pc, []).append(d)
    else:
        queue, seen = [root], {root.get("code")}
        while queue:
            node = queue.pop(0)
            kids = eva.list_docs([["tree_parent.code", "==", node["code"]]], TREE_FIELDS)
            children[node["code"]] = kids
            for k in kids:
                if k.get("code") not in seen:
                    seen.add(k.get("code"))
                    queue.append(k)
    for pc, kids in children.items():
        uniq = {k["code"]: k for k in kids if k.get("code")}
        children[pc] = sorted(uniq.values(), key=lambda k: (_orderno(k.get("orderno")) is None,
                                                            _orderno(k.get("orderno")) or 0,
                                                            (k.get("name") or "").lower(), k["code"]))
    return children


def walk(root: dict, children: Dict[str, List[dict]]):
    """Обход в глубину, корень первым: (документ, код родителя по обходу). Каждый код — один раз (циклы, два родителя)."""
    seen = set()    # type: set
    stack = [(root, None)]
    while stack:
        doc, parent = stack.pop()
        code = doc["code"]
        if code in seen:
            continue
        seen.add(code)
        yield doc, parent
        for kid in reversed(children.get(code, [])):
            if kid["code"] not in seen:
                stack.append((kid, code))


# ---------------------------------------------------------------------------
# Исполнитель: команда wiki_sync / wiki_fetch (читает Eva, отдаёт хабу)
# ---------------------------------------------------------------------------
class HubError(Exception):
    """Ошибка хаба внутри команды (стадия страниц, загрузка файла): источник — хаб, не Eva (из jira-agent, M-6 #18).
    Повтор — только при недоступности хаба и 5xx; команды wiki Eva только читают, повтор безопасен."""

    def __init__(self, cause: Exception):
        super().__init__("хаб: %s" % cause)
        self.retryable = isinstance(cause, ConnectionError) or (isinstance(cause, HttpError) and cause.code >= 500)


class Executor:
    def __init__(self, eva: EvaClient, hub: Optional[HubClient]):
        self.eva, self.hub = eva, hub

    def run(self, cmd: dict) -> dict:
        self.eva.reset_counters()
        handler = getattr(self, "do_" + str(cmd.get("type")), None)
        if handler is None:
            return self._out("failed", error="неизвестный тип команды %s" % cmd.get("type"))
        try:
            return handler(cmd.get("params") or {}, cmd)
        except HubError as e:
            out = self._out("failed", error=str(e))
            out["retryable"] = e.retryable
            out["peer"] = "хабом"
            return out
        except EvaError as e:                      # Eva ответила `error` (нет прав, неверный фильтр): повтор не поможет (M-1)
            return self._out("failed", error="Eva: %s" % e)     # текст уже очищен от токена (решение 23)
        except HttpError as e:
            out = self._out("failed", error="HTTP %d %s" % (e.code, e.text[:1500]))
            out["retryable"] = e.code >= 500       # исход неизвестен: Runner повторит
            return out
        except ConnectionError as e:
            out = self._out("failed", error=str(e))
            out["retryable"] = True
            return out

    def _out(self, status, result=None, error=None) -> dict:
        return {"status": status, "result": result, "error": error,
                "jira_calls": self.eva.calls, "rate_limited": self.eva.rate_limited}   # имена полей — протокол хаба

    def _hub(self, fn, *args):
        """Вызов хаба из команды: его сбой — HubError, чтобы не выдать его за обрыв связи с Eva."""
        try:
            return fn(*args)
        except (ConnectionError, HttpError) as e:
            raise HubError(e) from None

    def _send_pages(self, command_id, batch: list) -> None:
        """Пачка страниц в стадию хаба. 413 (тело больше предела хаба) — ошибка с кодом страницы или границами пачки (M-3)."""
        try:
            self.hub.wiki_pages(command_id, batch)
        except HttpError as e:
            if e.code == 413:
                what = ("страница %s" % batch[0]["code"] if len(batch) == 1 else
                        "пачка страниц %s…%s (%d шт.)" % (batch[0]["code"], batch[-1]["code"], len(batch)))
                e = HttpError(413, "%s больше 16 МБ — хаб не принял: %s" % (what, e.text[:300]), e.headers)
            raise HubError(e) from None
        except ConnectionError as e:
            raise HubError(e) from None

    def do_wiki_sync(self, p: dict, cmd: dict) -> dict:
        """Дерево страниц Eva от корня → хаб пачками; отчёт — коды в порядке обхода (корень первым). Eva только читается,
        поэтому после перезапуска команда выполняется заново целиком (design 9)."""
        root_code = str(p.get("root") or "").strip()
        if not root_code:
            return self._out("failed", error="wiki_sync: не задан root")
        root = self.eva.get_doc(root_code, TREE_FIELDS + ["text"])
        if not root:
            return self._out("failed", error="страница %s не найдена или нет прав" % root_code)
        children = build_children(self.eva, root)
        root_code = root.get("code") or root_code
        base = self.eva.base
        codes = []      # type: List[str]
        batch = []      # type: list
        size = 0
        for doc, parent_code in walk(root, children):
            code = doc["code"]
            full = root if code == root_code else self.eva.get_doc(code, ["code", "name", "text"])
            if not full:
                return self._out("failed", error="страница %s не найдена или нет прав" % code)
            html = full.get("text") if isinstance(full.get("text"), str) else ""
            attachments, links = classify_links(html, base)
            page = {"code": code, "eva_id": str(doc["id"]) if doc.get("id") is not None else None,
                    # хаб отклоняет пустой title (stage._check): без названия страница называется своим кодом
                    "title": (full.get("name") or doc.get("name") or "").strip() or code,
                    "parent_code": parent_code, "orderno": _orderno(doc.get("orderno")),
                    "url": urllib.parse.urljoin(base + "/", "project/Document/" + code),
                    "html": html, "attachments": attachments, "links": links}
            psize = len(json.dumps(page, ensure_ascii=False).encode("utf-8")) + 2
            if batch and (len(batch) >= MAX_PAGES or size + psize > MAX_BATCH_BYTES):
                self._send_pages(cmd["id"], batch)
                batch, size = [], 0
            batch.append(page)
            size += psize
            codes.append(code)
        if batch:
            self._send_pages(cmd["id"], batch)
        title = (root.get("name") or "").strip() or root_code
        return self._out("done", {"root_title": title, "codes": codes, "eva_calls": self.eva.calls})

    def do_wiki_fetch(self, p: dict, cmd: dict) -> dict:
        """Файл вложения Eva → хабу (POST /agent/files). Eva только читается; адрес — лишь своего хоста (design 8, 24);
        содержимое в лог не попадает, в памяти одна копия (плюс base64 на время отправки)."""
        url = str(p.get("url") or "").strip()
        if not url:
            return self._out("failed", error="wiki_fetch: не задан url")
        try:
            data = self.eva.download(url, MAX_ATTACHMENT)
        except ForeignHost as e:
            return self._out("failed", error=str(e))
        except TooBig:
            return self._out("failed", error="файл больше 10 МБ — откройте в Eva")
        except HttpError as e:
            if e.code == 404:
                return self._out("failed", error="файл не найден в Eva")
            raise
        except RedirectAway as e:
            return self._out("failed", error="Eva перенаправила скачивание на %s" % e.host)
        log("  вложение #%s (%d байт) <- Eva" % (p.get("attachment_id"), len(data)))
        got = self._hub(self.hub.upload_file, data)
        del data
        return self._out("done", {"attachment_id": p.get("attachment_id"), "file": got.get("file"), "size": got.get("size")})


# ---------------------------------------------------------------------------
# Цикл: claim → пауза → журнал → выполнение → отчёт до подтверждения; восстановление при старте
# ---------------------------------------------------------------------------
class Updater:
    """Самообновление агента (спека agent «Самообновление агента», design 7–12). Работает только под run.py."""
    RETRY_SECONDS = 300       # временный сбой подготовки — повтор не раньше (R-8)
    HINT_SECONDS = 3600       # подсказка «запускайте через run.py» — не чаще (R-10)
    SELFTEST_TIMEOUT = 120
    VERSION_RE = re.compile(r"^AGENT_VERSION = (\d+)[ \t]*\r?$", re.M)
    SHA_RE = re.compile(r"[0-9a-f]{64}")

    def __init__(self, hub, base_dir: str, launcher_version: Optional[int], enabled: bool = True,
                 config_path: Optional[str] = None):
        self.hub, self.dir, self.launcher_version, self.enabled = hub, base_dir, launcher_version, enabled
        self.config_path = config_path or os.path.join(base_dir, "agent.json")
        self.state_path = os.path.join(base_dir, "update.json")
        self.new_path = os.path.join(base_dir, "agent.py.new")
        self.agent_path = os.path.join(base_dir, "agent.py")
        self.retry_after = {}         # (version, sha256) -> clock(), раньше которого не повторять
        self.hints = {}               # вид подсказки -> clock() последней записи в лог (не чаще HINT_SECONDS)
        self.clock = time.monotonic
        self.should_stop = lambda: False
        self.run_selftest = self._run_selftest

    def state(self) -> Optional[dict]:
        return read_update_state(self.state_path)

    def confirm(self) -> None:
        """Первый успешный claim новой версии: trial этой версии → ok (запускатель перестаёт ждать отката)."""
        st = self.state()
        if st and st.get("state") == "trial" and st.get("to_version") == AGENT_VERSION:
            try:
                write_update_state(self.state_path, "ok", st.get("from_version"), AGENT_VERSION, st.get("sha256"))
            except OSError as e:                   # диск полон — не падать; trial остался, повтор на следующем claim (M-3)
                log("не удалось записать update.json: %s" % e)
                return
            log("обновление до v%d подтверждено" % AGENT_VERSION)

    def _hint(self, kind: str, msg: str) -> None:
        now = self.clock()
        at = self.hints.get(kind)
        if at is None or now - at >= self.HINT_SECONDS:
            self.hints[kind] = now
            log(msg)

    def check(self, release) -> bool:
        """Вызывается после claim без команды. True — agent.py.new проверен, update.json = staged: run выходит с 75."""
        if not self.enabled or not isinstance(release, dict):
            return False
        version, sha = release.get("version"), release.get("sha256")
        if type(version) is not int or not isinstance(sha, str) or not self.SHA_RE.fullmatch(sha):
            self._hint("format", "хаб прислал неполную публикацию: version=%.80r, sha256=%.80r" % (version, sha))  # design 10
            return False
        if version <= AGENT_VERSION:
            return False
        st = self.state()
        if st and st.get("state") in ("rejected", "rolled_back") and st.get("to_version") == version \
                and st.get("sha256") == sha:
            return False
        if self.launcher_version is None:
            self._hint("launcher", "доступна v%d; для самообновления запускайте через run.py" % version)
            return False
        until = self.retry_after.get((version, sha))
        if until is not None and self.clock() < until:
            return False
        return self.prepare(version, sha)

    def prepare(self, version: int, sha: str) -> bool:
        log("доступна v%d — готовлю обновление" % version)
        try:
            rel = self.hub.release()
        except (ConnectionError, HttpError) as e:
            return self._transient(version, sha, "скачивание: %s" % e)
        content = rel.get("content") if isinstance(rel, dict) else None
        if not isinstance(content, str) or rel.get("version") != version or rel.get("sha256") != sha:
            return self._transient(version, sha, "публикация сменилась между запросами")
        data = content.encode("utf-8")
        if hashlib.sha256(data).hexdigest() != sha:
            return self._reject(version, sha, "sha256 не сходится")
        try:
            compile(content, "agent.py.new", "exec")
        except (SyntaxError, ValueError) as e:
            return self._reject(version, sha, "не компилируется: %s" % e)
        if self.VERSION_RE.findall(content) != [str(version)]:
            return self._reject(version, sha, "AGENT_VERSION в файле не %d" % version)
        tmp = self.new_path + ".tmp"
        try:                                       # диск полон, нет прав — временный сбой, не трейсбек (design 9)
            with open(tmp, "wb") as f:
                f.write(data)
            try:
                os.chmod(tmp, os.stat(self.agent_path).st_mode & 0o777)
            except OSError:
                pass
            os.replace(tmp, self.new_path)
        except OSError as e:
            return self._transient(version, sha, "файлы: %s" % e)
        if self.should_stop():
            self._drop_new()
            return False
        try:
            code, out = self.run_selftest(self.new_path)
        except OSError as e:
            return self._transient(version, sha, "selftest: %s" % e)
        if self.should_stop():                     # Ctrl-C уронил и selftest — это не отказ версии
            self._drop_new()
            return False
        if code == 0:
            try:
                write_update_state(self.state_path, "staged", AGENT_VERSION, version, sha)
            except OSError as e:
                return self._transient(version, sha, "update.json: %s" % e)
            log("v%d готова, перезапуск через run.py" % version)
            return True
        if code is None:
            return self._transient(version, sha, "selftest: таймаут %d с" % self.SELFTEST_TIMEOUT)
        if code == 3:
            return self._transient(version, sha, "selftest: хаб или Eva недоступны")
        return self._reject(version, sha, "selftest: код %s: %s" % (code, out[-500:]))

    def _run_selftest(self, path: str):
        try:
            p = subprocess.run([sys.executable, path, "--config", self.config_path, "selftest"], cwd=self.dir,
                               stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=self.SELFTEST_TIMEOUT)
        except subprocess.TimeoutExpired:
            return None, ""
        return p.returncode, p.stdout.decode("utf-8", "replace")

    def _drop_new(self) -> None:
        for path in (self.new_path, self.new_path + ".tmp"):
            try:
                os.remove(path)
            except OSError:
                pass

    def _transient(self, version, sha, reason) -> bool:
        self._drop_new()
        self.retry_after[(version, sha)] = self.clock() + self.RETRY_SECONDS
        log("обновление до v%d отложено на %d мин: %s" % (version, self.RETRY_SECONDS // 60, reason))
        return False

    def _reject(self, version, sha, reason) -> bool:
        self._drop_new()
        try:
            write_update_state(self.state_path, "rejected", AGENT_VERSION, version, sha, reason)
        except OSError as e:
            log("не удалось записать update.json: %s" % e)
        log("v%d отвергнута: %s" % (version, reason))
        return False


class Runner:
    RETRY_ATTEMPTS = 3        # повторов команды после обрыва связи с Eva / 5xx (R-10)
    RETRY_MIN_WAIT = 30       # секунд до повтора

    def __init__(self, cfg: dict, hub: HubClient, eva: EvaClient, journal: Journal):
        self.cfg, self.hub, self.eva, self.journal = cfg, hub, eva, journal
        self.settings = None          # type: Optional[dict]
        self.executor = None          # type: Optional[Executor]
        self.stop = False
        self.sleep = time.sleep
        self.idle_limit = None        # type: Optional[int]
        self.updater = None           # type: Optional[Updater]
        self.exit_code = 0            # EXIT_UPDATE — выход для подмены версии

    @property
    def poll(self) -> float:
        return max(float(self.cfg.get("poll_seconds", 5) or 0), 0)

    def hub_wait(self, e: Exception, base: float) -> float:
        """Залогировать сбой хаба и вернуть, сколько ждать. 401/403 — токен неверен: повторять часто бессмысленно."""
        if isinstance(e, HttpError) and e.code in (401, 403):
            log("хаб: HTTP %d — неверный hub_token в agent.json, повтор через 60 с" % e.code)
            return 60
        log("хаб: %s, повтор через %d с" % (e, base))
        return base

    def load_settings(self) -> None:
        self.settings = self.hub.settings()
        self.executor = Executor(self.eva, self.hub)
        self.eva.pacing = float(self.settings.get("wiki_pacing", 0.2))
        log("настройки хаба v%s" % self.settings.get("settings_version"))

    def restore_pace(self) -> None:
        """После перезапуска время последнего вызова Eva неизвестно (живёт в памяти). Берём самое позднее время записи
        журнала: finished пишется сразу после последнего вызова команды, оценка не раньше настоящего вызова (M-9).
        Время в будущем (часы ушли назад) — считаем «сейчас»: пауза полная (M-6 #12). Оценка служит только паузе:
        last_call_iso (уходит хабу в claim) ставит лишь настоящий вызов Eva (M-7 #12)."""
        if self.eva.last_call_at is not None:
            return
        at = self.journal.last_at()
        if at is None:
            return
        ago = (datetime.now(timezone.utc) - at).total_seconds()
        if ago < 0:
            log("журнал: время последней записи в будущем (%s) — пауза перед первой командой полная" % at.isoformat())
            self.eva.last_call_at = time.monotonic()
            return                                 # вторая строка «не позже <будущее>» противоречила бы первой (M-2 #18)
        self.eva.last_call_at = time.monotonic() - ago
        log("темп: последний вызов Eva не позже %s (по журналу)"
            % at.astimezone(timezone.utc).replace(microsecond=0).isoformat())

    def send_result(self, rec: dict) -> bool:
        """Отчёт хабу; повторяется каждые 10 с, пока хаб не подтвердит. 404/409/422 — хаб отверг (команда отменена
        или удалена): считаем закрытой, чтобы не зациклиться. При остановке по сигналу делается одна попытка."""
        cid = rec["command_id"]
        payload = {"status": rec["status"], "result": rec.get("result"), "error": rec.get("error"),
                   "jira_calls": rec.get("jira_calls", 0), "rate_limited": rec.get("rate_limited", 0),
                   "agent_version": AGENT_VERSION}
        while True:
            try:
                self.hub.result(cid, payload)
                self.journal.acked(cid)
                return True
            except HttpError as e:
                if e.code in (404, 409, 422):
                    log("хаб отверг отчёт по #%s: %s — считаю закрытым" % (cid, e))
                    self.journal.acked(cid)
                    return True
                wait = self.hub_wait(e, 10)
            except ConnectionError as e:
                wait = self.hub_wait(e, 10)
            if self.stop:
                log("остановка: отчёт по #%s не подтверждён, будет дослан при следующем старте" % cid)
                return False
            self.sleep(wait)

    def wait_seconds(self, seconds) -> bool:
        """Спать заданное время шагами по секунде; False — остановлены во время ожидания."""
        remaining = float(seconds or 0)
        if remaining > 0:
            log("  пауза %.0f с" % remaining)
        while remaining > 0 and not self.stop:
            step = min(remaining, 1.0)
            self.sleep(step)
            remaining -= step
        return not self.stop

    def wait_pause(self, pause_seconds) -> bool:
        """Досыпать до зазора от последнего вызова Eva; False — остановлены во время паузы.
        Если вызовов Eva ещё не было — пауза не нужна."""
        pause = float(pause_seconds or 0)
        if pause > 0 and self.eva.last_call_at is not None:
            return self.wait_seconds(pause - (time.monotonic() - self.eva.last_call_at))
        return not self.stop

    def run_once(self, cmd: dict) -> dict:
        try:
            return self.executor.run(cmd)
        except Exception as e:                     # баг в обработчике не должен ронять цикл и зацикливать восстановление
            log("  ! необработанная ошибка: %r" % e)
            return {"status": "failed", "result": None,
                   "error": "внутренняя ошибка агента: %s: %s" % (type(e).__name__, str(e)[:300]),
                   "jira_calls": self.eva.calls, "rate_limited": self.eva.rate_limited}

    def execute(self, cmd: dict, journal_claimed: bool = True) -> None:
        cid = cmd["id"]
        log("#%s %s %s" % (cid, cmd["type"], json.dumps(cmd.get("params"), ensure_ascii=False)[:200]))
        if journal_claimed:
            self.journal.claimed(cid)              # сразу после claim, до паузы (R-9): остановка не теряет команду
        if not self.wait_pause(cmd.get("pause_seconds")):
            return
        self.journal.started(cid)                  # started — только после паузы
        calls = limited = attempt = 0
        while True:
            out = self.run_once(cmd)
            calls += out["jira_calls"]
            limited += out["rate_limited"]
            peer = out.pop("peer", "Eva")         # с кем оборвалась связь: Eva или хаб
            if not out.pop("retryable", False):
                break
            if attempt >= self.RETRY_ATTEMPTS:
                out = {"status": "failed", "result": None,
                       "error": "результат неизвестен (обрыв связи с %s), повтор безопасен: %s" % (peer, out.get("error"))}
                break
            attempt += 1
            log("  ! обрыв связи с %s, повтор %d/%d через проверку: %s" % (peer, attempt, self.RETRY_ATTEMPTS, out.get("error")))
            # F-1: обрыв не обновляет last_call_at, поэтому ждём фиксированно от момента отказа, а не от последнего вызова
            if not self.wait_seconds(max(float(cmd.get("pause_seconds") or 0), self.RETRY_MIN_WAIT)):
                return                             # остановка: finished не пишем, команду подхватит recover
        out["jira_calls"], out["rate_limited"] = calls, limited
        self.journal.finished(cid, out["status"], out.get("result"), out.get("error"), calls, limited)
        log("  → %s %s" % (out["status"], (out.get("error") or json.dumps(out.get("result"), ensure_ascii=False))[:300]))
        rec = dict(out)
        rec["command_id"] = cid
        self.send_result(rec)

    def fetch_command(self, cid):
        """Команда из хаба по id; None — хаб такой не знает; при недоступности хаба ждём и повторяем."""
        while True:
            try:
                return self.hub.command(cid)
            except HttpError as e:
                if e.code in (404, 409, 422):
                    log("#%s: хаб ответил %s" % (cid, e))
                    return None
                wait = self.hub_wait(e, 10)
            except ConnectionError as e:
                wait = self.hub_wait(e, 10)
            if self.stop:
                raise InterruptedError(cid)
            self.sleep(wait)

    def recover(self) -> None:
        for rec in self.journal.unacked_finished():
            log("досылаю отчёт по #%s" % rec["command_id"])
            self.send_result(rec)
        for cid in self.journal.unfinished_started():
            if self.stop:
                return
            try:
                cmd = self.fetch_command(cid)
            except InterruptedError:
                return
            if cmd is None:
                self.journal.acked(cid)
                continue
            if cmd.get("status") != "claimed":
                log("#%s уже %s в хабе — пропускаю" % (cid, cmd.get("status")))
                self.journal.acked(cid)
                continue
            log("#%s выдана до перезапуска — выполняю через проверку" % cid)
            self.execute(cmd, journal_claimed=False)   # claimed/started для неё уже в журнале (FM-4); пауза — из хаба (M-9)

    def reload_settings_before(self, cmd) -> bool:
        """Настройки сменились: перечитать ДО выполнения. False — остановка, команда не должна выполняться
        (если она уже выдана хабом, фиксируем claimed — подхватит recover)."""
        while True:
            try:
                self.load_settings()
                return True
            except (ConnectionError, HttpError) as e:
                wait = self.hub_wait(e, max(self.poll, 1))
            if self.stop:
                if cmd:
                    self.journal.claimed(cmd["id"])
                return False
            if not cmd:
                self.sleep(wait)
                return False
            self.sleep(wait)

    def loop(self) -> None:
        self.restore_pace()
        idle, recovered = 0, False
        while not self.stop:
            if self.settings is None:
                try:
                    self.load_settings()
                except (ConnectionError, HttpError) as e:
                    self.sleep(self.hub_wait(e, max(self.poll, 1)))
                    continue
            if not recovered:
                self.recover()
                recovered = not self.stop
                continue
            try:
                upd = self.updater
                resp = self.hub.claim(AGENT_VERSION, self.settings.get("settings_version"), self.eva.last_call_iso,
                                      upd.state() if upd else None, upd.launcher_version if upd else None)
            except (ConnectionError, HttpError) as e:
                self.sleep(self.hub_wait(e, max(self.poll, 1)))
                continue
            if self.updater:
                self.updater.confirm()
            cmd = resp.get("command")
            if resp.get("settings_version") != self.settings.get("settings_version"):
                if not self.reload_settings_before(cmd):
                    continue
            if cmd:
                idle = 0
                self.execute(cmd)
                continue
            if self.updater and not self.stop and self.updater.check(resp.get("agent_release")):
                if not self.stop:            # Ctrl-C во время подготовки — обычная остановка; staged доделает run.py
                    self.exit_code = EXIT_UPDATE
                return
            idle += 1
            if self.idle_limit is not None and idle >= self.idle_limit:
                return
            self.sleep(self.poll)


def acquire_lock(journal_path: str):
    """Исключительная блокировка <журнал>.lock: один процесс агента на журнал. Хаб довыдаёт команду без отчёта при
    следующем claim, поэтому второй процесс выполнял бы ту же команду одновременно с первым. None — уже занято."""
    f = open(journal_path + ".lock", "a")
    try:
        fcntl.flock(f.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
    except OSError:
        f.close()
        return None
    return f


def cmd_run(cfg: dict, journal_path: str, config_path: Optional[str] = None, once: bool = False) -> int:
    lock = acquire_lock(journal_path)
    if lock is None:
        log("агент уже запущен с журналом %s — второй процесс не стартует" % journal_path)
        return EXIT_LOCKED
    try:
        return _run(cfg, journal_path, config_path, once)
    finally:
        lock.close()


def cmd_prune(journal_path: str) -> int:
    """Удалить подтверждённые записи старше 30 дней. Под той же блокировкой, что run: prune перезаписывает файл, и запись,
    добавленная работающим агентом (finished — неотправленный отчёт), пропала бы (T2.3)."""
    lock = acquire_lock(journal_path)
    if lock is None:
        log("агент запущен с журналом %s — prune невозможен, сначала остановите агента" % journal_path)
        return 1
    try:
        removed = Journal(journal_path).prune(30)
        log("журнал: удалено записей %d" % removed)
        return 0
    finally:
        lock.close()


def launcher_version() -> Optional[int]:
    """Версия запускателя из окружения (run.py ставит JIRA_AGENT_LAUNCHER — имя общее с jira-agent); без запускателя — None."""
    try:
        return int(os.environ["JIRA_AGENT_LAUNCHER"])
    except (KeyError, ValueError):
        return None


def _run(cfg: dict, journal_path: str, config_path: Optional[str] = None, once: bool = False) -> int:
    runner = Runner(cfg, HubClient(cfg), EvaClient(cfg), Journal(journal_path))
    if once:
        runner.idle_limit = 1          # выполнить всё, что выдаст хаб, и выйти при первом пустом claim
    runner.updater = Updater(runner.hub, HERE, launcher_version(), cfg.get("auto_update", True) is not False, config_path)
    runner.updater.should_stop = lambda: runner.stop

    def on_signal(signum, frame):
        try:
            if runner.stop:
                log("повторный сигнал — выхожу немедленно")
            else:
                log("сигнал %s: доделываю текущую команду и выхожу" % signum)
        except RuntimeError:        # print внутри обработчика, прервавшего другой print (reentrant call)
            pass
        if runner.stop:
            sys.exit(130)
        runner.stop = True

    signal.signal(signal.SIGINT, on_signal)
    signal.signal(signal.SIGTERM, on_signal)
    log("eva-agent v%d: хаб %s, Eva %s, журнал %s" % (AGENT_VERSION, cfg["hub_url"], cfg["eva"]["base_url"], journal_path))
    runner.loop()
    log("выход для обновления (код %d)" % EXIT_UPDATE if runner.exit_code == EXIT_UPDATE else "остановлен")
    return runner.exit_code


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def cmd_check(cfg: dict) -> int:
    ok = True
    hub = HubClient(cfg)
    via = ""
    if hub.folder is not None:
        hub.timeout = 10                       # реле молчит дольше 10 с — хаб недоступен
        via = " через папку %s" % hub.folder.root
    try:
        s = hub.settings()
        log("хаб: доступен%s, настройки v%s" % (via, s.get("settings_version")))
    except (ConnectionError, HttpError) as e:
        log("хаб: НЕДОСТУПЕН — %s" % e)
        ok = False
    eva = EvaClient(cfg)
    eva.ATTEMPTS = 1                           # проверка отвечает сразу, без повторов с паузами
    try:
        eva.call("CmfDocument.list", filter=[], fields=["code"], slice=[0, 1])
        log("eva: ok")
    except (ConnectionError, HttpError, EvaError) as e:
        log("eva: НЕДОСТУПНА — %s" % e if isinstance(e, ConnectionError) else "eva: ошибка — %s" % e)
        ok = False
    return 0 if ok else 3


def cmd_selftest(cfg: dict) -> int:
    """Самопроверка версии перед обновлением: её запускает предыдущая версия. Контракт кодов MUST сохраняться:
    0 — хаб и Eva доступны, 3 — что-то недоступно (временно); любой другой код — файл негоден (design 10)."""
    log("eva-agent v%d" % AGENT_VERSION)
    return 0 if cmd_check(cfg) == 0 else 3


def main(argv=None) -> int:
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("--config", default=os.path.join(HERE, "agent.json"))
    ap.add_argument("--journal", default=os.path.join(HERE, "journal.jsonl"))
    ap.add_argument("--once", action="store_true", help="для run: выполнить выданные команды и выйти при первом пустом claim")
    ap.add_argument("cmd", nargs="?", default="run", choices=["run", "check", "selftest", "prune"])
    args = ap.parse_args(argv)
    cfg = load_config(args.config)
    if args.cmd == "check":
        return cmd_check(cfg)
    if args.cmd == "selftest":
        return cmd_selftest(cfg)
    if args.cmd == "prune":
        return cmd_prune(args.journal)
    return cmd_run(cfg, args.journal, args.config, args.once)


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