api/src/core/config.py

@@ -83,13 +83,13 @@ class Settings(BaseSettings):
     SCAN_HOTFOLDER_PATH: str = ""  # корень горячей папки в режиме local (каталог должен существовать)
     SCAN_INPUT_DIR: str = "Input"
     SCAN_OUTPUT_DIR: str = "Output"
-    SCAN_TMP_DIR: str = "Tmp"  # служебный каталог временных файлов записи (рядом с Input, на той же ФС/шаре)
     SCAN_POLL_SECONDS: PositiveFloat = 5  # период опроса «Исходящих»
     # набор *.json ответа готов, когда имена и размеры не менялись столько секунд; 0 — готов сразу (тесты)
     SCAN_SETTLE_SECONDS: NonNegativeFloat = 2
-    SCAN_TIMEOUT_SECONDS: PositiveFloat = 300  # потолок ожидания ответа ИС (дополнительно к reply_deadline запроса)
+    # Потолок ожидания ответа ИС (дополнительно к reply_deadline запроса): 20 мин, в паре со сроком ответа OCR ядра
+    SCAN_TIMEOUT_SECONDS: PositiveFloat = 1200
     SCAN_MAX_INFLIGHT: PositiveInt = 4  # одновременно обрабатываемых заданий
-    # Очистка горячей папки после прочтения набора ответа (D-2): любое удаление, включая .part в Tmp, только при true.
+    # Очистка горячей папки после прочтения набора ответа (D-2): ответ, входящий файл, .done — только при true.
     # На SMB-шаре контура права на удаление не подтверждены — по умолчанию выключено
     SCAN_CLEANUP: bool = False
     SCAN_RULES_DIR: str = "rules"  # справочники нормализатора (yaml), путь от текущего каталога (api/)

api/src/gateway/__init__.py

@@ -18,7 +18,6 @@ def create_hotfolder(settings: Settings) -> HotFolder:
             settings.SCAN_HOTFOLDER_PATH,
             settings.SCAN_INPUT_DIR,
             settings.SCAN_OUTPUT_DIR,
-            settings.SCAN_TMP_DIR,
             settle_seconds=settings.SCAN_SETTLE_SECONDS,
             cleanup=settings.SCAN_CLEANUP,
         )

api/src/gateway/hotfolder.py

@@ -1,17 +1,16 @@
 """Горячие папки ИС Сканирование: протокол `HotFolder` и реализация в обычном каталоге (D-1, D-2, R-4, R-5).
 
-Файл кладётся во входящий каталог как `<job_id>.<ext>` (D-1): сначала временный `.<имя>.part` в служебном каталоге
-`Tmp` (на той же ФС/шаре), затем жёсткая ссылка (без замены) во входящий (решение №17). Ответ ИС — набор файлов каталога
+Файл кладётся во входящий каталог как `<job_id>.<ext>` (D-1) сразу, открытием `wb` с перезаписью существующего — как в
+старом АРМ (решение владельца 09.10.2026, change `hotfolder-direct-write`). Ответ ИС — набор файлов каталога
 `<output>/<job_id>/`: ИС кладёт туда все распознанные документы задания, по одному `*.json` на документ (OQ-78).
 `poll` отдаёт все `*.json` по возрастанию имени, когда набор готов: есть хотя бы один `*.json`, а набор имён и размеров
-не менялся `settle_seconds` (`SCAN_SETTLE_SECONDS`, `SettleTracker`). Имена каталогов (`Input`/`Output`/`Tmp`)
-настраиваются, а не зашиты в код.
+не менялся `settle_seconds` (`SCAN_SETTLE_SECONDS`, `SettleTracker`). Имена каталогов (`Input`/`Output`) настраиваются,
+а не зашиты в код; `Tmp` — служебный каталог эмулятора (сборка набора ответа), шлюз его не использует.
 
-Очистка (D-2) удаляет следы задания: ответ, входящий файл, временные файлы — при `SCAN_CLEANUP=true`.
+Очистка (D-2) удаляет следы задания: ответ, входящий файл и его `.done` — при `SCAN_CLEANUP=true`.
 """
 
 import contextlib
-import errno
 import os
 import re
 import shutil
@@ -29,8 +28,6 @@ _ILLEGAL_CHARS = frozenset(':<>"|?*') | frozenset(chr(code) for code in range(0x
 _FORBIDDEN_CHARS = _ILLEGAL_CHARS | frozenset(_PATH_SEPARATORS)
 # Расширение входящего файла — суффикс после последней точки, в нижнем регистре, 1–16 символов ASCII [0-9a-z]
 _EXTENSION_RE = re.compile(r"[0-9a-z]{1,16}")
-# errno `os.link`, означающие «ФС не умеет жёсткие ссылки» (запасной путь записи — `os.replace`)
-_NO_HARD_LINKS = (errno.EPERM, errno.EXDEV, errno.ENOTSUP, errno.EOPNOTSUPP)
 
 
 def validate_name(name: str) -> None:
@@ -171,7 +168,8 @@ class LocalHotFolder:
     """Горячая папка в обычном каталоге: dev, compose-том, unit-тесты.
 
     `root`, `input_dir`, `output_dir`, `tmp_dir`, `answer_dir` и `answer_path` открыты намеренно (R-4): эмулятор
-    ИС Сканирование работает с ними обычными операциями файловой системы (листинг входящих, `.done`, запись ответа).
+    ИС Сканирование работает с ними обычными операциями файловой системы (листинг входящих, `.done`, запись ответа);
+    `tmp_dir` нужен только эмулятору (сборка набора ответа), `put` его не использует.
     `settle_seconds` и `clock` (монотонные часы) — готовность набора ответа (`SettleTracker`).
     `cleanup_enabled` (D-2) — очистка включена при `SCAN_CLEANUP`.
     """
@@ -204,76 +202,22 @@ class LocalHotFolder:
         return self.answer_dir(job_id) / f"{job_id}.json"
 
     def put(self, name: str, data: bytes) -> None:
-        """Атомарная запись без замены: `.<name>.part` в `tmp_dir`, затем жёсткая ссылка в `input_dir/<name>` и
-        удаление `.part` как часть успешной публикации (R-8). Назначение уже есть — файл считается доставленным
-        (повтор задания, R-17): ничего не пишется, `.part` снимается только при включённой очистке. Корень не
-        создаётся (нет корня — `FileNotFoundError`)."""
+        """Записать `input_dir/<name>` сразу (`wb`, перезапись существующего, как в старом АРМ — решение владельца
+        09.10.2026). `Input` создаётся без родителей, корень — нет (нет корня — `FileNotFoundError`). При ошибке записи
+        недописанный файл снимается только при включённой очистке (R-3), ошибка снятия подавляется."""
         validate_name(name)
-        self.tmp_dir.mkdir(exist_ok=True)
         self.input_dir.mkdir(exist_ok=True)
-        part = self.tmp_dir / f".{name}.part"
         target = self.input_dir / name
-        if target.exists():
-            # R-17: `.part` от прошлой публикации может быть жёсткой ссылкой на этот файл (удаление после публикации
-            # подавляется, R-8, OQ-144) — писать и даже удалять его незачем, файл уже доставлен
-            self._drop_part(part)
-            return
-        # `.part` создаётся заново (№32): остаток от прерванной записи не должен ронять задание, поэтому ошибка его
-        # удаления подавляется; если он всё же остался, `_open_part` не даст записать «поверх» ссылки на чужой файл
-        with contextlib.suppress(OSError):
-            part.unlink(missing_ok=True)
-        fd = self._open_part(part)
         try:
-            with os.fdopen(fd, "wb") as handle:
-                handle.write(data)
+            target.write_bytes(data)
         except BaseException:
-            self._drop_part(part)
-            raise
-        try:
-            os.link(part, target)  # без замены: существующее назначение даёт FileExistsError
-        except FileExistsError:
-            self._drop_part(part)
-            return
-        except OSError as err:
-            # ФС без жёстких ссылок (некоторые тома Docker/сетевые монтирования): запасной путь через replace с
-            # проверкой назначения (окно гонки крошечное, повтор задания с тем же job_id редок). Только для самой
-            # линковки: ошибка записи выше поднимается как есть
-            if err.errno in _NO_HARD_LINKS and not target.exists():
-                self._replace_part(part, target)
-                return
-            self._drop_part(part)
-            raise
-        except BaseException:
-            self._drop_part(part)
-            raise
-        # завершение успешной записи (как rename в старой версии), не управляется флагом очистки:
-        # файл уже доставлен, ошибка удаления .part её не должна отменить (R-8)
-        with contextlib.suppress(OSError):
-            part.unlink(missing_ok=True)
-
-    @staticmethod
-    def _open_part(part: Path) -> int:
-        """Открыть `.part` на запись: новый файл (`O_EXCL`). Если старый не удалился (`FileExistsError`) — усечь его
-        только когда это обычный файл без других ссылок (`st_nlink == 1`); ссылку на опубликованный файл не трогаем,
-        исключение поднимается как есть (R-17)."""
-        try:
-            return os.open(part, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644)
-        except FileExistsError:
-            if os.lstat(part).st_nlink != 1:
-                raise
-            return os.open(part, os.O_WRONLY | os.O_TRUNC)
-
-    def _replace_part(self, part: Path, target: Path) -> None:
-        try:
-            os.replace(part, target)
-        except BaseException:
-            self._drop_part(part)
+            self._drop_partial(target)
             raise
 
-    def _drop_part(self, part: Path) -> None:
+    def _drop_partial(self, path: Path) -> None:
         if self.cleanup_enabled:
             with contextlib.suppress(OSError):
-                part.unlink(missing_ok=True)
+                path.unlink(missing_ok=True)
 
     def cleanup(self, job_id: str, name: str) -> bool:
         if not self.cleanup_enabled:
@@ -282,7 +226,7 @@ class LocalHotFolder:
         folder = self.answer_dir(job_id)
         if folder.is_dir():
             shutil.rmtree(folder)
-        for path in (self.input_dir / name, self.input_dir / f"{name}.done", self.tmp_dir / f".{name}.part"):
+        for path in (self.input_dir / name, self.input_dir / f"{name}.done"):
             try:
                 os.unlink(path)
             except FileNotFoundError:

api/tests/integration/test_roundtrip.py

@@ -3,7 +3,7 @@
 Настоящий lifespan приложения (`create_app` → S3, NATS, воркер, `OcrNormalizer` со справочниками `api/rules/`) и
 эмулятор ИС Сканирование (`src.scan_emulator.run` в потоке, `SCAN_EMULATOR_DEFAULT` пуст) работают в процессе
 теста; NATS 127.0.0.1:4222 и MinIO 127.0.0.1:9000 — из `api/.env.test` (`make up`), горячая папка — tmp-каталог.
-Входящий файл — `<job_id>.pdf` (расширение `original_name` в нижнем регистре, `.part` живёт в `Tmp`), пару фикстур
+Входящий файл — `<job_id>.pdf` (расширение `original_name` в нижнем регистре, пишется прямо в `Input`), пару фикстур
 эмулятор находит по sha256 содержимого: в его каталоге фикстур лежат PDF и ответы (реальные `aep_*` из
 `fixtures/ocr` и синтетический `ok_doc`).
 
@@ -16,7 +16,7 @@
 3. `aep_6.pdf` (один успешный файл) → `needs_check`, `result_s3` с двумя документами (`BALANCE`, `INCOME`);
 4. `ok_doc.pdf` (синтетический успешный ответ в форме реальных) → `needs_check` (редакция `fsbu4` пуста) с `result_s3`;
 5. `unknown.pdf` (содержимого нет в фикстурах) → ack, затем `INVALID_DOCUMENT` эмулятора;
-6. `aep_6.pdf` при `SCAN_CLEANUP=true` → после результата `Input/` и `Output/<job_id>/` пусты, `Tmp/` без `.part`.
+6. `aep_6.pdf` при `SCAN_CLEANUP=true` → после результата `Input/` и `Output/<job_id>/` пусты.
 Каждый конверт проверяется схемой `recognition-result-message`, `service.name` и `Nats-Msg-Id`; файл результата
 из MinIO — схемой `recognition-result`; склеенный сырой ответ в MinIO содержит ровно те файлы, что лежат в каталоге
 ответа, в порядке возрастания имени, с телами байт в байт; входящий файл совпадает с объектом S3. Готовность набора
@@ -178,10 +178,6 @@ class Roundtrip:
     def input_dir(self) -> Path:
         return self.root / self.settings.SCAN_INPUT_DIR
 
-    @property
-    def tmp_dir(self) -> Path:
-        return self.root / self.settings.SCAN_TMP_DIR
-
     def answer_dir(self, job_id: str) -> Path:
         return self.root / self.settings.SCAN_OUTPUT_DIR / job_id
 
@@ -260,12 +256,6 @@ class Roundtrip:
     def inbox(self, job_id: str) -> list[str]:
         return sorted(p.name for p in self.input_dir.iterdir() if p.name.startswith(job_id))
 
-    def stray_parts(self) -> list[str]:
-        """Временные `.part` записи во входящий каталог: после успешной публикации их в `Tmp` нет (решение №26)."""
-        return (
-            sorted(p.name for p in self.tmp_dir.iterdir() if p.name.endswith(".part")) if self.tmp_dir.is_dir() else []
-        )
-
     async def run(self, original_name: str, data: bytes, *, cleanup: bool = False) -> tuple[dict[str, Any], Envelope]:
         """Запрос → ack → итоговый конверт; общие проверки сквозного сценария. Возвращает запрос и итог."""
         body = await self.request(original_name, data)
@@ -283,8 +273,7 @@ class Roundtrip:
         """Сырой ответ в MinIO — склейка ответа эмулятора (D-4); входящий файл `<job_id>.pdf` = объект S3.
 
         Без очистки (`SCAN_CLEANUP=false`) каталог ответа и входящий файл (с суффиксом `.done` эмулятора) остаются; при
-        очистке (`SCAN_CLEANUP=true`) во входящем каталоге пусто, каталога `Output/<job_id>/` нет. `.part` в `Tmp` нет
-        в обоих режимах.
+        очистке (`SCAN_CLEANUP=true`) во входящем каталоге пусто, каталога `Output/<job_id>/` нет.
         """
         job_id = body["job_id"]
         assert result.body["file_id"] == body["file_id"]
@@ -297,7 +286,6 @@ class Roundtrip:
             assert entry["response"]["file_uid"] == job_id
         stat = await self.s3.client.stat_object(self.s3.bucket, f"raw/{job_id}.json")
         assert stat.content_type == JSON_CONTENT_TYPE
-        assert self.stray_parts() == []
         if cleanup:
             assert list(self.input_dir.iterdir()) == []
             assert not self.answer_dir(job_id).exists()
@@ -393,7 +381,7 @@ def _start_emulator(stack: AsyncExitStack, settings: Settings, root: Path, fixtu
         hotfolder_path=root,
         input_dir=settings.SCAN_INPUT_DIR,
         output_dir=settings.SCAN_OUTPUT_DIR,
-        tmp_dir=settings.SCAN_TMP_DIR,
+        tmp_dir="Tmp",  # служебный каталог эмулятора; у шлюза его нет (change hotfolder-direct-write)
         fixtures_dir=fixtures,
         delay=EMULATOR_DELAY,
         poll_interval=0.05,
@@ -650,4 +638,3 @@ async def test_cleanup_enabled_leaves_input_and_answer_empty(roundtrip_cleanup:
     assert [entry["name"] for entry in raw["files"]] == [f"{job_id}.json"]
     assert list(roundtrip_cleanup.input_dir.iterdir()) == []
     assert not roundtrip_cleanup.answer_dir(job_id).exists()
-    assert roundtrip_cleanup.stray_parts() == []

api/tests/unit/conftest.py

@@ -58,7 +58,6 @@ async def make_env(tmp_path: Path) -> AsyncIterator[EnvFactory]:
             root,
             settings.SCAN_INPUT_DIR,
             settings.SCAN_OUTPUT_DIR,
-            settings.SCAN_TMP_DIR,
             settle_seconds=settings.SCAN_SETTLE_SECONDS,
             cleanup=settings.SCAN_CLEANUP,
         )

api/tests/unit/test_config.py

@@ -29,10 +29,10 @@ def test_defaults_local_mode_without_smb_and_nats_credentials(monkeypatch, minim
     assert s.SCAN_HOTFOLDER_MODE == "local"
     assert s.SCAN_INPUT_DIR == "Input"
     assert s.SCAN_OUTPUT_DIR == "Output"
-    assert s.SCAN_TMP_DIR == "Tmp"
+    assert not hasattr(s, "SCAN_TMP_DIR")  # служебный каталог только у эмулятора (change hotfolder-direct-write)
     assert s.SCAN_SETTLE_SECONDS == 2
     assert s.SCAN_POLL_SECONDS == 5
-    assert s.SCAN_TIMEOUT_SECONDS == 300
+    assert s.SCAN_TIMEOUT_SECONDS == 1200
     assert s.SCAN_MAX_INFLIGHT == 4
     assert s.OPENBAO.SMB.SMB_HOST == ""
     assert s.OPENBAO.NATS.NATS_SEED == ""
@@ -80,6 +80,16 @@ def test_negative_settle_seconds_rejected(monkeypatch):
         get_settings()
 
 
+def test_legacy_scan_tmp_dir_is_ignored(monkeypatch, minimal_env_file):
+    """Старый `.env` с `SCAN_TMP_DIR` (до change hotfolder-direct-write) не ломает старт: `extra="ignore"`."""
+    _clean_env(monkeypatch)
+    monkeypatch.setenv("SCAN_TMP_DIR", "Tmp")
+
+    s = get_settings()
+
+    assert not hasattr(s, "SCAN_TMP_DIR")
+
+
 @pytest.mark.parametrize("settle", ["10", "11"])
 def test_settle_not_less_than_timeout_rejected(monkeypatch, settle):
     """M-4: набор не может настояться за срок, который короче или равен settle — каждое задание давало бы TIMEOUT."""
@@ -156,7 +166,7 @@ def test_env_test_file_loads_for_tests():
     assert s.SCAN_HOTFOLDER_PATH == "./.hotfolder-test"
     assert s.SCAN_POLL_SECONDS == 0.1
     assert s.SCAN_TIMEOUT_SECONDS == 5
-    assert s.SCAN_TMP_DIR == "Tmp"
+    assert not hasattr(s, "SCAN_TMP_DIR")
     assert s.SCAN_SETTLE_SECONDS == 0  # R-23: тесты воркера не ждут стабилизации набора
     assert s.OPENBAO.MINIO.MINIO_URL == "127.0.0.1:9000"
     assert s.OPENBAO.MINIO.MINIO_SECURE is False
@@ -172,7 +182,7 @@ def test_env_example_loads(monkeypatch):
 
     assert s.SCAN_HOTFOLDER_MODE == "local"
     assert s.SCAN_INPUT_DIR == "Input"
-    assert s.SCAN_TMP_DIR == "Tmp"
+    assert not hasattr(s, "SCAN_TMP_DIR")
     assert s.SCAN_SETTLE_SECONDS == 2
     assert s.OPENBAO.MINIO.MINIO_BUCKET == "arm-ka"
     assert s.OPENBAO.NATS.NATS_JWT == ""

api/tests/unit/test_hotfolder_factory.py

@@ -20,7 +20,6 @@ def _local_settings(root: Path) -> Settings:
             "SCAN_HOTFOLDER_MODE": "local",
             "SCAN_HOTFOLDER_PATH": str(root),
             "SCAN_INPUT_DIR": "In",
-            "SCAN_TMP_DIR": "Служебный",
             "SCAN_SETTLE_SECONDS": 1.5,
         }
     )
@@ -45,7 +44,7 @@ def test_local_mode_builds_local_hotfolder(tmp_path: Path) -> None:
     assert hf.root == tmp_path
     assert hf.input_dir == tmp_path / "In"
     assert hf.output_dir == tmp_path / "Output"
-    assert hf.tmp_dir == tmp_path / "Служебный"
+    assert hf.tmp_dir == tmp_path / "Tmp"  # служебный каталог эмулятора, шлюз его не использует
     assert hf.settle_seconds == 1.5
     assert hf.healthy() is True
 

api/tests/unit/test_local_hotfolder.py

@@ -1,6 +1,6 @@
 """Горячая папка в обычном каталоге (spec image-recognition-gateway, D-1, R-4/R-5, slice-0-skeleton).
 
-Быстрые тесты на `tmp_path`: атомарная запись входящего файла через служебный каталог `Tmp`, опрос набора ответов ИС
+Быстрые тесты на `tmp_path`: запись входящего файла прямо в `Input` с перезаписью, опрос набора ответов ИС
 с готовностью по стабилизации (settle, часы подменены), проверка здоровья, валидация имён.
 """
 
@@ -41,6 +41,17 @@ def _settled(tmp_path: Path, settle: float = 2) -> tuple[LocalHotFolder, FakeClo
     return LocalHotFolder(tmp_path, settle_seconds=settle, clock=clock), clock
 
 
+def _failing_write(self: Path, data: bytes) -> int:
+    """Недописанный файл: на диск попадают первые два байта, затем ошибка ввода-вывода."""
+    with open(self, "wb") as handle:
+        handle.write(data[:2])
+    raise OSError(errno.ENOSPC, "диск полон")
+
+
+def _failing_unlink(self: Path, missing_ok: bool = False) -> None:
+    raise PermissionError(errno.EACCES, "нет прав")
+
+
 class TestInboxName:
     def test_job_id_and_lowercase_extension(self) -> None:
         job_id = _job_id()
@@ -57,34 +68,12 @@ class TestInboxName:
             inbox_name("job", original)
 
 
-class TestPutWithoutReplace:
-    def test_existing_destination_is_kept(self, tmp_path: Path) -> None:
-        hf = LocalHotFolder(tmp_path)
-        hf.put("job.pdf", b"first")
-        hf.put("job.pdf", b"second")
-        assert (hf.input_dir / "job.pdf").read_bytes() == b"first"
-        # R-17: назначение есть — повтор ничего не пишет; прежнего `.part` после успешной публикации нет
-        assert not (hf.tmp_dir / ".job.pdf.part").exists()
-
-    def test_existing_destination_part_removed_when_cleanup_enabled(self, tmp_path: Path) -> None:
-        hf = LocalHotFolder(tmp_path, cleanup=True)
-        hf.put("job.pdf", b"first")
-        hf.put("job.pdf", b"second")
-        assert (hf.input_dir / "job.pdf").read_bytes() == b"first"
-        assert not (hf.tmp_dir / ".job.pdf.part").exists()
-
-    def test_no_temp_files_in_input(self, tmp_path: Path) -> None:
-        hf = LocalHotFolder(tmp_path)
-        hf.put("job.pdf", b"x")
-        assert [p.name for p in hf.input_dir.iterdir()] == ["job.pdf"]
-        assert not (hf.tmp_dir / ".job.pdf.part").exists()  # после успешной публикации .part нет
-
-
 class TestCleanup:
     def _fill(self, hf: LocalHotFolder, job_id: str) -> None:
         hf.put(f"{job_id}.pdf", b"x")
         (hf.input_dir / f"{job_id}.pdf.done").write_bytes(b"")
-        (hf.tmp_dir / f".{job_id}.pdf.part").write_bytes(b"orphan")
+        hf.tmp_dir.mkdir(exist_ok=True)
+        (hf.tmp_dir / f".{job_id}.pdf.part").write_bytes(b"legacy")  # остаток прежних версий шлюза — не наш
         _write_answer(hf, job_id, b"{}")
         _write_answer(hf, job_id, b"{}", name=f"{job_id}_ofr.json")
         (hf.answer_dir(job_id) / "readme.txt").write_bytes(b"keep")
@@ -104,7 +93,7 @@ class TestCleanup:
         assert hf.cleanup(job_id, f"{job_id}.pdf") is True
         assert not (hf.input_dir / f"{job_id}.pdf").exists()
         assert not (hf.input_dir / f"{job_id}.pdf.done").exists()
-        assert not (hf.tmp_dir / f".{job_id}.pdf.part").exists()
+        assert (hf.tmp_dir / f".{job_id}.pdf.part").exists()  # Tmp очистка не трогает
         assert not hf.answer_dir(job_id).exists()  # каталог удалён вместе с readme.txt
         assert hf.cleanup(job_id, f"{job_id}.pdf") is True  # повтор — не ошибка
 
@@ -181,60 +170,42 @@ class TestPut:
         hf.put(name, data)
         assert (tmp_path / "Input" / name).read_bytes() == data
 
-    def test_leaves_no_part_files(self, tmp_path: Path) -> None:
+    def test_creates_input_dir_only(self, tmp_path: Path) -> None:
         hf = LocalHotFolder(tmp_path)
-        job_id = _job_id()
-        name = inbox_name(job_id, "aep_5.pdf")
-        hf.put(name, b"data")
-        assert [p.name for p in hf.input_dir.iterdir()] == [name]
-        assert list(hf.tmp_dir.iterdir()) == []
-
-    def test_part_is_written_in_tmp_not_in_input(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
-        """OQ-79: во время записи во входящем каталоге нет временного файла — он в `Tmp/`, затем link в `Input`."""
-        hf = LocalHotFolder(tmp_path)
-        job_id = _job_id()
-        name = inbox_name(job_id, "aep_5.pdf")
-        seen: dict[str, Any] = {}
-        real_link = os.link
-
-        def spy(src: Any, dst: Any) -> None:
-            seen["src"], seen["dst"] = Path(src), Path(dst)
-            seen["input"] = sorted(p.name for p in hf.input_dir.iterdir())
-            seen["tmp"] = sorted(p.name for p in hf.tmp_dir.iterdir())
-            seen["part"] = Path(src).read_bytes()
-            real_link(src, dst)
-
-        monkeypatch.setattr("src.gateway.hotfolder.os.link", spy)
-        hf.put(name, b"data")
-        assert seen == {
-            "src": hf.tmp_dir / f".{name}.part",
-            "dst": hf.input_dir / name,
-            "input": [],
-            "tmp": [f".{name}.part"],
-            "part": b"data",
-        }
-        assert [p.name for p in hf.input_dir.iterdir()] == [name]
-        assert list(hf.tmp_dir.iterdir()) == []
-
-    def test_creates_input_and_tmp_dirs_only(self, tmp_path: Path) -> None:
-        hf = LocalHotFolder(tmp_path)
-        assert not hf.input_dir.exists()
-        assert not hf.tmp_dir.exists()
         hf.put("a.pdf", b"1")
-        assert hf.input_dir.is_dir()
-        assert hf.tmp_dir.is_dir()
-        assert not hf.output_dir.exists()
+        assert sorted(p.name for p in tmp_path.iterdir()) == ["Input"]  # Tmp не создаётся
 
-    def test_existing_tmp_dir_is_fine(self, tmp_path: Path) -> None:
-        (tmp_path / "Tmp").mkdir()
-        LocalHotFolder(tmp_path).put("a.pdf", b"1")
-        assert (tmp_path / "Input" / "a.pdf").read_bytes() == b"1"
+    def test_existing_file_is_overwritten(self, tmp_path: Path) -> None:
+        """Как в старом АРМ (решение владельца 09.10.2026): повтор задания перезаписывает входящий файл."""
+        hf = LocalHotFolder(tmp_path, settle_seconds=0)
+        hf.put("job.pdf", b"first, longer")
+        hf.put("job.pdf", b"second")
+        assert (hf.input_dir / "job.pdf").read_bytes() == b"second"
 
-    def test_custom_tmp_dir_name(self, tmp_path: Path) -> None:
-        hf = LocalHotFolder(tmp_path, tmp_dir="Служебный")
-        hf.put("a.pdf", b"1")
-        assert (tmp_path / "Служебный").is_dir()
-        assert not (tmp_path / "Tmp").exists()
+    def test_write_failure_keeps_partial_file_when_cleanup_disabled(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        hf = LocalHotFolder(tmp_path, settle_seconds=0)
+        monkeypatch.setattr(Path, "write_bytes", _failing_write)
+        with pytest.raises(OSError, match="диск полон"):
+            hf.put("job.pdf", b"data")
+        assert (hf.input_dir / "job.pdf").read_bytes() == b"da"
+
+    def test_write_failure_removes_partial_file_when_cleanup_enabled(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        hf = LocalHotFolder(tmp_path, settle_seconds=0, cleanup=True)
+        monkeypatch.setattr(Path, "write_bytes", _failing_write)
+        with pytest.raises(OSError, match="диск полон"):
+            hf.put("job.pdf", b"data")
+        assert not (hf.input_dir / "job.pdf").exists()
+
+    def test_remove_failure_does_not_mask_write_error(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+        hf = LocalHotFolder(tmp_path, settle_seconds=0, cleanup=True)
+        monkeypatch.setattr(Path, "write_bytes", _failing_write)
+        monkeypatch.setattr(Path, "unlink", _failing_unlink)
+        with pytest.raises(OSError, match="диск полон"):
+            hf.put("job.pdf", b"data")
 
     def test_existing_input_dir_is_fine(self, tmp_path: Path) -> None:
         (tmp_path / "Input").mkdir()
@@ -285,227 +256,6 @@ class TestPut:
             hf.put(name, b"x")
         assert sorted(p.name for p in hf.input_dir.iterdir()) == sorted(names)
 
-    def test_cleans_part_file_on_failure(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
-        """При cleanup=True и ошибке link: .part удаляется."""
-        hf = LocalHotFolder(tmp_path, cleanup=True)
-
-        def boom(src: Any, dst: Any) -> None:
-            raise OSError("диск недоступен")
-
-        monkeypatch.setattr("src.gateway.hotfolder.os.link", boom)
-        with pytest.raises(OSError, match="диск недоступен"):
-            hf.put("a.pdf", b"data")
-        assert list(hf.input_dir.iterdir()) == []
-        assert list(hf.tmp_dir.iterdir()) == []
-
-    def test_keeps_part_file_on_failure_when_cleanup_disabled(
-        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
-    ) -> None:
-        """При cleanup=False и ошибке link: .part остаётся в Tmp (T11.1)."""
-        hf = LocalHotFolder(tmp_path, cleanup=False)
-
-        def boom(src: Any, dst: Any) -> None:
-            raise OSError("диск недоступен")
-
-        monkeypatch.setattr("src.gateway.hotfolder.os.link", boom)
-        with pytest.raises(OSError, match="диск недоступен"):
-            hf.put("a.pdf", b"data")
-        assert list(hf.input_dir.iterdir()) == []
-        assert [p.name for p in hf.tmp_dir.iterdir()] == [".a.pdf.part"]
-
-    def test_successful_put_despite_final_unlink_failure(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
-        """Успешный link, но ошибка в завершающем unlink .part (R-8): put НЕ падает, файл доставлен (T11.2)."""
-        hf = LocalHotFolder(tmp_path)
-
-        unlink_calls = []
-
-        real_unlink = os.unlink
-
-        def unlink_with_failure(path: Any) -> None:
-            unlink_calls.append(path)
-            if str(path).endswith(".part"):
-                raise PermissionError("удаление запрещено")
-            real_unlink(path)
-
-        monkeypatch.setattr("src.gateway.hotfolder.os.unlink", unlink_with_failure)
-        hf.put("a.pdf", b"data")  # НЕ падает несмотря на ошибку unlink
-        assert (hf.input_dir / "a.pdf").read_bytes() == b"data"
-        assert (hf.tmp_dir / ".a.pdf.part").exists()  # .part не удалён из-за ошибки
-        assert unlink_calls  # попытка удаления была
-
-    def test_cleanup_failure_does_not_mask_original_error(
-        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
-    ) -> None:
-        """Ошибка удаления `.part` на пути ошибки link подавляется (T11.1); исходное исключение проходит."""
-        hf = LocalHotFolder(tmp_path, cleanup=True)
-
-        def boom(src: Any, dst: Any) -> None:
-            raise OSError("диск недоступен")
-
-        def unlink_fails(self: Path, missing_ok: bool = False) -> None:
-            raise PermissionError("удаление .part запрещено")
-
-        monkeypatch.setattr("src.gateway.hotfolder.os.link", boom)
-        monkeypatch.setattr(Path, "unlink", unlink_fails)
-        with pytest.raises(OSError, match="диск недоступен"):
-            hf.put("a.pdf", b"data")
-
-
-class TestPartIsCreatedFresh:
-    """Решение №32: `.part` создаётся заново (удаление старого + `O_EXCL`), fallback — только вокруг `os.link`."""
-
-    @pytest.mark.parametrize("cleanup", [False, True])
-    def test_existing_destination_is_not_touched_by_repeat(self, tmp_path: Path, cleanup: bool) -> None:
-        """R-17: назначение есть, `.part` — жёсткая ссылка на него: байты не меняются, put не падает."""
-        hf = LocalHotFolder(tmp_path, cleanup=cleanup)
-        hf.put("job.pdf", b"first")
-        published = hf.input_dir / "job.pdf"
-        stale = hf.tmp_dir / ".job.pdf.part"
-        os.link(published, stale)
-        hf.put("job.pdf", b"different data")
-        assert published.read_bytes() == b"first"
-        assert stale.exists() is (not cleanup)  # очистка снимает `.part`, иначе он остаётся как был
-        if not cleanup:
-            assert stale.read_bytes() == b"first" and published.stat().st_nlink == 2
-
-    def test_existing_destination_with_undeletable_part_does_not_fail(
-        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
-    ) -> None:
-        hf = LocalHotFolder(tmp_path)  # очистка выключена: удаление `.part` вообще не пытается
-        hf.put("job.pdf", b"first")
-        os.link(hf.input_dir / "job.pdf", hf.tmp_dir / ".job.pdf.part")
-
-        def deny(self: Path, missing_ok: bool = False) -> None:
-            raise PermissionError("удаление запрещено")
-
-        monkeypatch.setattr(Path, "unlink", deny)
-        hf.put("job.pdf", b"second")
-        assert (hf.input_dir / "job.pdf").read_bytes() == b"first"
-
-    def test_undeletable_stale_part_with_single_link_is_truncated(
-        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
-    ) -> None:
-        """Назначения нет, старый `.part` не удаляется, но это обычный файл (nlink == 1): put успешен."""
-        hf = LocalHotFolder(tmp_path)
-        hf.tmp_dir.mkdir()
-        stale = hf.tmp_dir / ".job.pdf.part"
-        stale.write_bytes(b"old and longer content")
-
-        def deny(self: Path, missing_ok: bool = False) -> None:
-            raise PermissionError("удаление запрещено")
-
-        monkeypatch.setattr(Path, "unlink", deny)
-        hf.put("job.pdf", b"new")
-        assert (hf.input_dir / "job.pdf").read_bytes() == b"new"
-
-    def test_undeletable_stale_part_with_other_links_is_not_written(
-        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
-    ) -> None:
-        """Назначения нет, старый `.part` не удаляется и имеет nlink == 2 (синтетически): запись не идёт, ошибка не
-        маскируется, второй файл не изменён."""
-        hf = LocalHotFolder(tmp_path)
-        hf.tmp_dir.mkdir()
-        other = tmp_path / "other.bin"
-        other.write_bytes(b"precious")
-        os.link(other, hf.tmp_dir / ".job.pdf.part")
-
-        def deny(self: Path, missing_ok: bool = False) -> None:
-            raise PermissionError("удаление запрещено")
-
-        monkeypatch.setattr(Path, "unlink", deny)
-        with pytest.raises(FileExistsError):
-            hf.put("job.pdf", b"new")
-        assert other.read_bytes() == b"precious"
-        assert not (hf.input_dir / "job.pdf").exists()
-
-    def test_stale_part_is_replaced_before_publication(self, tmp_path: Path) -> None:
-        hf = LocalHotFolder(tmp_path)
-        hf.tmp_dir.mkdir()
-        (hf.tmp_dir / ".job.pdf.part").write_bytes(b"old and longer content")
-        hf.put("job.pdf", b"new")
-        assert (hf.input_dir / "job.pdf").read_bytes() == b"new"
-        assert not (hf.tmp_dir / ".job.pdf.part").exists()
-
-    def test_part_is_opened_exclusively(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
-        hf = LocalHotFolder(tmp_path)
-        flags: list[int] = []
-        real_open = os.open
-
-        def spy(path: Any, flag: int, mode: int = 0o777, **kwargs: Any) -> int:
-            flags.append(flag)
-            return real_open(path, flag, mode, **kwargs)
-
-        monkeypatch.setattr("src.gateway.hotfolder.os.open", spy)
-        hf.put("job.pdf", b"x")
-        assert flags and flags[0] & os.O_EXCL and flags[0] & os.O_CREAT
-
-    def test_write_error_propagates_as_itself(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
-        """EPERM записи не превращается в `FileNotFoundError` от `os.replace` несуществующего `.part`."""
-        hf = LocalHotFolder(tmp_path, cleanup=True)
-
-        class FailingFile:
-            def __enter__(self) -> "FailingFile":
-                return self
-
-            def __exit__(self, *exc: object) -> None:
-                return None
-
-            def write(self, data: bytes) -> int:
-                raise OSError(errno.EPERM, "запись запрещена")
-
-        monkeypatch.setattr("src.gateway.hotfolder.os.fdopen", lambda fd, mode: (os.close(fd), FailingFile())[1])
-        with pytest.raises(OSError, match="запись запрещена") as caught:
-            hf.put("job.pdf", b"data")
-        assert caught.value.errno == errno.EPERM and not isinstance(caught.value, FileNotFoundError)
-        assert list(hf.input_dir.iterdir()) == [] and list(hf.tmp_dir.iterdir()) == []
-
-    @pytest.mark.parametrize("code", [errno.EPERM, errno.EXDEV, errno.ENOTSUP, errno.EOPNOTSUPP])
-    def test_fallback_to_replace_when_link_unsupported(
-        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, code: int
-    ) -> None:
-        hf = LocalHotFolder(tmp_path)
-
-        def no_links(src: Any, dst: Any) -> None:
-            raise OSError(code, "жёсткие ссылки не поддерживаются")
-
-        monkeypatch.setattr("src.gateway.hotfolder.os.link", no_links)
-        hf.put("job.pdf", b"data")
-        assert (hf.input_dir / "job.pdf").read_bytes() == b"data"
-        assert list(hf.tmp_dir.iterdir()) == []  # `.part` стал назначением
-
-    def test_no_fallback_when_destination_appeared_during_link(
-        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
-    ) -> None:
-        """Назначение появилось между проверкой и линковкой, а `os.link` дал EPERM: `os.replace` не затирает."""
-        hf = LocalHotFolder(tmp_path, cleanup=True)
-        hf.input_dir.mkdir()
-
-        def no_links(src: Any, dst: Any) -> None:
-            Path(dst).write_bytes(b"first")  # файл успел появиться (параллельный повтор того же задания)
-            raise OSError(errno.EPERM, "жёсткие ссылки не поддерживаются")
-
-        monkeypatch.setattr("src.gateway.hotfolder.os.link", no_links)
-        with pytest.raises(OSError, match="жёсткие ссылки"):
-            hf.put("job.pdf", b"second")
-        assert (hf.input_dir / "job.pdf").read_bytes() == b"first"
-
-    def test_replace_failure_after_link_fallback_propagates(
-        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
-    ) -> None:
-        hf = LocalHotFolder(tmp_path, cleanup=True)
-
-        def no_links(src: Any, dst: Any) -> None:
-            raise OSError(errno.EPERM, "жёсткие ссылки не поддерживаются")
-
-        def broken_replace(src: Any, dst: Any) -> None:
-            raise OSError(errno.EIO, "замена не удалась")
-
-        monkeypatch.setattr("src.gateway.hotfolder.os.link", no_links)
-        monkeypatch.setattr("src.gateway.hotfolder.os.replace", broken_replace)
-        with pytest.raises(OSError, match="замена не удалась"):
-            hf.put("job.pdf", b"data")
-        assert list(hf.tmp_dir.iterdir()) == []  # при очистке `.part` снят
-
 
 class TestPoll:
     """`poll` — все `*.json` каталога `<output>/<job_id>/` по имени, когда набор настоялся `settle_seconds`."""

api/tests/unit/test_worker.py

@@ -953,13 +953,14 @@ class TestSlice3a:
         assert "Очистка горячей папки не удалась" in warnings[0] and job_id in warnings[0] and "нет прав" in warnings[0]
 
     async def test_existing_input_waits_for_answer(self, make_env: EnvFactory) -> None:
-        """Review Focus 4: входящий файл уже лежит (повтор задания), ответа нет — TIMEOUT, файл не перезаписан."""
+        """Входящий файл уже лежит (повтор задания), ответа нет — TIMEOUT; файл перезаписан объектом S3, как в старом
+        АРМ (change hotfolder-direct-write, решение владельца 09.10.2026)."""
         env = await make_env()
         body = env.request(deadline_s=0.3)
         env.local.put(f"{body['job_id']}.pdf", b"already there")
         result = await env.run(body)
         assert result["error"]["code"] == "TIMEOUT"
-        assert (env.local.input_dir / f"{body['job_id']}.pdf").read_bytes() == b"already there"
+        assert (env.local.input_dir / f"{body['job_id']}.pdf").read_bytes() == PDF_BYTES
         assert env.inbox() == [f"{body['job_id']}.pdf"]
 
     async def test_ocr_normalizer_produces_documents(self, make_env: EnvFactory) -> None: