# Запись в горячую папку прямо в 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` заменить на:

```python
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`:

```python
    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**

```bash
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`:

```python
    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")
```

с модульными помощниками:

```python
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) и тест:

```python
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`:

```python
    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**

```bash
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**

```bash
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)"
```