"""Воркер scan-gateway: запрос NATS → ack → S3 → горячая папка → ответ ИС → S3 `raw/` → нормализатор → конверт.

Spec image-recognition-gateway; design D-2, D-3, D-4, D-7; решения R-14..R-20.

- `on_request(msg)` — колбэк подписки NATS. Только разбирает и проверяет тело, регистрирует `job_id` и запускает
  задачу задания; ничего не ждёт (nats-py вызывает колбэки одной подписки последовательно, T11.4).
  Тело не по схеме / MAJOR ≠ 1 / `pipeline` ≠ `image` → `SCHEMA_MISMATCH` без ack; конверт публикуется, только если
  из тела извлекаются `job_id` и `file_id` (R-16), иначе запрос записывается в лог и отбрасывается.
- Задача задания: (1) ack `processing` — до чтения S3 и до ожидания слота; (2) слот семафора `SCAN_MAX_INFLIGHT`;
  (3) `handle(request)` — обработка и публикация итогового конверта (R-20).
- Повторный запрос с `job_id`, который в работе или в LRU обработанных (`DONE_CACHE_SIZE`), получает повторный ack с
  `attempt` повторного запроса; файл в ИС второй раз не отправляется (D-2, R-17).
- Ошибки — результат, не молчание (таблица D-7): каждое задание заканчивается опубликованным конвертом; исключение
  задания наружу не выходит (кроме отмены при остановке); неудачная публикация — лог error (ядро закроет задание по
  `reply_deadline`).
- Методы горячей папки синхронные (R-9) — вызываются через `asyncio.to_thread`; нормализатор — тоже в потоке.
"""

import asyncio
import json
import re
import uuid
from collections import OrderedDict
from collections.abc import Awaitable, Callable, Coroutine
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
from typing import Any, Protocol

from loguru import logger
from pydantic import ValidationError

from src.contract import (
    CONTRACT_VERSION,
    ErrorCode,
    Metrics,
    RecognitionRequest,
    RecognitionResult,
    RecognitionResultMessage,
    ResultStatus,
    S3Ref,
    ServiceInfo,
    build_ack_message,
    build_done_message,
    build_error_message,
    is_supported_major,
)
from src.core.clients import JSON_CONTENT_TYPE, OCTET_STREAM
from src.core.config import Settings
from src.core.nats import PublishFailed
from src.gateway.hotfolder import HotFolder, inbox_name
from src.normalizer import Normalizer

__all__ = [
    "MAX_ERROR_MESSAGE_CHARS",
    "IncomingMessage",
    "ObjectStorage",
    "ResultPublisher",
    "Worker",
    "raw_key",
    "result_key",
    "utc_now",
]

# Предел текста `error.message` в конверте: конверт остаётся далеко меньше лимита NATS (1 МБ) даже при 4 байтах на
# символ (T10.1). Длинное сообщение обрезается с многоточием.
MAX_ERROR_MESSAGE_CHARS = 4000
# Предел фрагментов `repr()` (исключения, недоверенные значения) внутри сообщений и логов
MAX_FRAGMENT_CHARS = 300
_ELLIPSIS = "…"
_MAX_ATTEMPT = 2**31 - 1  # attempt из тела, отвергнутого схемой, берётся только из разумного диапазона
_UUID_RE = re.compile(r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}", re.ASCII)
_IMAGE_PIPELINE = "image"

SleepFn = Callable[[float], Awaitable[None]]
ClockFn = Callable[[], datetime]


class IncomingMessage(Protocol):
    """Сообщение NATS в колбэке подписки: воркеру нужно только тело (`nats.aio.msg.Msg` подходит)."""

    data: bytes


class ObjectStorage(Protocol):
    """Часть `S3Storage`, нужная воркеру; ошибки — классы `src.core.clients`."""

    bucket: str

    async def get_bytes(self, key: str, *, bucket: str | None = None) -> bytes: ...

    async def put_bytes(self, key: str, data: bytes, content_type: str = OCTET_STREAM) -> None: ...

    async def put_json(self, key: str, obj: Any) -> None: ...


class ResultPublisher(Protocol):
    """Часть `NatsGateway`: публикация конверта в `recognition.result` с `Nats-Msg-Id` и повторами."""

    async def publish_result(self, message: RecognitionResultMessage, attempt: int) -> None: ...


def utc_now() -> datetime:
    return datetime.now(UTC)


def raw_key(job_id: object) -> str:
    """Ключ сырого ответа ИС в S3 (D-4)."""
    return f"raw/{job_id}.json"


def result_key(file_id: object, job_id: object) -> str:
    """Ключ файла результата в S3 (контракт, «Схема ключей S3»)."""
    return f"results/{file_id}/{job_id}/result.json"


def clean_text(text: str, limit: int = MAX_ERROR_MESSAGE_CHARS) -> str:
    """Текст, безопасный для конверта: одиночные суррогаты (из `\\udXXX` в JSON) заменены на `?` — иначе публикация
    падает на кодировании в UTF-8 (T10.2); длина ограничена `limit` с многоточием в конце (T10.1)."""
    safe = text.encode("utf-8", "replace").decode("utf-8")
    if len(safe) > limit:
        safe = safe[: limit - len(_ELLIPSIS)] + _ELLIPSIS
    return safe


def _fragment(value: object) -> str:
    """Короткое безопасное представление недоверенного значения для сообщения или лога."""
    try:
        text = repr(value)
    except Exception:  # repr не должен ронять обработку ошибки
        text = f"<{type(value).__name__}>"
    return clean_text(text, MAX_FRAGMENT_CHARS)


def _describe(exc: BaseException) -> str:
    return f"{type(exc).__name__}: {clean_text(str(exc), MAX_FRAGMENT_CHARS)}"


def _parse_json(data: bytes) -> tuple[object, str | None]:
    """(разобранный JSON, None) или (None, описание ошибки разбора)."""
    try:
        return json.loads(data), None
    except (ValueError, RecursionError) as exc:  # UnicodeDecodeError — подкласс ValueError
        return None, _describe(exc)


@dataclass(frozen=True)
class _Rejection:
    """Тело запроса, не прошедшее проверку: причина и (если тело — JSON-объект) исходный словарь."""

    reason: str
    raw: dict[str, Any] | None


@dataclass
class _JobContext:
    """Состояние задания, нужное для конверта при непредвиденной ошибке: сырой ответ уже в S3."""

    raw_s3_key: str | None = None


@dataclass(frozen=True)
class _WaitResult:
    answer: bytes | None
    polls: int
    failures: int
    last_error: BaseException | None


class Worker:
    """Обработка запросов `recognition.image.request` (см. докстринг модуля).

    Зависимости внедряются: `storage` (S3), `publisher` (NATS), `hotfolder`, `normalizer`; `clock` — часы UTC для
    дедлайна (R-20), `sleep` — пауза между опросами (подмены для тестов).
    Запросы принимаются между `start()` и `stop()`.
    """

    def __init__(  # noqa: PLR0913 - зависимости воркера; всё, кроме настроек, — только именованные
        self,
        settings: Settings,
        *,
        storage: ObjectStorage,
        publisher: ResultPublisher,
        hotfolder: HotFolder,
        normalizer: Normalizer,
        clock: ClockFn = utc_now,
        sleep: SleepFn = asyncio.sleep,
    ) -> None:
        self._settings = settings
        self._storage = storage
        self._publisher = publisher
        self._hotfolder = hotfolder
        self._normalizer = normalizer
        self._clock = clock
        self._sleep = sleep
        self.service = ServiceInfo.model_validate({"name": settings.SERVICE_NAME, "version": settings.VERSION})
        self._semaphore = asyncio.Semaphore(settings.SCAN_MAX_INFLIGHT)
        self._active: set[str] = set()  # job_id в работе (ack, ожидание слота, обработка) — не вытесняются
        self._done: OrderedDict[str, None] = OrderedDict()  # LRU обработанных job_id
        self._tasks: set[asyncio.Task[None]] = set()
        self._accepting = False

    # --- жизненный цикл ---------------------------------------------------------------------------------------
    @property
    def active_jobs(self) -> int:
        """Задания в работе (включая ожидающие слот)."""
        return len(self._active)

    def start(self) -> None:
        """Начать приём запросов (колбэк `on_request` подписывается после этого)."""
        self._accepting = True
        logger.info(
            "Воркер scan-gateway запущен. max_inflight={max_inflight}, poll={poll}, timeout={timeout}",
            max_inflight=self._settings.SCAN_MAX_INFLIGHT,
            poll=self._settings.SCAN_POLL_SECONDS,
            timeout=self._settings.SCAN_TIMEOUT_SECONDS,
        )

    async def stop(self) -> None:
        """Прекратить приём, отменить задания в работе и дождаться их завершения. Повторный вызов безопасен.

        Отменённые задания конверта не публикуют: ядро закроет их по `reply_deadline` (D-2).
        """
        self._accepting = False
        tasks = list(self._tasks)
        for task in tasks:
            task.cancel()
        if tasks:
            await asyncio.gather(*tasks, return_exceptions=True)
            logger.info("Воркер остановлен, отменено заданий: {count}", count=len(tasks))

    async def wait_idle(self) -> None:
        """Дождаться завершения всех запущенных задач (ack, задания, публикации отказов) — для тестов и остановки."""
        while self._tasks:
            await asyncio.gather(*list(self._tasks), return_exceptions=True)

    # --- приём запроса ----------------------------------------------------------------------------------------
    async def on_request(self, msg: IncomingMessage) -> None:
        """Колбэк подписки: проверка тела, дедупликация по `job_id`, запуск задачи. Не ждёт ни сети, ни ИС."""
        if not self._accepting:
            logger.warning("Воркер не принимает запросы (остановлен): сообщение отброшено, ядро повторит запрос")
            return
        parsed = self._parse_request(msg.data)
        if isinstance(parsed, _Rejection):
            self._reject(parsed)
            return
        request = parsed
        job_id = str(request.job_id)
        if self._seen(job_id):
            logger.info(
                "Повторный запрос: повторный ack без второй отправки в ИС. job_id={job_id}, attempt={attempt}",
                job_id=job_id,
                attempt=request.attempt,
            )
            self._spawn(self._publish(self._ack(request), request.attempt), f"ack-repeat:{job_id}")
            return
        self._active.add(job_id)
        logger.info(
            "Запрос принят. job_id={job_id}, file_id={file_id}, attempt={attempt}, file_type={file_type}",
            job_id=job_id,
            file_id=request.file_id,
            attempt=request.attempt,
            file_type=request.file_type,
        )
        self._spawn(self._run_job(request), f"job:{job_id}")

    def _parse_request(self, data: bytes) -> RecognitionRequest | _Rejection:
        raw, error = _parse_json(data)
        if error is not None:
            return _Rejection(f"тело запроса не разбирается как JSON ({error})", None)
        if not isinstance(raw, dict):
            return _Rejection(f"тело запроса — не JSON-объект ({type(raw).__name__})", None)
        version = raw.get("schema_version")
        if not is_supported_major(version):
            return _Rejection(f"неподдерживаемая версия контракта {_fragment(version)}: ожидается MAJOR 1", raw)
        pipeline = raw.get("pipeline")
        if pipeline != _IMAGE_PIPELINE:
            return _Rejection(f"pipeline {_fragment(pipeline)} не обслуживается scan-gateway (ожидается image)", raw)
        try:
            return RecognitionRequest.model_validate(raw)
        except ValidationError as exc:
            details = "; ".join(
                f"{'.'.join(str(part) for part in item['loc']) or '<тело>'}: {item['msg']}"
                for item in exc.errors(include_url=False, include_input=False)[:5]
            )
            return _Rejection(f"запрос не соответствует схеме recognition-request: {details}", raw)

    def _reject(self, rejection: _Rejection) -> None:
        """`SCHEMA_MISMATCH` без ack (R-16): конверт — только если в теле есть корректные `job_id` и `file_id`.

        Для `job_id` в работе или уже обработанного конверт не публикуется (T12.1): его `Nats-Msg-Id`
        `result:<job_id>:<attempt>` совпал бы с настоящим результатом, и ядро отбросило бы настоящий как дубль.
        """
        raw = rejection.raw or {}
        job_id, file_id = raw.get("job_id"), raw.get("file_id")
        if not (_is_uuid(job_id) and _is_uuid(file_id)):
            logger.warning(
                "SCHEMA_MISMATCH: запрос отброшен без ответа (нет корректных job_id/file_id). reason={reason}",
                reason=clean_text(rejection.reason, MAX_FRAGMENT_CHARS),
            )
            return
        # `_active`/`_done` хранят канонический вид (нижний регистр), а UUID в теле может прийти в верхнем
        key = str(uuid.UUID(str(job_id)))
        if self._seen(key):
            logger.warning(
                "SCHEMA_MISMATCH для задания в работе/обработанного: конверт не публикуется. job_id={job_id}, "
                "reason={reason}",
                job_id=key,
                reason=clean_text(rejection.reason, MAX_FRAGMENT_CHARS),
            )
            return
        attempt = raw.get("attempt")
        if not (type(attempt) is int and 1 <= attempt <= _MAX_ATTEMPT):
            attempt = 1
        logger.warning(
            "SCHEMA_MISMATCH: запрос отвергнут, ack не публикуется. job_id={job_id}, reason={reason}",
            job_id=job_id,
            reason=clean_text(rejection.reason, MAX_FRAGMENT_CHARS),
        )
        message = build_error_message(
            str(job_id), str(file_id), self.service, ErrorCode.SCHEMA_MISMATCH, clean_text(rejection.reason)
        )
        self._spawn(self._publish(message, attempt), f"schema-mismatch:{job_id}")

    # --- дедупликация (D-2, R-17) -----------------------------------------------------------------------------
    def _seen(self, job_id: str) -> bool:
        if job_id in self._active:
            return True
        if job_id in self._done:
            self._done.move_to_end(job_id)
            return True
        return False

    def _finish(self, job_id: str) -> None:
        self._active.discard(job_id)
        self._done[job_id] = None
        self._done.move_to_end(job_id)
        while len(self._done) > self._settings.DONE_CACHE_SIZE:
            self._done.popitem(last=False)

    # --- задачи -----------------------------------------------------------------------------------------------
    def _spawn(self, coro: Coroutine[Any, Any, None], name: str) -> None:
        task = asyncio.create_task(coro, name=name)
        self._tasks.add(task)
        task.add_done_callback(self._tasks.discard)

    async def _run_job(self, request: RecognitionRequest) -> None:
        job_id = str(request.job_id)
        try:
            await self._publish(self._ack(request), request.attempt)  # до чтения S3 и до ожидания слота
            async with self._semaphore:
                await self.handle(request)
        except asyncio.CancelledError:
            logger.warning("Задание отменено при остановке воркера. job_id={job_id}", job_id=job_id)
            raise
        except Exception:  # handle не бросает; страховка, чтобы исключение не осталось в задаче
            logger.exception("Непредвиденная ошибка задания. job_id={job_id}", job_id=job_id)
        finally:
            self._finish(job_id)

    def _ack(self, request: RecognitionRequest) -> RecognitionResultMessage:
        return build_ack_message(request.job_id, request.file_id, self.service)

    async def _publish(self, message: RecognitionResultMessage, attempt: int) -> None:
        """Публикация с повторами `NatsGateway`; неудача — лог error, исключение не выходит (D-7)."""
        try:
            await self._publisher.publish_result(message, attempt)
        except PublishFailed as exc:
            logger.error(
                "Конверт не опубликован, ядро закроет задание по reply_deadline. job_id={job_id}, status={status}, "
                "msg_id={msg_id}, attempts={attempts}, cause={cause}",
                job_id=message.job_id,
                status=message.status.value,
                msg_id=exc.msg_id,
                attempts=exc.attempts,
                cause=_fragment(exc.__cause__),
            )
        except Exception as exc:
            logger.error(
                "Конверт не опубликован: непредвиденная ошибка. job_id={job_id}, status={status}, error={error}",
                job_id=message.job_id,
                status=message.status.value,
                error=_describe(exc),
            )

    # --- обработка задания ------------------------------------------------------------------------------------
    async def handle(self, request: RecognitionRequest) -> RecognitionResultMessage:
        """Обработать задание и опубликовать итоговый конверт; вернуть его.

        Не бросает исключений (кроме `asyncio.CancelledError`): любой сбой становится конвертом `error` по D-7.
        Ack и семафор — забота вызывающего (`on_request`); здесь — S3 → горячая папка → ответ → `raw/` →
        нормализатор → `results/` → конверт.
        """
        context = _JobContext()
        try:
            message = await self._process(request, context)
        except asyncio.CancelledError:
            raise
        except Exception as exc:
            logger.exception("Непредвиденная ошибка обработки. job_id={job_id}", job_id=request.job_id)
            message = self._error(
                request,
                ErrorCode.UNKNOWN_ERROR,
                f"непредвиденная ошибка scan-gateway: {_describe(exc)}",
                raw_s3_key=context.raw_s3_key,
            )
        logger.info(
            "Задание обработано. job_id={job_id}, status={status}, error_code={code}",
            job_id=request.job_id,
            status=message.status.value,
            code=message.error.code.value if message.error else None,
        )
        await self._publish(message, request.attempt)
        return message

    def _error(
        self,
        request: RecognitionRequest,
        code: ErrorCode,
        text: str,
        *,
        raw_s3_key: str | None = None,
        metrics: Metrics | None = None,
    ) -> RecognitionResultMessage:
        return build_error_message(
            request.job_id,
            request.file_id,
            self.service,
            code,
            clean_text(text),
            raw_s3_key=raw_s3_key,
            metrics=metrics,
        )

    async def _process(self, request: RecognitionRequest, context: _JobContext) -> RecognitionResultMessage:
        """Шаги задания; каждая неудача — конверт `error` по строке таблицы D-7."""
        sent = await self._send(request)
        if isinstance(sent, RecognitionResultMessage):
            return sent
        name, started = sent
        job_id = str(request.job_id)
        # 3. Опрос исходящего каталога до min(reply_deadline, старт + SCAN_TIMEOUT_SECONDS) (D-3, R-20)
        deadline = min(request.reply_deadline, started + timedelta(seconds=self._settings.SCAN_TIMEOUT_SECONDS))
        waited = await self._wait_answer(job_id, deadline)
        if waited.answer is None:
            if waited.failures == waited.polls and waited.last_error is not None:
                return self._error(
                    request,
                    ErrorCode.RECOGNITION_FAILED,
                    f"горячая папка недоступна: исходящий каталог не прочитан ни разу за {waited.polls} опросов "
                    f"({_describe(waited.last_error)})",
                )
            logger.warning(
                "ИС Сканирование не ответила до дедлайна. job_id={job_id}, deadline={deadline}, polls={polls}",
                job_id=job_id,
                deadline=deadline.isoformat(),
                polls=waited.polls,
            )
            return self._error(
                request,
                ErrorCode.TIMEOUT,
                f"ИС Сканирование не ответила до {deadline.isoformat()} (опросов: {waited.polls}); "
                f"входящий файл {name} остаётся в каталоге, повторной отправки нет",
            )
        return await self._accept_answer(request, waited.answer, context)

    async def _send(self, request: RecognitionRequest) -> tuple[str, datetime] | RecognitionResultMessage:  # noqa: PLR0911 - каждая ветка — строка таблицы D-7
        """Срок, исходник из S3, запись во входящий каталог: (имя файла, момент записи) или конверт `error`."""
        job_id = str(request.job_id)
        if request.reply_deadline <= self._clock():
            # задание дождалось слота после срока ответа: ядро уже закрывает его, файл в ИС не отправляется
            return self._error(
                request,
                ErrorCode.TIMEOUT,
                f"срок ответа {request.reply_deadline.isoformat()} истёк до отправки файла в ИС Сканирование",
            )

        # 1. Исходник из S3 (один бакет сервиса, R-19)
        source = request.s3
        if source.bucket != self._storage.bucket:
            return self._error(
                request,
                ErrorCode.FILE_READ_ERROR,
                f"бакет {_fragment(source.bucket)} не обслуживается: scan-gateway работает с одним бакетом "
                f"{_fragment(self._storage.bucket)}",
            )
        try:
            data = await self._storage.get_bytes(source.key, bucket=source.bucket)
        except Exception as exc:
            logger.warning(
                "Исходник не прочитан из S3. job_id={job_id}, error={error}", job_id=job_id, error=_describe(exc)
            )
            return self._error(
                request,
                ErrorCode.FILE_READ_ERROR,
                f"исходный файл {source.bucket}/{source.key} не прочитан из S3: {_describe(exc)}",
            )

        # 2. Во входящий каталог как <job_id>_<original_name> (R-5, R-15)
        try:
            name = inbox_name(job_id, request.original_name)
        except ValueError:
            return self._error(
                request,
                ErrorCode.RECOGNITION_FAILED,
                f"недопустимое имя файла {_fragment(request.original_name)}: файл не передан в ИС Сканирование",
            )
        try:
            await asyncio.to_thread(self._hotfolder.put, name, data)
        except Exception as exc:
            logger.error(
                "Горячая папка недоступна: файл не записан. job_id={job_id}, error={error}",
                job_id=job_id,
                error=_describe(exc),
            )
            return self._error(
                request,
                ErrorCode.RECOGNITION_FAILED,
                f"горячая папка недоступна: файл не записан во входящий каталог ({_describe(exc)})",
            )
        return name, self._clock()

    async def _accept_answer(  # noqa: PLR0911 - каждая ветка — строка таблицы D-7
        self, request: RecognitionRequest, answer: bytes, context: _JobContext
    ) -> RecognitionResultMessage:
        """Ответ ИС: сырой в S3, нормализатор, файл результата; конверт итога."""
        job_id = str(request.job_id)

        # 4. Сырой ответ в S3 как есть — до нормализации (D-4, R-18)
        parsed, parse_error = _parse_json(answer)
        key = raw_key(job_id)
        try:
            await self._storage.put_bytes(key, answer, JSON_CONTENT_TYPE if parse_error is None else OCTET_STREAM)
        except Exception as exc:
            logger.error(
                "Сырой ответ ИС не сохранён. job_id={job_id}, error={error}", job_id=job_id, error=_describe(exc)
            )
            return self._error(
                request, ErrorCode.UNKNOWN_ERROR, f"сырой ответ ИС не сохранён в S3 ({key}): {_describe(exc)}"
            )
        context.raw_s3_key = key
        if parse_error is not None:
            return self._error(
                request,
                ErrorCode.INVALID_FORMAT,
                f"ответ ИС Сканирование не разбирается как JSON: {parse_error}",
                raw_s3_key=key,
            )

        # 5. Нормализатор (D-5); его исключение — UNKNOWN_ERROR с сохранённым сырым ответом
        try:
            outcome = await asyncio.to_thread(self._normalizer.normalize, parsed)
        except Exception as exc:
            logger.exception("Нормализатор ответа ИС упал. job_id={job_id}", job_id=job_id)
            return self._error(
                request, ErrorCode.UNKNOWN_ERROR, f"нормализатор ответа ИС упал: {_describe(exc)}", raw_s3_key=key
            )
        if outcome.status is ResultStatus.ERROR:
            error = outcome.error
            if error is None:  # исключено инвариантом NormalizeOutcome
                raise ValueError("NormalizeOutcome со status=error без error")
            return self._error(request, error.code, error.message, raw_s3_key=key, metrics=outcome.metrics)

        # 6. Успех: файл результата в S3, затем конверт с result_s3
        result = RecognitionResult.model_validate(
            {
                "schema_version": CONTRACT_VERSION,
                "job_id": request.job_id,
                "file_id": request.file_id,
                "service": self.service,
                "status": outcome.status,
                "raw_s3_key": key,
                "documents": list(outcome.documents),
                **({"metrics": outcome.metrics} if outcome.metrics is not None else {}),
            }
        )
        target = result_key(request.file_id, request.job_id)
        try:
            await self._storage.put_json(target, result.to_wire())
        except Exception as exc:
            logger.error(
                "Файл результата не записан. job_id={job_id}, error={error}", job_id=job_id, error=_describe(exc)
            )
            return self._error(
                request,
                ErrorCode.UNKNOWN_ERROR,
                f"файл результата не записан в S3 ({target}): {_describe(exc)}",
                raw_s3_key=key,
            )
        return build_done_message(
            request.job_id,
            request.file_id,
            self.service,
            S3Ref(bucket=self._storage.bucket, key=target),
            needs_check=outcome.status is ResultStatus.NEEDS_CHECK,
            raw_s3_key=key,
            metrics=outcome.metrics,
        )

    async def _wait_answer(self, job_id: str, deadline: datetime) -> _WaitResult:
        """Опрос `poll(job_id)` каждые `SCAN_POLL_SECONDS` до дедлайна (последний опрос — в момент дедлайна).

        Исключения `poll` (SMB и прочие) — warning, опрос продолжается (R-14). Ответ не JSON может быть недописан
        (ИС пишет на шару без атомарного переименования): он перечитывается через период опроса и принимается, когда
        стал JSON или перестал меняться; на дедлайне берётся последний прочитанный.
        """
        polls = failures = 0
        last_error: BaseException | None = None
        pending: bytes | None = None
        while True:
            polls += 1
            answer: bytes | None
            try:
                answer = await asyncio.to_thread(self._hotfolder.poll, job_id)
            except Exception as exc:
                failures += 1
                last_error = exc
                answer = None
                logger.warning(
                    "Опрос исходящего каталога не удался, опрос продолжается. job_id={job_id}, error={error}",
                    job_id=job_id,
                    error=_describe(exc),
                )
            if answer is not None:
                if answer == pending or _parse_json(answer)[1] is None:
                    return _WaitResult(answer, polls, failures, last_error)
                pending = answer
            now = self._clock()
            if now >= deadline:
                return _WaitResult(pending, polls, failures, last_error)
            await self._sleep(min(self._settings.SCAN_POLL_SECONDS, (deadline - now).total_seconds()))


def _is_uuid(value: object) -> bool:
    return isinstance(value, str) and _UUID_RE.fullmatch(value) is not None