Design: Потоковая синхронизация почты (IMAP IDLE) и онлайновая копия ящика

Context

Сервер: mail.corpoffice.tech:143 (STARTTLS), Microsoft Exchange. Проверено 2026-09-15 (ручной IMAP-тест):
- CAPABILITY: IMAP4 IMAP4rev1 AUTH=PLAIN AUTH=NTLM AUTH=GSSAPI STARTTLS SASL-IR UIDPLUS MOVE ID UNSELECT CHILDREN IDLE NAMESPACE LITERAL+
- IDLE, MOVE, UIDPLUS, CHILDREN, LITERAL+ — всё есть, значит потоковая синхронизация и отслеживание перемещений возможны.
- Логин с чистого скрипта (LOGIN / AUTHENTICATE PLAIN) не прошёл (NO LOGIN failed). himalaya логинится успешно — вероятно, проблема в способе авторизации/лимитах для «незнакомого» клиента, а не в пароле. В design — использовать существующий himalaya-конфиг и проверенный путь.
Текущий архиватор: scripts/mail_archive.py — poll каждые 5 мин (Hermes cron mail-archive-every-5min, id 5f2305b2bbf8). Отслеживает last_uid по папкам в /opt/hermes/email/state/mail-archive-last-<folder>.json. Уже есть: fetch_attachments_imaplib() (сырой IMAP, BODY.PEEK[], не ставит \Seen), get_inbox_subfolders() (динамическое обнаружение 136 папок, CHILDREN), _imap_utf7_encode() (modified UTF-7 для кириллических папок).
Архив: /opt/hermes/email/<folder>/YYYY/MM/<uid>/email.md (frontmatter: id, folder, subject, from, to, date, flags, message_id, in_reply_to, references, cc, content_type; + classification, handled_*, has_attachment). 5467 писем на 2026-09-15.
Индексы: scripts/mail_index.py (SQLite FTS5 /opt/hermes/email/mail_index.db), sqlite_search.py, классификатор email_classifier.py (Qwen3:8b, Ollama), обработчики email_handlers.py (urgent→TG, task→VTODO, meeting→VEVENT), digest.py (еженедельный).
Пользователь активно меняет ящик в клиенте: переносит в папки, удаляет, отвечает. Эти изменения сейчас НЕ видны архиватору.

Решение

1. Выбор модели: 2 постоянных процесса (systemd user units)

Не один процесс на всё, а два — разделение ответственности (как ты предложил):

email-imap-stream.service — держит IMAP-соединение, IDLE, пишет события в SQLite (mailbox_events), архивирует новые письма. Единственный процесс, который разговаривает с IMAP (кроме fallback-cron).
email-change-analyzer.service — постоянно читает mailbox_events, обновляет online-копию (mailbox_state), запускает классификатор/обработчики, индексы, может писать «журнал изменений» для пользователя.

Почему 2, а не 1: падение анализатора не теряет события (они в ChangeLog, append-only); падение stream-процесса не останавливает анализ (fallback-cron докачает письма). Отвязка скорости IMAP от скорости LLM-анализа.

2. Библиотека для IMAP stream

Варианты:

(a) aioimaplib — Python asyncio IMAP с поддержкой IDLE, reconnect, ивенты. Мягкая зависимость; если нет — pip install. Совместим с Exchange (IDLE есть).
(b) Сырой socket (как уже сделано в fetch_attachments_imaplib) + select() на IDLE — без новых зависимостей, но больше кода (reconnect, парсинг untagged-ответов, литералы).

Выбор: (a) aioimaplib — проверенная библиотека, IDLE/reconnect из коробки; сырой IMAP оставляем только в fetch_attachments_imaplib (там он уже работает и менять не нужно). Если aioimaplib недоступен/не заводится на нашем Python — fallback (b).

Авторизация: переиспользовать учётку из config/himalaya-config.toml (login e.storozhenko, host mail.corpoffice.tech). Для stream-процесса — AUTH=PLAIN с SASL-IR (Exchange поддерживает; но в тесте не прошло — проверить в задаче 2: использовать himalaya account sync/envelope list как эталон, возможно, потребуется \r\n-формат или конкретный порядок SASL).

3. Схема SQLite (новая БД или таблицы в существующей)

Отдельная БД: /opt/hermes/email/state/mailbox.db (не смешивать с mail_index.db, чтобы не ломать FTS-индексы и было легко откатиться).

-- Онлайновая копия ящика: текущее состояние каждого письма
CREATE TABLE mailbox_state (
    uid INTEGER NOT NULL,          -- UID в папке
    folder TEXT NOT NULL,          -- папка (например 'INBOX', 'INBOX/Проекты')
    message_id TEXT,               -- Message-ID
    in_reply_to TEXT,
    references TEXT,               -- цепочка (References / In-Reply-To)
    subject TEXT,
    from_addr TEXT,
    date TEXT,
    flags TEXT,                    -- JSON-массив флагов IMAP
    has_attachment INTEGER DEFAULT 0,
    archive_path TEXT,             -- путь к email.md (если заархивировано)
    etag TEXT,                     -- ETag сервера (возможно, не нужен для IMAP)
    last_seen TEXT,                -- ISO-время последней синхронизации
    deleted INTEGER DEFAULT 0,     -- soft-delete (пользователь удалил на сервере)
    PRIMARY KEY (folder, uid)
);

-- Журнал изменений (ChangeLog), append-only
CREATE TABLE mailbox_events (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    ts TEXT NOT NULL,              -- ISO-время события
    event TEXT NOT NULL,           -- added | moved | deleted | flag_changed | replied | reconcile
    folder TEXT,
    uid INTEGER,
    message_id TEXT,
    details TEXT                   -- JSON: from_folder, to_folder, old_flags, new_flags и т.п.
);
CREATE INDEX idx_events_message ON mailbox_events(message_id);
CREATE INDEX idx_events_ts ON mailbox_events(ts);

4. Алгоритм stream-процесса (`imap_stream.py`)

loop:
    1. connect + login (AUTH=PLAIN / STARTTLS)
    2. folder list (get_inbox_subfolders через IMAP, dynamic)
    3. для каждой папки: SELECT; если нет в mailbox_state — полный reconcile
       (UID FETCH 1:* (FLAGS UID MESSAGE-ID ...) = initial sync)
    4. reconcile источником истины: для INBOX + подпапок
       - UID FETCH (новые UID > last_uid) → событие added, архивация
       - сравнение FLAGS (в т.ч. \Seen, \Answered, \Flagged) → flag_changed
       - UID SEARCH EXPUNGE / отсутствие в SELECT → deleted (если был в state)
    5. IDLE loop (для INBOX, и по очереди для подпапок если CPU позволяет):
       - IDLE → ждём untagged: EXISTS (новое письмо), EXPUNGE (удаление),
         FETCH FLAGS (изменение флагов), MOVE (если сервер шлёт)
       - на событие: обработать (архивировать/обновить state/записать event)
       - продлевать IDLE каждые ~29 мин (сервер обычно рвёт после 30)
    6. при обрыве: reconnect + reconcile (REQ-IMAP-SYNC-001/002)
    7. период паузы между reconcile: 60-300 сек (конфигурируемо)

Архивация новых писем — переиспользуем существующий код mail_archive.py: get_email_content() (himalaya message read --preview, не ставит Seen) + fetch_attachments_imaplib() (BODY.PEEK[]). Функции вынести в общий модуль или импортировать (mail_archive.py уже модульный).

Удаление: при событии EXPUNGE/deleted — НЕ удаляем файл email.md (письмо остаётся в архиве, REQ-IMAP-SYNC-004 soft-delete), помечаем deleted=1 в mailbox_state, пишем событие deleted в ChangeLog. Пользователь сможет видеть «это письмо вы удалили» в RAG/поиске.

Перемещение: при обнаружении (MOVE/UIDPLUS от сервера или reconcile: UID в старой папке исчез, в новой появился с тем же Message-ID) — обновить folder, записать moved с from/to. Если UID меняется (Exchange MOVE) — следить по Message-ID.

5. Анализатор (`change_analyzer.py`)

loop:
    1. читать mailbox_events от последнего обработанного id (offset в state)
    2. для каждого события:
       added      → классифицировать (email_classifier), обработать (email_handlers),
                    обновить индексы (mail_index --incremental)
       flag_changed → обновить flags в mailbox_state; если \Answered → событие replied
       moved      → обновить folder; если классификация была — можно переклассифицировать
       deleted    → пометить в RAG/индексе удалённым (не удаляя файл)
    3. записать обработанный id (persistent)
    4. спать 1-5 сек (или ждать сигнала от stream через очередь)

Связь stream ↔ analyzer: через SQLite (ChangeLog) — это проще и надёжнее, чем IPC/очереди. Stream пишет, analyzer читает с offset. Несколько analyser процессов не нужны (одна учётка).

6. Управление / systemd

Два user unit (в ~/.config/systemd/user/):

# email-imap-stream.service
[Unit]
Description=Email IMAP realtime sync stream
After=network-online.target

[Service]
Type=simple
ExecStart=/usr/bin/python3 /opt/hermes/email-assistant/scripts/imap_stream.py
Restart=always
RestartSec=5
Environment=HOME=/home/estorozhenko   # для himalaya (конфиг там)

[Install]
WantedBy=default.target

Аналогично email-change-analyzer.service. Включить: systemctl --user enable --now.

Hermes cron mail-archive-every-5min → оставить как fallback: если оба сервиса не работают (например, после ребута без автозапуска), cron всё равно архивирует раз в 5 мин (REQ-IMAP-SYNC-007). Чтобы не дублировать: stream-процесс и cron оба идемпотентны (last_uid / mailbox_state).

7. Что НЕ делаем (ограничения)

НЕ переписываем архиватор под IMAP с нуля — используем существующий mail_archive.py как библиотеку/подпроцесс.
НЕ удаляем файлы писем при удалении на сервере (RAG должен видеть «удалено», но данные не теряем).
НЕ реалтайм-синхронизация «один к одному» всех 137 папок через отдельные IDLE-коннекты — IDLE держим на INBOX (главный источник), подпапки — через reconcile (раз в 1-5 мин) + CHILDREN-обнаружение. Это баланс скорости и ресурсов (Exchange лимитирует коннекты на учётку).
НЕ реализуем SMTP/отправку — только чтение (как сейчас).

8. Риски и смягчение

Риск	Смягчение
Exchange рвёт IDLE после ~30 мин	авто-reconnect + reconcile после каждого обрыва
LOGIN с чистого скрипта не прошёл	использовать himalaya-путь; AUTH=PLAIN; задача 2 — проверить формат
IDLE не гарантирует доставку всех событий	reconcile каждые 1-5 мин (REQ-002)
MOVE меняет UID на Exchange	следить по Message-ID, а не UID
Много папок (137) — много коннектов	IDLE только на INBOX; подпапки — reconcile по очереди
aioimaplib не установлен	fallback на сырой socket (уже есть паттерн)

13 KiB Raw Blame History Unescape Escape