arm-scan-gateway/api/tests/unit/test_worker.py

"""Воркер шлюза (`src.core.worker`, задача 5.3): сценарии spec image-recognition-gateway, D-2, D-3, D-7, R-14..R-20.

Обвязка (фейки S3/NATS, шпион горячей папки, эмулятор ИС, журнал событий) — `tests/worker_helpers.py`, фикстура
`make_env` — `tests/unit/conftest.py`. Параллелизм и остановка при очереди на слот — `test_worker_concurrency.py`.
"""

import asyncio
import json
import shutil
import uuid
from collections.abc import Callable
from datetime import UTC, datetime, timedelta
from pathlib import Path
from typing import Any

import pytest
from jsonschema import Draft202012Validator, FormatChecker

from src.contract import (
    ErrorCode,
    MessageError,
    ResultStatus,
    StatementDocument,
)
from src.core.clients import JSON_CONTENT_TYPE, OCTET_STREAM
from src.core.worker import MAX_ERROR_MESSAGE_CHARS
from src.gateway.hotfolder import AnswerFile
from src.normalizer import NOT_IMPLEMENTED_MESSAGE, NormalizeOutcome
from tests.worker_helpers import (
    API_DIR,
    BUCKET,
    CONTRACTS_DIR,
    PDF_BYTES,
    SUCCESS_ANSWER,
    EnvFactory,
    emulator,
    eventually,
)

# `<корень scan-gateway>/../fixtures/ocr` (workspace и worktree `.wt/<срез>/`), как в тестах эмулятора
REAL_FIXTURES = API_DIR.parent.parent / "fixtures" / "ocr"
RESULT_VALIDATOR = Draft202012Validator(
    json.loads((CONTRACTS_DIR / "recognition-result.schema.json").read_text(encoding="utf-8")),
    format_checker=FormatChecker(),
)
STATEMENT_EXAMPLE = json.loads((CONTRACTS_DIR / "examples" / "result-statement-minimal.json").read_text("utf-8"))


# Логи воркера перехватываются (`log_lines` из conftest): вывод pytest чистый
pytestmark = pytest.mark.usefixtures("log_lines")


class RaisingNormalizer:
    def normalize(self, raw: object) -> NormalizeOutcome:
        raise RuntimeError("нормализатор сломан")


class StubNormalizer:
    """Возвращает заданный исход (done / needs_check / error) — в Срезе 0 успешный путь достижим только так."""

    def __init__(self, outcome: NormalizeOutcome) -> None:
        self.outcome = outcome
        self.calls: list[object] = []

    def normalize(self, raw: object) -> NormalizeOutcome:
        self.calls.append(raw)
        return self.outcome


def assert_error(body: dict[str, Any], code: ErrorCode, *, raw: bool) -> None:
    assert body["status"] == "error"
    assert body["error"]["code"] == code.value
    assert body["error"]["message"]
    if raw:
        assert body["raw_s3_key"] == f"raw/{body['job_id']}.json"
    else:
        assert "raw_s3_key" not in body
    assert "result_s3" not in body


# --- подтверждение запроса до обработки ---------------------------------------------------------------------------
async def test_ack_is_published_before_s3_read_and_before_file_appears(make_env: EnvFactory) -> None:
    env = await make_env()
    request = env.request()
    async with emulator(env):
        await env.run(request)
    job_id = request["job_id"]
    name = f"{job_id}_ok_doc.pdf"
    kinds = [event for event in env.events if event[1] in (f"ack:{job_id}:1", request["s3"]["key"], name)]
    assert kinds == [("publish", f"ack:{job_id}:1"), ("s3_get", request["s3"]["key"]), ("put", name)]


async def test_file_in_inbox_is_byte_for_byte_s3_object(make_env: EnvFactory) -> None:
    env = await make_env()
    request = env.request(original_name="aep_5.pdf")
    await env.deliver(request)
    path = env.local.input_dir / f"{request['job_id']}_aep_5.pdf"
    await eventually(path.exists)
    assert path.read_bytes() == PDF_BYTES
    await env.worker.stop()


async def test_repeated_request_gets_second_ack_and_one_file(make_env: EnvFactory) -> None:
    env = await make_env()
    request = env.request(original_name="aep_5.pdf")
    await env.deliver(request)
    job_id = request["job_id"]
    await eventually(lambda: len(env.hotfolder.puts) == 1)
    await env.deliver({**request, "attempt": 2})  # ядро не дождалось ack и повторило запрос
    await eventually(lambda: len(env.acks(job_id)) == 2)
    await asyncio.sleep(0.05)
    assert env.acks(job_id) == [f"ack:{job_id}:1", f"ack:{job_id}:2"]
    assert env.inbox() == [f"{job_id}_aep_5.pdf"]
    assert env.hotfolder.puts == [f"{job_id}_aep_5.pdf"]
    assert len(env.storage.get_calls) == 1


async def test_repeated_request_after_result_gets_ack_without_reprocessing(make_env: EnvFactory) -> None:
    env = await make_env()
    request = env.request()
    async with emulator(env):
        await env.run(request)
        await env.deliver(request)
        await asyncio.wait_for(env.worker.wait_idle(), 5)
    job_id = request["job_id"]
    assert env.acks(job_id) == [f"ack:{job_id}:1", f"ack:{job_id}:1"]
    assert len(env.results(job_id)) == 1
    assert env.hotfolder.puts == [f"{job_id}_ok_doc.pdf"]


async def test_done_cache_is_bounded_lru(make_env: EnvFactory) -> None:
    env = await make_env(DONE_CACHE_SIZE=2)
    requests = [env.request() for _ in range(3)]
    async with emulator(env):
        for request in requests:
            await env.run(request)
        # первый job_id вытеснен из LRU размера 2 — его повтор обрабатывается заново (в проде — 1000 заданий назад)
        await env.deliver(requests[0])
        await asyncio.wait_for(env.worker.wait_idle(), 5)
        # третий ещё в кэше — повтор только с ack
        await env.deliver(requests[2])
        await asyncio.wait_for(env.worker.wait_idle(), 5)
    assert len(env.results(requests[0]["job_id"])) == 2
    assert len(env.results(requests[2]["job_id"])) == 1
    assert len(env.acks(requests[2]["job_id"])) == 2


async def test_on_request_returns_without_waiting_for_publication(make_env: EnvFactory) -> None:
    env = await make_env()
    env.client.block = asyncio.Event()
    request = env.request()
    await asyncio.wait_for(env.deliver(request), 0.2)  # публикация ack висит, а колбэк уже вернулся (T11.4)
    assert env.client.published == []
    env.client.block.set()
    async with emulator(env):
        await asyncio.wait_for(env.worker.wait_idle(), 5)
    assert len(env.results(request["job_id"])) == 1


# --- результат по контракту ---------------------------------------------------------------------------------------
async def test_success_answer_gives_not_implemented_with_raw_key(make_env: EnvFactory) -> None:
    env = await make_env()
    request = env.request(attempt=3)
    async with emulator(env):
        body = await env.run(request)
    job_id = request["job_id"]
    assert_error(body, ErrorCode.NOT_IMPLEMENTED, raw=True)
    assert body["error"]["message"] == NOT_IMPLEMENTED_MESSAGE
    assert body["file_id"] == request["file_id"]
    assert [mid for mid, _ in env.published()] == [f"ack:{job_id}:3", f"result:{job_id}:3"]
    raw = env.storage.objects[(BUCKET, f"raw/{job_id}.json")]
    assert raw == env.local.answer_path(job_id).read_bytes()  # как есть, байт в байт
    assert json.loads(raw) == {**SUCCESS_ANSWER, "file_uid": job_id}
    assert env.storage.content_types[(BUCKET, f"raw/{job_id}.json")] == JSON_CONTENT_TYPE
    # сырой ответ сохранён раньше публикации результата
    assert env.events.index(("s3_put", f"raw/{job_id}.json")) < env.events.index(("publish", f"result:{job_id}:3"))
    assert env.inbox() == [f"{job_id}_ok_doc.pdf.done"]


@pytest.mark.fixtures
async def test_real_fixture_error_is_passed_through(make_env: EnvFactory, tmp_path: Path) -> None:
    if not REAL_FIXTURES.is_dir():
        pytest.skip("fixtures недоступен: положите репозиторий fixtures рядом с workspace")
    real = tmp_path / "real-fixtures"
    real.mkdir()
    for source in REAL_FIXTURES.glob("*.json"):
        shutil.copy(source, real / source.name)
    expected = json.loads((real / "aep_5.json").read_text(encoding="utf-8"))["error"]
    env = await make_env()
    request = env.request(original_name="aep_5.PDF")
    async with emulator(env, real):
        body = await env.run(request)
    assert_error(body, ErrorCode.INVALID_DOCUMENT, raw=True)
    assert body["error"]["message"] == expected["message"]
    assert (BUCKET, f"raw/{request['job_id']}.json") in env.storage.objects


async def test_unknown_name_gets_emulator_invalid_document(make_env: EnvFactory) -> None:
    env = await make_env()
    request = env.request(original_name="unknown.pdf")
    async with emulator(env):
        body = await env.run(request)
    assert_error(body, ErrorCode.INVALID_DOCUMENT, raw=True)
    assert "unknown.pdf" in body["error"]["message"]


@pytest.mark.parametrize("status", [ResultStatus.DONE, ResultStatus.NEEDS_CHECK])
async def test_successful_outcome_writes_result_file(make_env: EnvFactory, status: ResultStatus) -> None:
    document = StatementDocument.model_validate(STATEMENT_EXAMPLE["documents"][0])
    normalizer = StubNormalizer(NormalizeOutcome(status=status, documents=[document]))
    env = await make_env(normalizer=normalizer)
    request = env.request()
    async with emulator(env):
        body = await env.run(request)
    job_id, file_id = request["job_id"], request["file_id"]
    key = f"results/{file_id}/{job_id}/result.json"
    assert body["status"] == status.value
    assert body["result_s3"] == {"bucket": BUCKET, "key": key}
    assert body["result_schema_version"] == "1.0"
    assert body["raw_s3_key"] == f"raw/{job_id}.json"
    assert "error" not in body
    result = json.loads(env.storage.objects[(BUCKET, key)].decode("utf-8"))
    RESULT_VALIDATOR.validate(result)
    assert result["status"] == status.value
    assert result["job_id"] == job_id
    assert result["file_id"] == file_id
    assert result["service"]["name"] == "scan-gateway"
    assert result["raw_s3_key"] == f"raw/{job_id}.json"
    assert result["documents"] == STATEMENT_EXAMPLE["documents"]
    assert env.storage.content_types[(BUCKET, key)] == JSON_CONTENT_TYPE
    assert normalizer.calls == [{**SUCCESS_ANSWER, "file_uid": job_id}]
    assert env.events.index(("s3_put", key)) < env.events.index(("publish", f"result:{job_id}:1"))


async def test_result_file_write_failure_is_unknown_error_with_raw_key(make_env: EnvFactory) -> None:
    document = StatementDocument.model_validate(STATEMENT_EXAMPLE["documents"][0])
    env = await make_env(normalizer=StubNormalizer(NormalizeOutcome(status=ResultStatus.DONE, documents=[document])))
    env.storage.fail_put_prefixes = ("results/",)
    request = env.request()
    async with emulator(env):
        body = await env.run(request)
    assert_error(body, ErrorCode.UNKNOWN_ERROR, raw=True)
    assert "результат" in body["error"]["message"]


# --- сырой ответ и нормализатор -----------------------------------------------------------------------------------
async def test_normalizer_exception_is_unknown_error_with_raw_saved(make_env: EnvFactory) -> None:
    env = await make_env(normalizer=RaisingNormalizer())
    request = env.request()
    async with emulator(env):
        body = await env.run(request)
    assert_error(body, ErrorCode.UNKNOWN_ERROR, raw=True)
    assert "нормализатор" in body["error"]["message"]
    assert (BUCKET, f"raw/{request['job_id']}.json") in env.storage.objects


async def test_raw_write_failure_is_unknown_error_without_raw_key(make_env: EnvFactory) -> None:
    normalizer = StubNormalizer(
        NormalizeOutcome(status=ResultStatus.ERROR, error=MessageError(code="UNREADABLE", message="x"))
    )
    env = await make_env(normalizer=normalizer)
    env.storage.fail_put_prefixes = ("raw/",)
    request = env.request()
    async with emulator(env):
        body = await env.run(request)
    assert_error(body, ErrorCode.UNKNOWN_ERROR, raw=False)
    assert "сырой ответ" in body["error"]["message"]
    assert normalizer.calls == []  # без сохранённого сырого ответа нормализатор не вызывается (D-4)


async def test_answer_not_json_is_invalid_format_and_raw_stored_as_is(make_env: EnvFactory) -> None:
    env = await make_env()
    request = env.request()
    job_id = request["job_id"]
    garbage = b"\xff\xfe<html>not json"
    await env.deliver(request)
    await eventually(lambda: len(env.hotfolder.puts) == 1)
    env.answer(job_id, garbage)
    await asyncio.wait_for(env.worker.wait_idle(), 5)
    [(_, body)] = env.results(job_id)
    assert_error(body, ErrorCode.INVALID_FORMAT, raw=True)
    assert env.storage.objects[(BUCKET, f"raw/{job_id}.json")] == garbage
    assert env.storage.content_types[(BUCKET, f"raw/{job_id}.json")] == OCTET_STREAM


async def test_partially_written_answer_is_reread(make_env: EnvFactory) -> None:
    """ИС пишет ответ не атомарно (SMB): недописанный JSON перечитывается через период опроса."""
    env = await make_env()
    full = json.dumps({**SUCCESS_ANSWER, "file_uid": "x"}).encode()
    env.hotfolder.poll_script = [None, [AnswerFile("x.json", full[:10])], [AnswerFile("x.json", full)]]
    body = await env.run(env.request())
    assert_error(body, ErrorCode.NOT_IMPLEMENTED, raw=True)
    assert env.storage.objects[(BUCKET, f"raw/{body['job_id']}.json")] == full


async def test_several_answer_files_first_by_name_is_used(make_env: EnvFactory) -> None:
    """Переходно (Task 3 заменит склейкой всех файлов): из готового набора берётся первый файл (по имени)."""
    env = await make_env()
    first = json.dumps({**SUCCESS_ANSWER, "file_uid": "x"}).encode()
    env.hotfolder.poll_script = [[AnswerFile("x.json", first), AnswerFile("x_ofr.json", b'{"file_uid": "second"}')]]
    body = await env.run(env.request())
    assert_error(body, ErrorCode.NOT_IMPLEMENTED, raw=True)
    assert env.storage.objects[(BUCKET, f"raw/{body['job_id']}.json")] == first


# --- таймаут ------------------------------------------------------------------------------------------------------
async def test_timeout_by_scan_timeout_keeps_file_and_never_resends(make_env: EnvFactory) -> None:
    env = await make_env(SCAN_TIMEOUT_SECONDS=0.2)
    request = env.request(original_name="aep_5.pdf", deadline_s=60)
    started = asyncio.get_running_loop().time()
    body = await env.run(request)
    elapsed = asyncio.get_running_loop().time() - started
    assert_error(body, ErrorCode.TIMEOUT, raw=False)
    assert 0.15 < elapsed < 2
    assert env.inbox() == [f"{request['job_id']}_aep_5.pdf"]  # входящий файл остаётся
    assert env.hotfolder.puts == [f"{request['job_id']}_aep_5.pdf"]  # повторной отправки нет
    assert env.hotfolder.polls >= 5


async def test_timeout_by_reply_deadline(make_env: EnvFactory) -> None:
    env = await make_env(SCAN_TIMEOUT_SECONDS=300)
    request = env.request(deadline_s=0.2)
    started = asyncio.get_running_loop().time()
    body = await env.run(request)
    assert_error(body, ErrorCode.TIMEOUT, raw=False)
    assert asyncio.get_running_loop().time() - started < 2
    assert len(env.hotfolder.puts) == 1


async def test_deadline_passed_before_sending_gives_timeout_without_file(make_env: EnvFactory) -> None:
    env = await make_env()
    request = env.request(deadline_s=-1)
    body = await env.run(request)
    assert_error(body, ErrorCode.TIMEOUT, raw=False)
    assert env.hotfolder.puts == []
    assert env.storage.get_calls == []
    assert env.acks(request["job_id"]) == [f"ack:{request['job_id']}:1"]


async def test_deadline_uses_injected_clock(make_env: EnvFactory) -> None:
    """Дедлайн считается по инжектируемым часам (R-20): часы «убежали» на час — таймаут на первом же опросе."""
    shift = {"value": timedelta(0)}

    def clock() -> datetime:
        return datetime.now(UTC) + shift["value"]

    env = await make_env(clock=clock, SCAN_TIMEOUT_SECONDS=60)
    request = env.request(deadline_s=120)
    await env.deliver(request)
    await eventually(lambda: len(env.hotfolder.puts) == 1)
    shift["value"] = timedelta(hours=1)
    await asyncio.wait_for(env.worker.wait_idle(), 2)
    [(_, body)] = env.results(request["job_id"])
    assert_error(body, ErrorCode.TIMEOUT, raw=False)


# --- ошибки транспорта --------------------------------------------------------------------------------------------
async def test_missing_hotfolder_root_is_recognition_failed(make_env: EnvFactory) -> None:
    env = await make_env(root_exists=False)
    request = env.request()
    body = await env.run(request)
    assert_error(body, ErrorCode.RECOGNITION_FAILED, raw=False)
    assert "горячая папка недоступна" in body["error"]["message"]
    assert not env.root.exists()  # корень не создаётся (R-5)
    assert env.hotfolder.healthy() is False


async def test_unusable_original_name_is_recognition_failed(make_env: EnvFactory) -> None:
    env = await make_env()
    body = await env.run(env.request(original_name=".."))
    assert_error(body, ErrorCode.RECOGNITION_FAILED, raw=False)
    assert "недопустимое имя файла" in body["error"]["message"]
    assert env.hotfolder.puts == []


async def test_transient_poll_errors_do_not_fail_the_job(make_env: EnvFactory) -> None:
    env = await make_env()
    env.hotfolder.poll_errors = [OSError(5, "STATUS_SHARING_VIOLATION"), ConnectionResetError("smb")]
    async with emulator(env):
        body = await env.run(env.request())
    assert_error(body, ErrorCode.NOT_IMPLEMENTED, raw=True)


async def test_poll_failing_until_deadline_is_recognition_failed(make_env: EnvFactory) -> None:
    env = await make_env(SCAN_TIMEOUT_SECONDS=0.1)
    env.hotfolder.poll_always = OSError(13, "STATUS_ACCESS_DENIED")
    body = await env.run(env.request())
    assert_error(body, ErrorCode.RECOGNITION_FAILED, raw=False)
    assert "горячая папка недоступна" in body["error"]["message"]


async def test_poll_failing_sometimes_then_silence_is_timeout(make_env: EnvFactory) -> None:
    env = await make_env(SCAN_TIMEOUT_SECONDS=0.1)
    env.hotfolder.poll_errors = [OSError(13, "STATUS_ACCESS_DENIED")]
    body = await env.run(env.request())
    assert_error(body, ErrorCode.TIMEOUT, raw=False)


async def test_missing_s3_object_is_file_read_error(make_env: EnvFactory) -> None:
    env = await make_env()
    request = env.request(data=None)
    body = await env.run(request)
    assert_error(body, ErrorCode.FILE_READ_ERROR, raw=False)
    assert request["s3"]["key"] in body["error"]["message"]
    assert env.hotfolder.puts == []
    assert env.acks(request["job_id"]) == [f"ack:{request['job_id']}:1"]


async def test_foreign_bucket_is_file_read_error_without_reading(make_env: EnvFactory) -> None:
    env = await make_env()
    request = env.request(bucket="other-bucket")
    body = await env.run(request)
    assert_error(body, ErrorCode.FILE_READ_ERROR, raw=False)
    assert "other-bucket" in body["error"]["message"]
    assert env.storage.get_calls == []
    assert env.acks(request["job_id"]) == [f"ack:{request['job_id']}:1"]


async def test_exception_outside_steps_is_unknown_error(make_env: EnvFactory) -> None:
    class BadOutcomeNormalizer:
        def normalize(self, raw: object) -> NormalizeOutcome:
            return "не исход"  # type: ignore[return-value]

    env = await make_env(normalizer=BadOutcomeNormalizer())
    async with emulator(env):
        body = await env.run(env.request())
    assert_error(body, ErrorCode.UNKNOWN_ERROR, raw=True)


# --- SCHEMA_MISMATCH без ack (R-16) -------------------------------------------------------------------------------
def _mutations() -> list[tuple[str, Callable[[dict[str, Any]], dict[str, Any]]]]:
    return [
        ("major_2", lambda r: {**r, "schema_version": "2.0"}),
        ("pipeline_tabular", lambda r: {**r, "pipeline": "tabular", "file_type": "XLSX"}),
        ("no_s3", lambda r: {k: v for k, v in r.items() if k != "s3"}),
        ("bad_file_type", lambda r: {**r, "file_type": "DOCX"}),
        ("naive_deadline", lambda r: {**r, "reply_deadline": "2026-10-03T12:45:00"}),
        ("attempt_zero", lambda r: {**r, "attempt": 0}),
        ("empty_name", lambda r: {**r, "original_name": ""}),
    ]


@pytest.mark.parametrize(("case", "mutate"), _mutations(), ids=[case for case, _ in _mutations()])
async def test_invalid_request_is_schema_mismatch_without_ack(
    make_env: EnvFactory, case: str, mutate: Callable[[dict[str, Any]], dict[str, Any]]
) -> None:
    env = await make_env()
    request = env.request()
    body = mutate(request)
    await env.deliver(body)
    await asyncio.wait_for(env.worker.wait_idle(), 5)
    job_id = request["job_id"]
    published = env.published()
    assert env.acks(job_id) == []
    attempt = body["attempt"] if body["attempt"] >= 1 else 1
    assert [mid for mid, _ in published] == [f"result:{job_id}:{attempt}"]
    assert_error(published[0][1], ErrorCode.SCHEMA_MISMATCH, raw=False)
    assert published[0][1]["file_id"] == request["file_id"]
    assert env.storage.get_calls == []
    assert env.hotfolder.puts == []


@pytest.mark.parametrize(
    "data",
    [
        b"not json at all",
        b"\xff\xfe\x00broken utf-8",
        b"[1, 2, 3]",
        b'"string"',
        b"[" * 100_000,  # глубокая вложенность: RecursionError парсера не роняет воркер
        json.dumps({"schema_version": "1.0", "job_id": "not-a-uuid", "file_id": str(uuid.uuid4())}).encode(),
        json.dumps({"schema_version": "1.0", "job_id": str(uuid.uuid4())}).encode(),
        json.dumps({"job_id": 5, "file_id": None}).encode(),
    ],
    ids=["text", "bad_utf8", "array", "string", "deep", "bad_job_id", "no_file_id", "wrong_types"],
)
async def test_unparsable_request_without_ids_is_dropped(
    make_env: EnvFactory, data: bytes, log_lines: list[str]
) -> None:
    env = await make_env()
    await env.deliver(data)
    await asyncio.wait_for(env.worker.wait_idle(), 5)
    assert env.client.published == []
    assert env.hotfolder.puts == []
    assert any("SCHEMA_MISMATCH" in line for line in log_lines)


@pytest.mark.parametrize(("attempt", "expected"), [(3, 3), (True, 1), ("2", 1), (-1, 1), (2**40, 1)])
async def test_schema_mismatch_attempt_from_body(make_env: EnvFactory, attempt: Any, expected: int) -> None:
    env = await make_env()
    request = env.request()
    await env.deliver({**request, "schema_version": "2.0", "attempt": attempt})
    await asyncio.wait_for(env.worker.wait_idle(), 5)
    assert [mid for mid, _ in env.published()] == [f"result:{request['job_id']}:{expected}"]


async def test_schema_mismatch_does_not_register_job_id(make_env: EnvFactory) -> None:
    """Исправленный запрос с тем же job_id после SCHEMA_MISMATCH обрабатывается (это не повтор)."""
    env = await make_env()
    request = env.request()
    await env.deliver({**request, "schema_version": "2.0"})
    async with emulator(env):
        await env.deliver(request)
        await asyncio.wait_for(env.worker.wait_idle(), 5)
    assert len(env.hotfolder.puts) == 1


async def test_schema_mismatch_for_active_or_done_job_id_is_not_published(
    make_env: EnvFactory, log_lines: list[str]
) -> None:
    """T12.1: битый запрос с job_id задания в работе или уже обработанного не публикует `result:<job>:<attempt>`.

    Иначе ядро (дедуп по `result_msg_id`) отбросило бы настоящий итог с тем же Nats-Msg-Id. UUID в теле может прийти
    в верхнем регистре — ключ дедупликации нормализуется.
    """
    env = await make_env()
    request = env.request()
    job_id = request["job_id"]
    broken = {**request, "schema_version": "2.0"}
    await env.deliver(request)
    await eventually(lambda: len(env.hotfolder.puts) == 1)  # задание ждёт ответа ИС
    await env.deliver(broken)
    await env.deliver({**broken, "job_id": job_id.upper()})
    async with emulator(env):
        await asyncio.wait_for(env.worker.wait_idle(), 5)
    await env.deliver(broken)  # задание уже обработано
    await env.deliver({**broken, "job_id": job_id.upper()})
    await asyncio.wait_for(env.worker.wait_idle(), 5)
    published = env.published()
    assert [mid for mid, _ in published] == [f"ack:{job_id}:1", f"result:{job_id}:1"]
    assert_error(published[1][1], ErrorCode.NOT_IMPLEMENTED, raw=True)
    assert sum("SCHEMA_MISMATCH для задания в работе/обработанного" in line for line in log_lines) == 4


# --- публикация, размер и кодировка -------------------------------------------------------------------------------
async def test_publish_failure_is_logged_and_worker_keeps_working(make_env: EnvFactory, log_lines: list[str]) -> None:
    env = await make_env()
    env.client.fail = True
    first = env.request()
    async with emulator(env):
        await env.deliver(first)
        await asyncio.wait_for(env.worker.wait_idle(), 5)
        assert env.client.published == []
        assert any(line.startswith("ERROR|") and first["job_id"] in line for line in log_lines)
        env.client.fail = False
        second = env.request()
        body = await env.run(second)
    assert_error(body, ErrorCode.NOT_IMPLEMENTED, raw=True)
    assert len(env.hotfolder.puts) == 2


async def test_huge_error_message_is_truncated(make_env: EnvFactory) -> None:
    huge = "я" * 2_000_000
    normalizer = StubNormalizer(
        NormalizeOutcome(status=ResultStatus.ERROR, error=MessageError(code="INVALID_DOCUMENT", message=huge))
    )
    env = await make_env(normalizer=normalizer)
    async with emulator(env):
        body = await env.run(env.request())
    assert_error(body, ErrorCode.INVALID_DOCUMENT, raw=True)
    message = body["error"]["message"]
    assert len(message) == MAX_ERROR_MESSAGE_CHARS
    assert message.endswith("…")
    assert all(len(payload) < 900_000 for _, payload, _ in env.client.published)


async def test_lone_surrogate_in_is_error_message_does_not_break_publishing(make_env: EnvFactory) -> None:
    env = await make_env()
    request = env.request()
    job_id = request["job_id"]
    await env.deliver(request)
    await eventually(lambda: len(env.hotfolder.puts) == 1)
    env.answer(job_id, b'{"file_uid": "x", "error": {"code": "INVALID_DOCUMENT", "message": "bad \\ud800 tail"}}')
    await asyncio.wait_for(env.worker.wait_idle(), 5)
    [(_, body)] = env.results(job_id)
    assert_error(body, ErrorCode.INVALID_DOCUMENT, raw=True)
    assert body["error"]["message"].startswith("bad ")
    assert body["error"]["message"].endswith(" tail")


async def test_lone_surrogate_in_original_name_is_schema_mismatch(make_env: EnvFactory) -> None:
    """Одиночный суррогат в `original_name` схема запроса не пропускает (pydantic `string_unicode`): SCHEMA_MISMATCH
    без ack и без файла в горячей папке; сам конверт публикуется."""
    env = await make_env()
    request = env.request(original_name="скан\udc80.pdf")
    data = json.dumps(request, ensure_ascii=True).encode("ascii")  # в JSON суррогат приходит экранированным
    await env.deliver(data)
    await asyncio.wait_for(env.worker.wait_idle(), 5)
    job_id = request["job_id"]
    [(_, body)] = env.results(job_id)
    assert_error(body, ErrorCode.SCHEMA_MISMATCH, raw=False)
    assert env.acks(job_id) == []
    assert env.hotfolder.puts == []


# --- остановка и прямой вызов handle (параллелизм — test_worker_concurrency.py) ---------------------------------------
async def test_stop_cancels_jobs_and_rejects_new_requests(make_env: EnvFactory) -> None:
    env = await make_env()
    request = env.request()
    await env.deliver(request)
    await eventually(lambda: len(env.hotfolder.puts) == 1)
    await asyncio.wait_for(env.worker.stop(), 1)
    assert env.worker.active_jobs == 0
    assert env.results(request["job_id"]) == []
    await env.deliver(env.request())
    await asyncio.wait_for(env.worker.wait_idle(), 1)
    assert len(env.client.published) == 1  # только ack первого запроса
    await env.worker.stop()  # повторная остановка безопасна


async def test_service_info_and_handle_returns_published_message(make_env: EnvFactory) -> None:
    from src.contract import RecognitionRequest  # noqa: PLC0415

    env = await make_env()
    request = RecognitionRequest.model_validate(env.request(data=None))
    message = await env.worker.handle(request)
    assert message.service.name == "scan-gateway"
    assert message.service.version == env.settings.VERSION
    assert message.error is not None and message.error.code is ErrorCode.FILE_READ_ERROR
    [(_, body)] = env.results(str(request.job_id))
    assert body == message.to_wire()