api/.env.example

@@ -56,7 +56,8 @@ SCAN_MAX_INFLIGHT=4
 
 # --- Эмулятор ИС Сканирование (python -m src.scan_emulator, make emulator; только dev) ---
 # Эти параметры читает сам эмулятор, сервис шлюза их не использует. Корень горячей папки и подкаталоги
-# эмулятор берёт из SCAN_HOTFOLDER_PATH / SCAN_INPUT_DIR / SCAN_OUTPUT_DIR выше.
+# эмулятор берёт из SCAN_HOTFOLDER_PATH / SCAN_INPUT_DIR / SCAN_OUTPUT_DIR выше; набор ответа на задание он собирает
+# во временном каталоге SCAN_TMP_DIR (выше) и переименовывает в <Output>/<uuid>/ целиком.
 # Каталог с фикстурами ответов ИС (aep_N.json); путь от api/ (репозиторий fixtures лежит рядом с workspace)
 SCAN_FIXTURES_DIR=../../fixtures/ocr
 # Задержка перед выдачей ответа, секунды (имитация работы ИС)

api/src/scan_emulator/__init__.py

@@ -1,15 +1,19 @@
 """Эмулятор ИС Сканирование (Content AI) для разработки и тестов (D-6, R-11, spec scan-emulator).
 
 Воспроизводит протокол горячих папок: для нового файла `<uuid>_<name>.<ext>` во входящем каталоге через задержку кладёт
-`<output>/<uuid>/<uuid>.json` с эталонным ответом из `SCAN_FIXTURES_DIR` и переименовывает входящий файл в `….done`.
-Эмулятор читает собственные переменные окружения (`EmulatorSettings`), а не настройки шлюза: в его контейнере нет
-переменных MinIO/NATS. С каталогами работает через `LocalHotFolder`, тем самым тестируя и его.
+каталог `<output>/<uuid>/` с набором эталонных ответов из `SCAN_FIXTURES_DIR` (все `<name>.json` -> `<uuid>.json` и
+`<name>_<suffix>.json` -> `<uuid>_<suffix>.json`) и переименовывает входящий файл в `….done`. Набор собирается во
+временном каталоге служебной папки (`SCAN_TMP_DIR`, та же ФС, что и `<output>`) и появляется в `<output>` одним rename
+каталога: шлюз не видит частичного набора. Эмулятор читает собственные переменные окружения (`EmulatorSettings`), а не
+настройки шлюза: в его контейнере нет переменных MinIO/NATS. С каталогами работает через `LocalHotFolder`,
+тем самым тестируя и его.
 """
 
 import json
 import math
 import os
 import re
+import shutil
 import threading
 import time
 from collections.abc import Mapping
@@ -54,6 +58,7 @@ class EmulatorSettings:
     hotfolder_path: Path
     input_dir: str = "Input"
     output_dir: str = "Output"
+    tmp_dir: str = "Tmp"
     fixtures_dir: Path = Path("../../fixtures/ocr")
     delay: float = 2.0
     poll_interval: float = 1.0
@@ -63,8 +68,8 @@ class EmulatorSettings:
         """Разбор окружения. Без аргумента: `os.environ` поверх файла `ENV_FILE` (по умолчанию `.env`, если он есть).
 
         `SCAN_HOTFOLDER_PATH` обязателен (`ValueError`); `SCAN_INPUT_DIR`=Input, `SCAN_OUTPUT_DIR`=Output,
-        `SCAN_FIXTURES_DIR`=../../fixtures/ocr, `SCAN_EMULATOR_DELAY`=2 (с, >= 0),
-        `SCAN_EMULATOR_POLL_SECONDS`=1 (с, > 0).
+        `SCAN_TMP_DIR`=Tmp (служебный каталог для сборки набора ответа), `SCAN_FIXTURES_DIR`=../../fixtures/ocr,
+        `SCAN_EMULATOR_DELAY`=2 (с, >= 0), `SCAN_EMULATOR_POLL_SECONDS`=1 (с, > 0).
         """
         if environ is None:
             env_file = os.environ.get("ENV_FILE", ".env")
@@ -79,74 +84,125 @@ class EmulatorSettings:
             hotfolder_path=Path(path),
             input_dir=env.get("SCAN_INPUT_DIR") or "Input",
             output_dir=env.get("SCAN_OUTPUT_DIR") or "Output",
+            tmp_dir=env.get("SCAN_TMP_DIR") or "Tmp",
             fixtures_dir=Path(env.get("SCAN_FIXTURES_DIR") or "../../fixtures/ocr"),
             delay=_read_float(env, "SCAN_EMULATOR_DELAY", 2.0, positive=False),
             poll_interval=_read_float(env, "SCAN_EMULATOR_POLL_SECONDS", 1.0, positive=True),
         )
 
 
+_EMPTY_FIELD: dict[str, Any] = {"form_name": "", "confidence": 0}
+
+
 def _error_answer(job_id: str, message: str) -> dict[str, Any]:
-    return {"file_uid": job_id, "error": {"code": _ERROR_CODE, "message": message}, "data_ocr": []}
+    """Ответ-ошибка в форме реальных ответов Content AI (`aep_5.json`): `data_ocr` — объект с пустыми полями (R-28)."""
+    data_ocr: dict[str, Any] = {
+        field: dict(_EMPTY_FIELD) for field in ("label", "organization", "inn", "units", "date")
+    }
+    data_ocr["table_structure"] = []
+    return {"file_uid": job_id, "error": {"code": _ERROR_CODE, "message": message}, "data_ocr": data_ocr}
 
 
-def _find_fixture(fixtures: Path, stem: str) -> Path | None:
-    """`<stem>.json`, затем `<stem>*.json` (первый по алфавиту). Имя сравнивается буквально, без glob."""
+def _select_fixtures(fixtures: Path, stem: str) -> list[tuple[str, Path]]:
+    """Фикстуры задания: `<stem>.json` и `<stem>_*.json` -> `(suffix, путь)`, `suffix` пуст для `<stem>.json`.
+
+    Имя сравнивается буквально, без glob; расширение `.json` — без учёта регистра. Не префикс `<stem>*`: `aep_30.json`
+    не входит в набор `aep_3`. Результат отсортирован по имени файла; при совпадении `suffix` берётся первый файл.
+    """
+    if not stem:
+        return []
     try:
         names = sorted(p.name for p in fixtures.iterdir() if p.name.lower().endswith(".json") and p.is_file())
     except OSError:
-        return None
-    if f"{stem}.json" in names:
-        return fixtures / f"{stem}.json"
-    for name in names:
-        if name.startswith(stem):
-            return fixtures / name
-    return None
+        return []
+    selected: dict[str, Path] = {}
+    for fixture_name in names:
+        base = fixture_name[: -len(".json")]
+        if base == stem:
+            suffix = ""
+        elif base.startswith(f"{stem}_") and len(base) > len(stem) + 1:
+            suffix = base[len(stem) + 1 :]
+        else:
+            continue
+        selected.setdefault(suffix, fixtures / fixture_name)
+    return list(selected.items())
 
 
-def _build_answer(job_id: str, name: str, fixtures: Path) -> dict[str, Any]:
-    """Ответ ИС на файл `name` (без префикса uuid): фикстура с подставленным `file_uid` либо ошибка Content AI."""
-    fixture = _find_fixture(fixtures, Path(name).stem)
-    if fixture is None:
-        return _error_answer(job_id, f"{name}: нет эталонного ответа в фикстурах")
+def _answer_file_name(job_id: str, suffix: str) -> str:
+    return f"{job_id}_{suffix}.json" if suffix else f"{job_id}.json"
+
+
+def _build_answers(job_id: str, name: str, fixtures: Path) -> dict[str, dict[str, Any]]:
+    """Набор ответов ИС на файл `name` (без префикса uuid): `{имя файла ответа: ответ}`.
+
+    Каждый ответ — фикстура с подставленным `file_uid`; повреждённая фикстура даёт на своём месте ответ-ошибку; если
+    фикстур нет совсем — один ответ `<uuid>.json` с `INVALID_DOCUMENT`.
+    """
+    selected = _select_fixtures(fixtures, Path(name).stem)
+    if not selected:
+        return {_answer_file_name(job_id, ""): _error_answer(job_id, f"{name}: нет эталонного ответа в фикстурах")}
+    answers: dict[str, dict[str, Any]] = {}
+    for suffix, fixture in selected:
+        try:
+            data = json.loads(fixture.read_text(encoding="utf-8"))
+            if not isinstance(data, dict):
+                raise ValueError("корень не объект JSON")
+        except (OSError, ValueError) as exc:  # json.JSONDecodeError и UnicodeDecodeError — подклассы ValueError
+            logger.error("Эмулятор: фикстура {} повреждена ({}); для {} отдаём {}", fixture, exc, name, _ERROR_CODE)
+            answers[_answer_file_name(job_id, suffix)] = _error_answer(
+                job_id, f"{name}: эталонный ответ {fixture.name} повреждён и не разобран"
+            )
+            continue
+        answers[_answer_file_name(job_id, suffix)] = {
+            "file_uid": job_id,
+            **{key: value for key, value in data.items() if key != "file_uid"},
+        }
+    return answers
+
+
+def _write_json(path: Path, answer: dict[str, Any]) -> None:
+    path.write_text(json.dumps(answer, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+
+
+def _publish_answers(hotfolder: LocalHotFolder, job_id: str, answers: dict[str, dict[str, Any]]) -> None:
+    """Атомарная публикация набора (R-26): файлы пишутся в `<tmp>/.<uuid>.answer`, затем каталог переименовывается в
+    `<output>/<uuid>/`. Служебный и исходящий каталоги лежат в одном корне (одна ФС), поэтому rename атомарен, а
+    каталог ответа либо отсутствует, либо содержит весь набор. Каталог ответа, оставшийся от прошлого запуска,
+    заменяется (единственный момент, когда его на время нет). При сбое временный каталог удаляется."""
+    target = hotfolder.answer_dir(job_id)
+    hotfolder.tmp_dir.mkdir(exist_ok=True)
+    hotfolder.output_dir.mkdir(parents=True, exist_ok=True)
+    staging = hotfolder.tmp_dir / f".{job_id}.answer"
+    shutil.rmtree(staging, ignore_errors=True)  # остаток прерванной сборки
     try:
-        data = json.loads(fixture.read_text(encoding="utf-8"))
-        if not isinstance(data, dict):
-            raise ValueError("корень не объект JSON")
-    except (OSError, ValueError) as exc:  # json.JSONDecodeError и UnicodeDecodeError — подклассы ValueError
-        logger.error("Эмулятор: фикстура {} повреждена ({}); для {} отдаём {}", fixture, exc, name, _ERROR_CODE)
-        return _error_answer(job_id, f"{name}: эталонный ответ {fixture.name} повреждён и не разобран")
-    return {"file_uid": job_id, **{key: value for key, value in data.items() if key != "file_uid"}}
-
-
-def _write_answer(hotfolder: LocalHotFolder, job_id: str, answer: dict[str, Any]) -> None:
-    """Атомарная запись ответа: `.part` в том же каталоге, затем `os.replace`."""
-    target = hotfolder.answer_path(job_id)
-    target.parent.mkdir(parents=True, exist_ok=True)
-    part = target.parent / f".{target.name}.part"
-    try:
-        part.write_text(json.dumps(answer, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
-        os.replace(part, target)
+        staging.mkdir()
+        for file_name, answer in answers.items():
+            _write_json(staging / file_name, answer)
+        if target.exists():
+            shutil.rmtree(target)
+        os.replace(staging, target)
     except BaseException:
-        part.unlink(missing_ok=True)
+        shutil.rmtree(staging, ignore_errors=True)
         raise
 
 
-def process_once(
+def process_once(  # noqa: PLR0913 — корень, фикстуры, задержка + имена трёх служебных каталогов
     root: Path | str,
     fixtures: Path | str,
     delay: float,
     *,
     input_dir: str = "Input",
     output_dir: str = "Output",
+    tmp_dir: str = "Tmp",
 ) -> list[str]:
     """Один проход по входящему каталогу; возвращает uuid обработанных файлов.
 
     Берёт новые `<uuid>_<name>.<ext>` (не dot-файлы, не `*.done`, не каталоги), при наличии работы один раз за проход
-    ждёт `delay` секунд, для каждого файла пишет `<output>/<uuid>/<uuid>.json` и только затем переименовывает входящий
-    файл в `….done`. Имена не по шаблону `<uuid>_` игнорируются (одно предупреждение за проход). Нет входящего
-    каталога — пустой результат.
+    ждёт `delay` секунд, для каждого файла публикует набор ответов `<output>/<uuid>/` (атомарно, через `tmp_dir`) и
+    только затем переименовывает входящий файл в `….done`. Имена не по шаблону `<uuid>_` игнорируются (одно
+    предупреждение за проход). Нет входящего каталога — пустой результат.
     """
-    hotfolder = LocalHotFolder(root, input_dir, output_dir)
+    hotfolder = LocalHotFolder(root, input_dir, output_dir, tmp_dir)
     fixtures_dir = Path(fixtures)
     try:
         entries = sorted(hotfolder.input_dir.iterdir(), key=lambda p: p.name)
@@ -171,13 +227,13 @@ def process_once(
     processed: list[str] = []
     for entry, job_id, name in pending:
         try:
-            _write_answer(hotfolder, job_id, _build_answer(job_id, name, fixtures_dir))
+            _publish_answers(hotfolder, job_id, _build_answers(job_id, name, fixtures_dir))
             os.replace(entry, entry.with_name(entry.name + _DONE_SUFFIX))
         except FileNotFoundError:
             logger.warning("Эмулятор: файл {} исчез во время обработки", entry.name)
             continue
         processed.append(job_id)
-        logger.info("Эмулятор: {} -> {}", entry.name, hotfolder.answer_path(job_id))
+        logger.info("Эмулятор: {} -> {}", entry.name, hotfolder.answer_dir(job_id))
     return processed
 
 
@@ -188,7 +244,7 @@ def _count_fixtures(fixtures: Path) -> int:
 def run(settings: EmulatorSettings, stop: threading.Event, *, max_iterations: int | None = None) -> None:
     """Цикл эмулятора: `process_once` каждые `poll_interval` секунд, пока не взведён `stop` (или не выполнено
     `max_iterations` проходов — для тестов). Сбой прохода логируется и не останавливает цикл."""
-    hotfolder = LocalHotFolder(settings.hotfolder_path, settings.input_dir, settings.output_dir)
+    hotfolder = LocalHotFolder(settings.hotfolder_path, settings.input_dir, settings.output_dir, settings.tmp_dir)
     logger.info(
         "Эмулятор ИС Сканирование: корень {}, входящий каталог {}, исходящий каталог {}, задержка {} с, опрос {} с",
         hotfolder.root,
@@ -212,6 +268,7 @@ def run(settings: EmulatorSettings, stop: threading.Event, *, max_iterations: in
                 settings.delay,
                 input_dir=settings.input_dir,
                 output_dir=settings.output_dir,
+                tmp_dir=settings.tmp_dir,
             )
         except Exception:  # noqa: BLE001 — эмулятор живёт, пока его не остановят
             logger.exception("Эмулятор: сбой прохода по входящему каталогу")

api/tests/unit/test_emulator.py

@@ -1,7 +1,8 @@
 """Эмулятор ИС Сканирование (spec scan-emulator, D-6, R-11, slice-0-skeleton).
 
-Сценарии spec «Точное имя», «Префикс», «Нет пары» — на копиях реальных `../fixtures/ocr/*.json` (маркер `fixtures`);
-остальное — на синтетических фикстурах в `tmp_path`, без внешних репозиториев и без Docker.
+Сценарии spec «Точное имя», «Суффикс», «Несколько файлов на одно задание», «Нет пары» — на копиях реальных
+`../fixtures/ocr/*.json` (маркер `fixtures`); остальное — на синтетических фикстурах в `tmp_path`, без внешних
+репозиториев и без Docker.
 """
 
 import json
@@ -18,7 +19,10 @@ from typing import Any
 import pytest
 from loguru import logger
 
+import src.scan_emulator as emulator
 from src.gateway.hotfolder import LocalHotFolder, inbox_name
+from src.gateway.merge import merge_answers
+from src.normalizer import NullNormalizer
 from src.scan_emulator import EmulatorSettings, process_once, run
 from src.scan_emulator.__main__ import main
 
@@ -53,11 +57,28 @@ def _drop(hf: LocalHotFolder, name: str, data: bytes = b"%PDF") -> str:
     return job_id
 
 
-def _answer(hf: LocalHotFolder, job_id: str) -> dict[str, Any]:
-    result: dict[str, Any] = json.loads(hf.answer_path(job_id).read_text(encoding="utf-8"))
+def _answer(hf: LocalHotFolder, job_id: str, name: str | None = None) -> dict[str, Any]:
+    """Файл ответа каталога `Output/<uuid>/` (по умолчанию основной `<uuid>.json`)."""
+    path = hf.answer_path(job_id) if name is None else hf.answer_dir(job_id) / name
+    result: dict[str, Any] = json.loads(path.read_text(encoding="utf-8"))
     return result
 
 
+def _names(hf: LocalHotFolder, job_id: str) -> list[str]:
+    return sorted(p.name for p in hf.answer_dir(job_id).iterdir())
+
+
+# форма `data_ocr` реального ответа-ошибки ИС (aep_5.json); R-28
+EMPTY_DATA_OCR: dict[str, Any] = {
+    "label": {"form_name": "", "confidence": 0},
+    "organization": {"form_name": "", "confidence": 0},
+    "inn": {"form_name": "", "confidence": 0},
+    "units": {"form_name": "", "confidence": 0},
+    "date": {"form_name": "", "confidence": 0},
+    "table_structure": [],
+}
+
+
 @pytest.fixture
 def log_messages() -> Iterator[list[str]]:
     messages: list[str] = []
@@ -90,43 +111,102 @@ class TestRealFixtures:
         assert (hf.input_dir / f"{job_id}_{name}.done").is_file()
         assert not (hf.input_dir / f"{job_id}_{name}").exists()
 
-    def test_prefix(self, tmp_path: Path, real_fixtures: Path) -> None:
+    def test_suffix(self, tmp_path: Path, real_fixtures: Path) -> None:
+        """`aep_8.pdf` -> `aep_8_form4.json` как `<uuid>_form4.json` (раньше — `<uuid>.json`, правило `<name>*`)."""
         hf = _hotfolder(tmp_path)
         job_id = _drop(hf, "aep_8.pdf")
         process_once(hf.root, real_fixtures, 0)
         expected = json.loads((real_fixtures / "aep_8_form4.json").read_text(encoding="utf-8"))
         expected["file_uid"] = job_id
-        assert _answer(hf, job_id) == expected
+        assert _names(hf, job_id) == [f"{job_id}_form4.json"]
+        assert _answer(hf, job_id, f"{job_id}_form4.json") == expected
         assert (hf.input_dir / f"{job_id}_aep_8.pdf.done").is_file()
 
+    def test_two_files_for_one_job(self, tmp_path: Path, real_fixtures: Path) -> None:
+        hf = _hotfolder(tmp_path)
+        job_id = _drop(hf, "aep_3.pdf")
+        assert process_once(hf.root, real_fixtures, 0) == [job_id]
+        assert _names(hf, job_id) == [f"{job_id}.json", f"{job_id}_ofr.json"]
+        main = json.loads((real_fixtures / "aep_3.json").read_text(encoding="utf-8"))
+        ofr = json.loads((real_fixtures / "aep_3_ofr.json").read_text(encoding="utf-8"))
+        assert main["error"] == {"code": "", "message": ""}
+        assert ofr["error"]["code"] == "INVALID_DOCUMENT"
+        assert _answer(hf, job_id) == {**main, "file_uid": job_id}
+        assert _answer(hf, job_id, f"{job_id}_ofr.json") == {**ofr, "file_uid": job_id}
+        assert (hf.input_dir / f"{job_id}_aep_3.pdf.done").is_file()
+
+    def test_single_success_file(self, tmp_path: Path, real_fixtures: Path) -> None:
+        hf = _hotfolder(tmp_path)
+        job_id = _drop(hf, "aep_6.pdf")
+        process_once(hf.root, real_fixtures, 0)
+        assert _names(hf, job_id) == [f"{job_id}.json"]
+        expected = json.loads((real_fixtures / "aep_6.json").read_text(encoding="utf-8"))
+        assert _answer(hf, job_id) == {**expected, "file_uid": job_id}
+
     def test_no_pair(self, tmp_path: Path, real_fixtures: Path) -> None:
         hf = _hotfolder(tmp_path)
         job_id = _drop(hf, "unknown.pdf")
         process_once(hf.root, real_fixtures, 0)
+        assert _names(hf, job_id) == [f"{job_id}.json"]
         assert _answer(hf, job_id) == {
             "file_uid": job_id,
             "error": {"code": "INVALID_DOCUMENT", "message": "unknown.pdf: нет эталонного ответа в фикстурах"},
-            "data_ocr": [],
+            "data_ocr": EMPTY_DATA_OCR,
         }
         assert (hf.input_dir / f"{job_id}_unknown.pdf.done").is_file()
 
+    def test_no_pair_has_the_shape_of_real_error_answers(self, tmp_path: Path, real_fixtures: Path) -> None:
+        """R-28: форма `data_ocr` совпадает с реальным ответом-ошибкой `aep_5.json` (кроме `0` вместо `0.0`)."""
+        hf = _hotfolder(tmp_path)
+        job_id = _drop(hf, "unknown.pdf")
+        process_once(hf.root, real_fixtures, 0)
+        real = json.loads((real_fixtures / "aep_5.json").read_text(encoding="utf-8"))
+        assert _answer(hf, job_id)["data_ocr"] == real["data_ocr"]
+        assert list(_answer(hf, job_id)["data_ocr"]) == list(real["data_ocr"])
+
 
 class TestFixtureLookup:
-    def test_exact_beats_prefix(self, tmp_path: Path) -> None:
+    def test_name_with_other_characters_is_not_taken(self, tmp_path: Path) -> None:
+        """В набор входят `<name>.json` и `<name>_*.json`; `a_b-1.json` (нет `_` после имени) — нет."""
         hf = _hotfolder(tmp_path)
-        # `a_b-1.json` по алфавиту раньше `a_b.json` ('-' 0x2D < '.' 0x2E): без шага «точное имя» победил бы префикс
-        assert sorted(["a_b.json", "a_b-1.json"])[0] == "a_b-1.json"
-        fixtures = _fixtures_dir(tmp_path, {"a_b-1.json": {"v": "prefix"}, "a_b.json": {"file_uid": "x", "v": "exact"}})
+        fixtures = _fixtures_dir(tmp_path, {"a_b-1.json": {"v": "dash"}, "a_b.json": {"file_uid": "x", "v": "exact"}})
         job_id = _drop(hf, "a_b.pdf")
         process_once(hf.root, fixtures, 0)
+        assert _names(hf, job_id) == [f"{job_id}.json"]
         assert _answer(hf, job_id) == {"file_uid": job_id, "v": "exact"}
 
-    def test_prefix_first_alphabetical(self, tmp_path: Path) -> None:
+    def test_all_suffix_fixtures_are_taken(self, tmp_path: Path) -> None:
         hf = _hotfolder(tmp_path)
         fixtures = _fixtures_dir(tmp_path, {"a_zz.json": {"v": "zz"}, "a_b.json": {"v": "b"}, "a_c.json": {"v": "c"}})
         job_id = _drop(hf, "a.pdf")
         process_once(hf.root, fixtures, 0)
-        assert _answer(hf, job_id)["v"] == "b"
+        assert _names(hf, job_id) == [f"{job_id}_b.json", f"{job_id}_c.json", f"{job_id}_zz.json"]
+        assert [_answer(hf, job_id, f"{job_id}_{s}.json")["v"] for s in ("b", "c", "zz")] == ["b", "c", "zz"]
+
+    def test_foreign_prefix_is_not_taken(self, tmp_path: Path) -> None:
+        """`aep_3` не забирает `aep_30*`: правило `<name>.json` и `<name>_*.json`, а не `<name>*`."""
+        hf = _hotfolder(tmp_path)
+        fixtures = _fixtures_dir(
+            tmp_path,
+            {
+                "aep_3.json": {"v": "main"},
+                "aep_3_ofr.json": {"v": "ofr"},
+                "aep_30.json": {"v": 30},
+                "aep_30_x.json": {"v": 31},
+            },
+        )
+        job_id = _drop(hf, "aep_3.pdf")
+        process_once(hf.root, fixtures, 0)
+        assert _names(hf, job_id) == [f"{job_id}.json", f"{job_id}_ofr.json"]
+        assert _answer(hf, job_id)["v"] == "main"
+        assert _answer(hf, job_id, f"{job_id}_ofr.json")["v"] == "ofr"
+
+    def test_underscore_without_suffix_is_not_taken(self, tmp_path: Path) -> None:
+        hf = _hotfolder(tmp_path)
+        fixtures = _fixtures_dir(tmp_path, {"a_.json": {"v": 1}})
+        job_id = _drop(hf, "a.pdf")
+        process_once(hf.root, fixtures, 0)
+        assert _answer(hf, job_id)["error"]["code"] == "INVALID_DOCUMENT"
 
     def test_file_uid_added_when_absent_and_replaced_when_present(self, tmp_path: Path) -> None:
         hf = _hotfolder(tmp_path)
@@ -171,7 +251,7 @@ class TestFixtureLookup:
         assert process_once(hf.root, fixtures, 0) == [job_id]
         answer = _answer(hf, job_id)
         assert answer["file_uid"] == job_id
-        assert answer["data_ocr"] == []
+        assert answer["data_ocr"] == EMPTY_DATA_OCR
         assert answer["error"]["code"] == "INVALID_DOCUMENT"
         assert "a.pdf" in answer["error"]["message"]
         assert "a.json" in answer["error"]["message"]
@@ -283,6 +363,120 @@ class TestProcessOnce:
         assert json.loads(files[0].data) == {"file_uid": job_id, "data_ocr": [{"k": "v"}]}
 
 
+class TestAnswerSet:
+    def _two_file_job(self, tmp_path: Path) -> tuple[LocalHotFolder, Path, str]:
+        hf = _hotfolder(tmp_path)
+        fixtures = _fixtures_dir(tmp_path, {"a.json": {"v": "main"}, "a_ofr.json": {"v": "ofr"}})
+        return hf, fixtures, _drop(hf, "a.pdf")
+
+    def test_directory_appears_only_by_rename_of_complete_set(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """R-26: до rename каталога `Output/<uuid>/` нет вовсе, набор уже собран во временном каталоге `Tmp`;
+        шлюз (`poll`) в это время ничего не видит, после rename — видит оба файла сразу."""
+        hf, fixtures, job_id = self._two_file_job(tmp_path)
+        gateway_side = LocalHotFolder(hf.root, settle_seconds=0)
+        seen: dict[str, Any] = {}
+        original_replace = os.replace
+
+        def spy(src: Path, dst: Path) -> None:
+            if Path(dst) == hf.answer_dir(job_id):
+                seen["output_entries_before"] = sorted(p.name for p in hf.output_dir.iterdir())
+                seen["poll_before"] = gateway_side.poll(job_id)
+                seen["staging_parent"] = Path(src).parent
+                seen["staging_files"] = sorted(p.name for p in Path(src).iterdir())
+            original_replace(src, dst)
+
+        monkeypatch.setattr("src.scan_emulator.os.replace", spy)
+        process_once(hf.root, fixtures, 0)
+        assert seen["output_entries_before"] == []
+        assert seen["poll_before"] is None
+        assert seen["staging_parent"] == hf.tmp_dir
+        assert seen["staging_files"] == [f"{job_id}.json", f"{job_id}_ofr.json"]
+        after = gateway_side.poll(job_id)
+        assert after is not None
+        assert [f.name for f in after] == [f"{job_id}.json", f"{job_id}_ofr.json"]
+        assert list(hf.tmp_dir.iterdir()) == []  # временный каталог переименован, не скопирован
+
+    def test_failure_while_writing_leaves_no_answer_dir_and_no_done(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        hf, fixtures, job_id = self._two_file_job(tmp_path)
+        calls: list[Path] = []
+        original = emulator._write_json
+
+        def flaky(path: Path, answer: dict[str, Any]) -> None:
+            calls.append(path)
+            if len(calls) == 2:
+                raise OSError("диск переполнен")
+            original(path, answer)
+
+        monkeypatch.setattr(emulator, "_write_json", flaky)
+        with pytest.raises(OSError, match="диск"):
+            process_once(hf.root, fixtures, 0)
+        assert len(calls) == 2
+        assert not hf.answer_dir(job_id).exists()
+        assert list(hf.tmp_dir.iterdir()) == []
+        assert (hf.input_dir / f"{job_id}_a.pdf").is_file()
+        assert not (hf.input_dir / f"{job_id}_a.pdf.done").exists()
+        monkeypatch.undo()
+        assert process_once(hf.root, fixtures, 0) == [job_id]  # следующий проход обрабатывает задание целиком
+        assert _names(hf, job_id) == [f"{job_id}.json", f"{job_id}_ofr.json"]
+
+    def test_stale_staging_dir_is_discarded(self, tmp_path: Path) -> None:
+        hf, fixtures, job_id = self._two_file_job(tmp_path)
+        stale = hf.tmp_dir / f".{job_id}.answer"
+        stale.mkdir(parents=True)
+        (stale / "junk.json").write_text("{}")
+        process_once(hf.root, fixtures, 0)
+        assert _names(hf, job_id) == [f"{job_id}.json", f"{job_id}_ofr.json"]
+        assert list(hf.tmp_dir.iterdir()) == []
+
+    def test_existing_answer_dir_is_replaced(self, tmp_path: Path) -> None:
+        hf, fixtures, job_id = self._two_file_job(tmp_path)
+        hf.answer_dir(job_id).mkdir(parents=True)
+        (hf.answer_dir(job_id) / "old.json").write_text("{}")
+        process_once(hf.root, fixtures, 0)
+        assert _names(hf, job_id) == [f"{job_id}.json", f"{job_id}_ofr.json"]
+
+    def test_custom_tmp_dir_name(self, tmp_path: Path) -> None:
+        root = tmp_path / "hf"
+        root.mkdir()
+        hf = LocalHotFolder(root, tmp_dir="Work")
+        fixtures = _fixtures_dir(tmp_path, {"a.json": {"v": 1}})
+        job_id = _drop(hf, "a.pdf")
+        assert process_once(root, fixtures, 0, tmp_dir="Work") == [job_id]
+        assert (root / "Work").is_dir()
+        assert not (root / "Tmp").exists() or list((root / "Tmp").iterdir()) == []
+
+    def test_gateway_poll_waits_for_settle_then_returns_whole_set(self, tmp_path: Path) -> None:
+        """Сквозной «put -> эмулятор -> poll (settle, ложные часы) -> merge_answers -> NullNormalizer» на `aep_3`."""
+        if not REAL_FIXTURES.is_dir():
+            pytest.skip("fixtures недоступен: положите репозиторий fixtures рядом с workspace")
+        root = tmp_path / "hotfolder"
+        root.mkdir()
+        now = [100.0]
+        hf = LocalHotFolder(root, settle_seconds=2, clock=lambda: now[0])
+        job_id = _uid()
+        assert hf.poll(job_id) is None
+        hf.put(inbox_name(job_id, "aep_3.pdf"), b"%PDF-1.4")
+        assert process_once(root, REAL_FIXTURES, 0) == [job_id]
+        assert hf.poll(job_id) is None  # набор впервые замечен: settle отсчитывается с этого опроса
+        now[0] += 1.9
+        assert hf.poll(job_id) is None
+        now[0] += 0.2
+        files = hf.poll(job_id)
+        assert files is not None
+        assert [f.name for f in files] == [f"{job_id}.json", f"{job_id}_ofr.json"]
+        merged = merge_answers(job_id, files)
+        outcome = NullNormalizer().normalize(merged)
+        assert outcome.error is not None
+        assert outcome.error.code == "NOT_IMPLEMENTED"
+        assert outcome.warnings is not None
+        assert len(outcome.warnings) == 1
+        assert outcome.warnings[0].startswith(f"{job_id}_ofr.json: INVALID_DOCUMENT: ")
+
+
 class TestSettings:
     def test_defaults(self, tmp_path: Path) -> None:
         settings = EmulatorSettings.from_env({"SCAN_HOTFOLDER_PATH": str(tmp_path)})
@@ -290,6 +484,7 @@ class TestSettings:
         assert (settings.input_dir, settings.output_dir) == ("Input", "Output")
         assert settings.delay == 2
         assert settings.poll_interval == 1
+        assert settings.tmp_dir == "Tmp"
         assert settings.fixtures_dir == Path("../../fixtures/ocr")
 
     def test_overrides(self, tmp_path: Path) -> None:
@@ -298,12 +493,14 @@ class TestSettings:
                 "SCAN_HOTFOLDER_PATH": str(tmp_path),
                 "SCAN_INPUT_DIR": "In",
                 "SCAN_OUTPUT_DIR": "Out",
+                "SCAN_TMP_DIR": "Work",
                 "SCAN_FIXTURES_DIR": "fx",
                 "SCAN_EMULATOR_DELAY": "0.5",
                 "SCAN_EMULATOR_POLL_SECONDS": "0.1",
             }
         )
         assert (settings.input_dir, settings.output_dir) == ("In", "Out")
+        assert settings.tmp_dir == "Work"
         assert settings.fixtures_dir == Path("fx")
         assert (settings.delay, settings.poll_interval) == (0.5, 0.1)
 

docs/superpowers/plans/2026-10-04-multi-file-response.md

@@ -45,7 +45,7 @@
 
 ### Task 4: [R4] Эмулятор: набор файлов на задание
 
-- [ ] `src/scan_emulator/`: по входящему `<uuid>_<name>.<ext>` собрать из `SCAN_FIXTURES_DIR` все `<name>.json` и `<name>_*.json` (точное правило из Global Constraints), записать в `Output/<uuid>/` под именами `<uuid>.json` / `<uuid>_<suffix>.json` с `file_uid = <uuid>`; пары нет — прежний `INVALID_DOCUMENT`; `aep_3` не забирает `aep_30*`. Тесты (маркер `fixtures`, `tmp_path`): `aep_3.pdf` → два файла (`<uuid>.json` успех, `<uuid>_ofr.json` ошибка), `aep_6.pdf` → один, `aep_5.PDF` → один с ошибкой, `aep_8.pdf` → `aep_8_form4.json` как раньше (правило `<name>_*.json` покрывает `aep_8_form4`), неизвестное имя → `INVALID_DOCUMENT`; сквозной тест «put → эмулятор → poll (с settle) → merge» на `aep_3`. Проверка: `make check` зелёный.
+- [x] `src/scan_emulator/`: по входящему `<uuid>_<name>.<ext>` собрать из `SCAN_FIXTURES_DIR` все `<name>.json` и `<name>_*.json` (точное правило из Global Constraints), записать в `Output/<uuid>/` под именами `<uuid>.json` / `<uuid>_<suffix>.json` с `file_uid = <uuid>`; пары нет — прежний `INVALID_DOCUMENT`; `aep_3` не забирает `aep_30*`. Тесты (маркер `fixtures`, `tmp_path`): `aep_3.pdf` → два файла (`<uuid>.json` успех, `<uuid>_ofr.json` ошибка), `aep_6.pdf` → один, `aep_5.PDF` → один с ошибкой, `aep_8.pdf` → `aep_8_form4.json` как раньше (правило `<name>_*.json` покрывает `aep_8_form4`), неизвестное имя → `INVALID_DOCUMENT`; сквозной тест «put → эмулятор → poll (с settle) → merge» на `aep_3`. Проверка: `make check` зелёный.
 
 ### Task 5: [R5] Интеграционные тесты и документация
 

openspec/changes/slice-0-skeleton/design.md

@@ -210,3 +210,4 @@ warnings; `aep_5.PDF` → `INVALID_DOCUMENT` с `raw_s3_key`; объект `raw/
 | 72 | 2026-10-04 | R-33 (design #72): `warnings` исхода дописываются в конец `metrics.warnings` исхода; итоговые `metrics` одинаковы в конверте и в файле результата (done/needs_check); пустой список не порождает `metrics` | одно место правды для предупреждений; файл результата и конверт не расходятся | нет |
 | 73 | 2026-10-04 | T2.1, T2.2, T2.4 (design #73): поправки по ревью Task 2 — T2.1: после чтения набора каталог ответа листается повторно (Local и SMB), снимок отличается от настоявшегося → `None` (набор настаивается заново); T2.2: `SmbHotFolder` — ENOTDIR из листинга `Output/<job_id>` → `None`; T2.4: тест сброса состояния settle при ENOENT/SHARING_VIOLATION/ENOTDIR | файл, появившийся или пропавший за время чтения, иначе дал бы неполный или устаревший набор; поведение SMB = Local | один лишний листинг на каждый выданный набор |
 | 74 | 2026-10-04 | R-34 (design #74): интеграционный `tests/integration/test_roundtrip.py` минимально переведён на формат склейки `raw/` и успех `error = {"code": "", "message": ""}` (без новых сценариев — они в 7.4) | иначе он проверял бы старый формат `raw/`; полноценная переработка — задача 7.4 | не прогнан: в поднятом compose нет бакета `arm-ka` (`NoSuchBucket`), проверка — 7.4 после `make up` |
+| 75 | 2026-10-04 | R-35 (design #75): решения исполнителя Task 4 (7.3): эмулятор читает `SCAN_TMP_DIR` (по умолчанию `Tmp`) через `EmulatorSettings.tmp_dir`, как шлюз, не завися от настроек шлюза; набор собирается в `<root>/<SCAN_TMP_DIR>/.<uuid>.answer` и целиком переименовывается в `Output/<uuid>/`; остаток прерванной сборки удаляется, ответ с тем же uuid, уже лежащий в `Output`, заменяется (rmtree + rename); повреждённая фикстура набора даёт ответ-ошибку `INVALID_DOCUMENT` на своём месте, остальные файлы набора остаются; фикстуры с одним `suffix` (`a.JSON`/`a.json`) — берётся первая по имени; `aep_8.pdf` теперь даёт `<uuid>_form4.json`, а не `<uuid>.json` (прежнее правило `<name>*`) | spec scan-emulator: набор появляется атомарно, rename каталога возможен только в пределах одной ФС, служебная папка корня уже есть у шлюза (`Tmp`) | повторная публикация uuid на миг убирает каталог; потребители, ждавшие `aep_8` как `<uuid>.json`, должны брать набор целиком |

openspec/changes/slice-0-skeleton/tasks.md

@@ -94,7 +94,7 @@ design №54–66. Контракт v1.0 не меняется; нормализ
       нормализатор на реальных фикстурах `aep_6.json`, `aep_3.json` (не ошибка), `aep_3_ofr.json`, `aep_5.json`
       (`INVALID_DOCUMENT`), `aep_3.json` + `aep_3_ofr.json` в склейке под именами `<job_id>.json` / `<job_id>_ofr.json` →
       `NOT_IMPLEMENTED` с одной строкой warnings `<job_id>_ofr.json: INVALID_DOCUMENT: …`, воркер с фейками; `make check` зелёный
-- [ ] 7.3 Эмулятор: набор файлов на задание — `<name>.json` и `<name>_*.json` в `Output/<uuid>/` как `<uuid>.json` /
+- [x] 7.3 Эмулятор: набор файлов на задание — `<name>.json` и `<name>_*.json` в `Output/<uuid>/` как `<uuid>.json` /
       `<uuid>_<suffix>.json` с `file_uid = <uuid>`, атомарная запись каталога; проверка: unit-тесты (маркер `fixtures`) —
       `aep_3.pdf` → два файла, `aep_6.pdf` → один, `aep_5.PDF` → один с ошибкой, `aep_8.pdf` → `<uuid>_form4.json`, `aep_30*` не
       попадает в набор `aep_3`, неизвестное имя → `INVALID_DOCUMENT`; сквозной «put → эмулятор → poll (с settle) → склейка» на