arm-scan-gateway/docs/superpowers/plans/2026-10-09-hotfolder-direct-write.md

Запись в горячую папку прямо в Input — Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:executing-plans (план из 3 задач, ≤5 — Native по ~/.claude/CLAUDE.md). Steps use checkbox (- [ ]) syntax for tracking.

Goal: put пишет <job_id>.<ext> сразу в Input (wb, перезапись), без Tmp/.part и rename; SCAN_TMP_DIR уходит из шлюза; SCAN_TIMEOUT_SECONDS по умолчанию 1200.

Architecture: Меняются только SmbHotFolder.put/cleanup и LocalHotFolder.put/cleanup; протокол HotFolder, воркер и эмулятор не меняются. LocalHotFolder.tmp_dir остаётся для эмулятора.

Tech Stack: Python 3.11, smbprotocol (smbclient), pytest, ruff, mypy.

Spec: openspec/changes/hotfolder-direct-write/ (proposal, design D-1…D-5, R-1…R-4, specs/image-recognition-gateway)

Рабочий каталог: /Users/andrey.epifancev/projects/awp-ca/.wt/smb-direct-write/scan-gateway (ветка feat/26-hotfolder-direct-write). Git — git -C /Users/andrey.epifancev/projects/awp-ca/.wt/smb-direct-write/scan-gateway. pytest — из api/ (.venv/bin/pytest), make — из корня репозитория.

Global Constraints

• Запись: smbclient.open_file(<Input>\<имя>, mode="wb", **credentials) / open(<input>/<имя>, "wb"); существующий файл перезаписывается.
• Временных файлов, Tmp и rename у шлюза нет; Input создаётся терпимо (_ensure_dir / mkdir(exist_ok=True)), корень — нет.
• Ошибка записи: недописанный <Input>/<имя> снимается только при cleanup=True; ошибка снятия подавляется, поднимается исходная.
• При SCAN_CLEANUP=false ничего не удаляется; cleanup удаляет Output/<job_id>/, Input/<имя>, Input/<имя>.done — и больше ничего.
• SCAN_TIMEOUT_SECONDS: PositiveFloat = 1200; SCAN_TMP_DIR в Settings нет (у эмулятора — есть).
• Учётные данные передаются в каждый вызов smbclient (как сейчас).

Review Focus

• Повторный put того же имени с другим содержимым → файл перезаписан новым содержимым (оба режима).
• Ошибка посреди write при cleanup=False → обрезанный файл остаётся, исключение исходное.
• Ошибка посреди write при cleanup=True и падающем remove → поднимается исходная ошибка записи.
• Старый .env с SCAN_TMP_DIR=Tmp → шлюз стартует (extra="ignore").
• put не создаёт Tmp ни в одном режиме; эмулятор по-прежнему собирает ответ в Tmp.

---

Task 1: SmbHotFolder — прямая запись

Files:

• Modify: api/src/gateway/smb.py (докстринг модуля, __init__, put, _drop_part → _drop_partial, cleanup)
• Test: api/tests/unit/test_smb_hotfolder.py (TMP, TestPaths, TestPut, TestPutWithoutReplace → TestPutOverwrite, TestCleanup*)

Interfaces:

• Produces: SmbHotFolder(host, path, input_dir="Input", output_dir="Output", *, login="", password="", settle_seconds=2, clock=..., cleanup=False) — без tmp_dir и атрибута tmp_dir.

• Step 1: Переписать тесты put и cleanup

Удалить константу TMP, из TestPaths.test_default_layout — assert hf.tmp_dir == TMP, из test_custom_dir_names — tmp_dir=... и его assert. TestPut и TestPutWithoutReplace заменить на:

class TestPut:
    def test_writes_directly_into_input(self, fake: FakeSmbClient, share_root: Path) -> None:
        name = inbox_name(_job_id(), "aep_5.PDF")
        data = bytes(range(256)) * 10
        _hotfolder().put(name, data)
        assert fake.calls == [("mkdir", (INPUT,), {}), ("open_file", (f"{INPUT}\\{name}", "wb"), {})]
        assert (share_root / "Input" / name).read_bytes() == data
        assert sorted(p.name for p in share_root.iterdir()) == ["Input"]  # Tmp не создаётся

    def test_existing_input_dir_is_fine(self, fake: FakeSmbClient, share_root: Path) -> None:
        (share_root / "Input").mkdir()
        _hotfolder().put("a.pdf", b"1")
        assert (share_root / "Input" / "a.pdf").read_bytes() == b"1"

    def test_missing_root_raises_and_writes_nothing(self, fake: FakeSmbClient) -> None:
        with pytest.raises(OSError) as excinfo:
            _hotfolder().put("a.pdf", b"1")
        assert isinstance(excinfo.value, SMBOSError)
        assert fake.funcs() == ["mkdir"]
        assert not fake.base.exists()

    def test_mkdir_denied_but_input_exists_is_fine(self, fake: FakeSmbClient, share_root: Path) -> None:
        """Урезанные ACL: нельзя создавать подкаталоги в корне, но `Input` уже есть и в него писать можно."""
        (share_root / "Input").mkdir()
        fake.fail["mkdir"] = SMBOSError(NtStatus.STATUS_ACCESS_DENIED, INPUT)
        _hotfolder().put("a.pdf", b"1")
        assert (share_root / "Input" / "a.pdf").read_bytes() == b"1"
        assert fake.funcs() == ["mkdir", "isdir", "open_file"]

    def test_mkdir_denied_and_no_input_raises_original(self, fake: FakeSmbClient, share_root: Path) -> None:
        denied = SMBOSError(NtStatus.STATUS_ACCESS_DENIED, INPUT)
        fake.fail["mkdir"] = denied
        with pytest.raises(SMBOSError) as excinfo:
            _hotfolder().put("a.pdf", b"1")
        assert excinfo.value is denied
        assert fake.funcs() == ["mkdir", "isdir"]

    def test_mkdir_denied_and_isdir_fails_raises_original(self, fake: FakeSmbClient, share_root: Path) -> None:
        denied = SMBOSError(NtStatus.STATUS_ACCESS_DENIED, INPUT)
        fake.fail["mkdir"] = denied
        fake.fail["isdir"] = SMBOSError(NtStatus.STATUS_ACCESS_DENIED, INPUT)
        with pytest.raises(SMBOSError) as excinfo:
            _hotfolder().put("a.pdf", b"1")
        assert excinfo.value is denied
        assert fake.funcs() == ["mkdir", "isdir"]

    def test_empty_data_is_written(self, fake: FakeSmbClient, share_root: Path) -> None:
        _hotfolder().put("empty.pdf", b"")
        assert (share_root / "Input" / "empty.pdf").read_bytes() == b""

    def test_spaces_and_cyrillic_are_accepted(self, fake: FakeSmbClient, share_root: Path) -> None:
        name = inbox_name(_job_id(), "Отчёт за 2024 год (копия).pdf")
        _hotfolder().put(name, b"x")
        assert (share_root / "Input" / name).read_bytes() == b"x"

    # test_rejects_invalid_names_before_any_call — без изменений

    def test_write_failure_keeps_partial_file_when_cleanup_disabled(
        self, fake: FakeSmbClient, share_root: Path
    ) -> None:
        fake.fail["open_file"] = SMBOSError(NtStatus.STATUS_ACCESS_DENIED, INPUT)
        with pytest.raises(SMBOSError):
            _hotfolder().put("a.pdf", b"data")
        assert "remove" not in fake.funcs()

    def test_write_failure_removes_partial_file_when_cleanup_enabled(
        self, fake: FakeSmbClient, share_root: Path
    ) -> None:
        fake.fail["open_file"] = SMBOSError(NtStatus.STATUS_ACCESS_DENIED, INPUT)
        with pytest.raises(SMBOSError):
            _hotfolder(cleanup=True).put("a.pdf", b"data")
        assert fake.calls[-1] == ("remove", (f"{INPUT}\\a.pdf",), {})

    def test_cleanup_error_does_not_mask_original(self, fake: FakeSmbClient, share_root: Path) -> None:
        original = SMBOSError(NtStatus.STATUS_ACCESS_DENIED, INPUT)
        fake.fail["open_file"] = original
        fake.fail["remove"] = SMBException("соединение потеряно")
        with pytest.raises(SMBOSError) as excinfo:
            _hotfolder(cleanup=True).put("a.pdf", b"data")
        assert excinfo.value is original
        assert fake.funcs() == ["mkdir", "open_file", "remove"]

    def test_passes_credentials_to_every_call(self, fake: FakeSmbClient, share_root: Path) -> None:
        _hotfolder(login="user", password="secret").put("a.pdf", b"1")
        assert [kwargs for _, _, kwargs in fake.calls] == [{"username": "user", "password": "secret"}] * 2


class TestPutOverwrite:
    def test_existing_file_is_overwritten(self, fake: FakeSmbClient, share_root: Path) -> None:
        """Как в старом АРМ (решение владельца 09.10.2026): повтор задания перезаписывает входящий файл."""
        hf = _hotfolder()
        hf.put("job.pdf", b"first, longer")
        hf.put("job.pdf", b"second")
        assert fake._local(INPUT + "\\job.pdf").read_bytes() == b"second"
        assert "rename" not in fake.funcs() and "replace" not in fake.funcs()

В TestCleanupDetails.test_orphan_part_and_done_marker_are_removed → переименовать в test_done_marker_is_removed_and_tmp_is_not_touched: .part в Tmp остаётся, removed == [f"{INPUT}\\{name}", f"{INPUT}\\{name}.done"], assert (root / "Tmp" / f".{name}.part").exists(). test_passes_credentials_to_every_call (cleanup): ["scandir", "remove", "remove", "rmdir", "remove", "remove"]. test_missing_answer_dir_and_files_are_fine: ["scandir", "remove", "remove"].

• Step 2: Прогнать — ожидаемо падают

Run: cd api && .venv/bin/pytest -q tests/unit/test_smb_hotfolder.py Expected: FAIL (TestPut, TestPutOverwrite, TestCleanupDetails).

• Step 3: Реализация smb.py

__init__: убрать параметр tmp_dir и self.tmp_dir. put:

    def put(self, name: str, data: bytes) -> None:
        """Записать `<input>\\<name>` сразу, открытием `wb` (как в старом АРМ, решение владельца 09.10.2026): временного
        файла и rename нет, существующий файл перезаписывается. `smbclient.open_file` открывает файл монопольно — пока
        запись идёт, ИС его не прочитает (ASSUMPTION(OQ-79)). Имя проверяется до любого обращения к шаре. При ошибке
        записи недописанный файл снимается только при включённой очистке (`SCAN_CLEANUP`, R-3); ошибка снятия исходную
        не маскирует. Нет корня — `SMBOSError` (ENOENT) от `mkdir`.
        """
        validate_name(name)
        self._ensure_dir(self.input_dir)
        target = ntpath.join(self.input_dir, name)
        try:
            with smbclient.open_file(target, mode="wb", **self._credentials) as fd:
                fd.write(data)
        except BaseException:
            self._drop_partial(target)
            raise

_drop_part переименовать в _drop_partial (тело то же, докстринг «недописанный входящий файл»). В cleanup из кортежа путей убрать ntpath.join(self.tmp_dir, f".{name}.part"); докстринг — «ответ из output, входящий файл и .done». Импорт NtStatus и _has_errno(..., EEXIST ...) в put больше не нужны — удалить, если не используются (_NOT_READY использует NtStatus — оставить). Докстринг модуля: «файл пишется сразу в \\<host>\<SMB_PATH>\<input_dir>\<имя> (wb, перезапись, как в старом АРМ)»; убрать упоминания tmp_dir/.part/OQ-79-rename. Комментарий noqa: PLR0913 у __init__ оставить, если ruff его требует (параметров по-прежнему > 5).

• Step 4: Прогнать

Run: cd api && .venv/bin/pytest -q tests/unit/test_smb_hotfolder.py Expected: PASS.

• Step 5: Commit

git -C /Users/andrey.epifancev/projects/awp-ca/.wt/smb-direct-write/scan-gateway add api/src/gateway/smb.py api/tests/unit/test_smb_hotfolder.py
git -C /Users/andrey.epifancev/projects/awp-ca/.wt/smb-direct-write/scan-gateway commit -m "feat(smb): запись прямо в Input с перезаписью, как в старом АРМ (#26)"

Task 2: LocalHotFolder, фабрика, настройки

Files:

• Modify: api/src/gateway/hotfolder.py (докстринг модуля, LocalHotFolder докстринг, put, удалить _open_part/_replace_part, _drop_part → _drop_partial, cleanup, _NO_HARD_LINKS)
• Modify: api/src/gateway/__init__.py (не передавать SCAN_TMP_DIR), api/src/core/config.py (SCAN_TMP_DIR удалить, SCAN_TIMEOUT_SECONDS = 1200, комментарий SCAN_CLEANUP)
• Test: api/tests/unit/test_local_hotfolder.py, api/tests/unit/test_hotfolder_factory.py, api/tests/unit/test_config.py, api/tests/integration/test_roundtrip.py

Interfaces:

• Consumes: SmbHotFolder без tmp_dir (Task 1).

• Produces: LocalHotFolder(root, input_dir="Input", output_dir="Output", tmp_dir="Tmp", *, ...) — сигнатура та же, tmp_dir используется только эмулятором; Settings без SCAN_TMP_DIR.

• Step 1: Тесты

test_local_hotfolder.py: удалить TestPutWithoutReplace, TestPartIsCreatedFresh и из TestPut — test_leaves_no_part_files, test_part_is_written_in_tmp_not_in_input, test_existing_tmp_dir_is_fine, test_custom_tmp_dir_name, test_cleans_part_file_on_failure, test_keeps_part_file_on_failure_when_cleanup_disabled, test_successful_put_despite_final_unlink_failure, test_cleanup_failure_does_not_mask_original_error; test_creates_input_and_tmp_dirs_only → test_creates_input_dir_only (sorted(p.name for p in root.iterdir()) == ["Input"]). Добавить в TestPut:

    def test_existing_file_is_overwritten(self, tmp_path: Path) -> None:
        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_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 _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, "нет прав")

TestCleanup: ожидания про Tmp/.<name>.part — .part не удаляется (файл в Tmp остаётся, если тест его создаёт). test_hotfolder_factory.py: в _local_settings убрать SCAN_TMP_DIR; assert'ы hf.tmp_dir == tmp_path / "Служебный" и hf.tmp_dir == "...\\T" удалить, для local — assert hf.tmp_dir == tmp_path / "Tmp" (умолчание эмулятора). test_config.py: строки assert s.SCAN_TMP_DIR == "Tmp" → assert not hasattr(s, "SCAN_TMP_DIR"); добавить assert s.SCAN_TIMEOUT_SECONDS == 1200 в тест умолчаний (строка 32) и тест:

def test_legacy_scan_tmp_dir_is_ignored(monkeypatch):
    monkeypatch.setenv("SCAN_TMP_DIR", "Tmp")
    s = Settings(_env_file=None)
    assert not hasattr(s, "SCAN_TMP_DIR")

(обязательные переменные — как в соседних тестах файла; взять их фикстуру/хелпер). test_roundtrip.py: убрать stray_parts, tmp_dir у объекта стенда, tmp_dir=settings.SCAN_TMP_DIR (эмулятору передать "Tmp"), докстринги про .part; в сценарии очистки — Input/ и Output/<job_id>/ пусты.

• Step 2: Прогнать — ожидаемо падают

Run: cd api && .venv/bin/pytest -q tests/unit/test_local_hotfolder.py tests/unit/test_hotfolder_factory.py tests/unit/test_config.py Expected: FAIL.

• Step 3: Реализация

LocalHotFolder.put:

    def put(self, name: str, data: bytes) -> None:
        """Записать `input_dir/<name>` сразу (`wb`, перезапись существующего, как в старом АРМ — решение владельца
        09.10.2026). `Input` создаётся без родителей, корень — нет (нет корня — `FileNotFoundError`). При ошибке записи
        недописанный файл снимается только при включённой очистке (R-3), ошибка снятия подавляется."""
        validate_name(name)
        self.input_dir.mkdir(exist_ok=True)
        target = self.input_dir / name
        try:
            target.write_bytes(data)
        except BaseException:
            self._drop_partial(target)
            raise

    def _drop_partial(self, path: Path) -> None:
        if self.cleanup_enabled:
            with contextlib.suppress(OSError):
                path.unlink(missing_ok=True)

Удалить _open_part, _replace_part, _NO_HARD_LINKS и неиспользуемые импорты (errno, os — проверить ruff). cleanup: кортеж путей без self.tmp_dir / f".{name}.part". Докстринги модуля и класса: прямая запись в Input; tmp_dir — служебный каталог эмулятора (сборка набора ответа), put его не использует. gateway/__init__.py: в LocalHotFolder(...) не передавать settings.SCAN_TMP_DIR (умолчание "Tmp"), в SmbHotFolder(...) убрать аргумент. config.py: удалить SCAN_TMP_DIR; SCAN_TIMEOUT_SECONDS: PositiveFloat = 1200 (комментарий: «20 мин, в паре со сроком ответа OCR ядра»); комментарий SCAN_CLEANUP — без .part в Tmp.

• Step 4: Прогнать unit

Run: cd api && .venv/bin/pytest -q tests/unit Expected: PASS (включая test_emulator.py — эмулятор пользуется tmp_dir).

• Step 5: Commit

git -C /Users/andrey.epifancev/projects/awp-ca/.wt/smb-direct-write/scan-gateway add -A api
git -C /Users/andrey.epifancev/projects/awp-ca/.wt/smb-direct-write/scan-gateway commit -m "feat(hotfolder): local тоже пишет прямо в Input; без SCAN_TMP_DIR; таймаут 1200 с (#26)"

Task 3: Документация и полная проверка

Files:

• Modify: README.md (схема конвейера п. 5, раздел протокола горячих папок, «Очистка», таблица env: убрать SCAN_TMP_DIR шлюза, SCAN_TIMEOUT_SECONDS = 1200, права на SMB), CLAUDE.md («Что это», «Горячие папки»), api/.env.example (SCAN_TMP_DIR — в блок эмулятора, SCAN_TIMEOUT_SECONDS), openspec/changes/hotfolder-direct-write/tasks.md (отметки)

• Step 1: Правки текста

git grep -n "Tmp\|\.part\|SCAN_TMP_DIR\|SCAN_TIMEOUT_SECONDS\|300" -- README.md CLAUDE.md api/.env.example и привести каждое место к: «файл пишется сразу в Input/<job_id>.<ext> (wb, существующий перезаписывается, как в старом АРМ; ИС не берёт файл, пока он открыт на запись — ASSUMPTION(OQ-79))»; Tmp и SCAN_TMP_DIR упоминаются только у эмулятора; права учётки шары — запись в Input, чтение Output, удаление при SCAN_CLEANUP; таймаут 1200 с.

• Step 2: Полная проверка

Run: make check-full (из корня репозитория; нужен поднятый стенд make up) Expected: exit 0. Вывод — в описание PR.

• Step 3: Commit

git -C /Users/andrey.epifancev/projects/awp-ca/.wt/smb-direct-write/scan-gateway add -A README.md CLAUDE.md api/.env.example openspec
git -C /Users/andrey.epifancev/projects/awp-ca/.wt/smb-direct-write/scan-gateway commit -m "docs: прямая запись в Input, SCAN_TMP_DIR только у эмулятора, таймаут 1200 с (#26)"