Система интерактивных сессий

ErisPulse реализует "постоянное взаимодействие с пользователем" на уровне фреймворка: от одной wait_reply до периодических напоминаний, ожидания нескольких потоков, взаимоисключающих сессий и восстановления после перезапуска — всё это управляется единым менеджером интерактивных сессий (Core/Event/interaction.py, sdk.interaction).

{!--< tips >!--} Каждая из описанных в этой статье возможностей имеет владельца (owner): ожидание, аренды, таймеры все записывают имя модуля, при регистрации, и при выгрузке модуля / отключении адаптера фреймворк автоматически очищает ожидания и уведомляет ожидаемую сторону, а не заставляет ждать истечения таймаута — это расширение системы владения в контексте взаимодействия (см. Система владения). {!--< /tips >!--}

Ожидание ответа: wait_reply

wait_reply — основа интерактивных сессий: приостанавливает текущую корутину, ожидая ответа от пользователя в следующем сообщении.

from ErisPulse.Core.Event import command

@command("ask")
async def ask_command(event):
    reply = await event.wait_reply(prompt="Введите ваше имя:", timeout=30)
    if reply is None:
        await event.reply("Таймаут")
        return
    await event.reply(f"Привет, {reply.get_text()}!")

Полный список параметров

# Принимает только цифровые суммы, иначе продолжает ожидание
reply = await event.wait_reply("Введите сумму:", regex=r"\d+\s*元", timeout=30)

# Сессионное ожидание: в сценариях совместной работы в группе, любой участник может ответить
reply = await event.wait_reply(session=True, prompt="Кто-нибудь может ответить?")

Когда ожидание будет отменено

Ожидание больше не приводит к "только ожиданию истечения таймаута" — следующие условия немедленно приводят к прекращению (возврат None из wait_reply), а не к ожиданию истечения таймаута:

Нижележащее исключение — InteractionCancelled (входит в иерархию InteractionError), wait_reply уже преобразует его в возврат None; вызывающие, которым нужна причина, могут использовать низкоуровневый API sdk.interaction.register().

Полная цепочка проверки при ответе

Когда приходит сообщение с ответом, менеджер взаимодействия проверяет его по следующей последовательности (выполняется до сопоставления команд, приоритет сохраняется — даже если сообщение уже обработано другим процессором, приостановленный диалог может быть завершен):

Соответствие сессионному ключу (точная пользовательская мера → сессионная отмена)
  → Фильтрация текста по pattern / regex (не совпадает — продолжить ожидание)
  → Проверка validator (не проходит — продолжить ожидание)
  → Повторная проверка прав (по scope и владельцу, не проходит — завершить ожидание)
  → Пробуждение ожидающей стороны + признание события (mark_processed)

Сессионные таймеры: remind / escalate

Превращает "таймаут" из возвращаемого значения в управляемый примитив. Таймеры привязаны к сессии взаимодействия и автоматически отменяются при выгрузке модуля / остановке адаптера, лимит активных remind на одну сессию — 5 шт.

remind: напоминание при отсутствии ответа

@command("ticket")
async def ticket_command(event):
    await event.reply("Заявка создана, результат будет здесь")
    # Если нет ответа за 5 минут, мягко напомнить; любой ответ от пользователя автоматически отменяет напоминание
    event.remind(300, "Есть ли кто-нибудь? С результатом свяжемся")
    reply = await event.wait_reply(timeout=3600)
    ...

• event.remind(delay, text=None, *, callback=None): отправляет text (или вызывает callback(event), поддерживает синхронные и асинхронные) по истечении delay. Обязательно: text и callback должны быть выбраны (если не указаны — выброс ValueError)
• Возвращает Reminder-дескриптор: reminder.cancel() для ручной отмены, reminder.expired для проверки состояния
• Автоматически отменяется при ответе пользователя — это и есть семантика "напоминания": напоминание появляется только при молчании пользователя
• В Conversation также доступно: conv.remind(120, "Еще думаете?")

escalate: обязательное уведомление по истечении времени

event.escalate(1800, lambda e: notify_master(f"Заявка не обработана 30 минут: {event.get_command_args()}"))

Единственное отличие от remind: не отменяется ответом пользователя — действие "повышения" (уведомление владельца, передача в ручной обход) — это обязательное уведомление по истечении времени, отменяется только вручную cancel() / выгрузкой модуля / остановкой адаптера.

Многопоточное ожидание: expect + select

Ожидание нескольких условий одновременно, первое поступившее — выполняется — типичный сценарий: ожидание одобрения администратором и одновременно ожидание отмены пользователем, голосование между участниками.

which, reply = await event.select(
    event.expect(pattern="同意*", user="10001"),
    event.expect(pattern="拒绝*", user="10002"),
    event.expect(validator=lambda e: e.get_text() == "搁置", session=True),
    timeout=60,
)
if which is None:
    await event.reply("В течение 60 секунд не было ответов")
elif which == 0:
    await event.reply("Одобрено")
elif which == 1:
    await event.reply("Отклонено")

• event.expect(...) создаёт описание ожидания (не регистрирует ожидание): поддерживает pattern / regex / validator / user (ограничение пользователя, отвечающего) / session (любой может ответить)
• event.select(*expectations, timeout=60): регистрирует все ожидания → возвращает (индекс, событие ответа) при первом совпадении → несовпадающие ожидания автоматически отменяются; при истечении таймаута возвращаются (None, None). Обязательно: передать хотя бы одно ожидание, иначе выброс ValueError
• Событие, которое было захвачено, уже признано фреймворком (mark_processed), не будет повторно обработано другими процессорами

{!--< tips >!--} select по сравнению с ручной обработкой asyncio.wait в многопоточном режиме: автоматическая очистка несовпадающих ожиданий, автоматическое признание захваченного события, все проверки прав и очистка владения автоматически выполняются — не нужно управлять Future вручную. {!--< /tips >!--}

Взаимоисключающие сессии: acquire / hold / get_owner_of

Владение переходит от "ресурсов" к "сессиям" — "кто сейчас занят этим пользователем" становится первостепенным вопросом.

# Запрос: кто сейчас взаимодействует с этой сессией? (если свободна — возвращает None)
owner = sdk.interaction.get_owner_of(event)
if owner and owner != "MyModule":
    return  # Другой модуль уже в диалоге, избегаем мешать

# Взаимоисключающая аренда: эксклюзивная сессия (политика deny, если занята — возвращает None)
lease = sdk.interaction.acquire(event)          # По умолчанию TTL 1 час, можно передать ttl=
if lease is None:
    return  # Уже занята
try:
    ...  # Эксклюзивное взаимодействие
finally:
    lease.release()

Форма контекстного менеджера (при неудаче выбрасывает SessionOccupiedError):

with sdk.interaction.hold(event) as lease:
    ...  # Автоматически освобождается при выходе

Аренда поддерживает renew(ttl) для продления; TTL ленивый — просроченная аренда автоматически удаляется при следующем обращении.

Conversation.resume() при возобновлении диалога автоматически приобретает аренду (см. Многошаговый диалог "Восстановление = захват") — восстановленный диалог автоматически владеет сессией, другие модули не могут вмешаться.

Почтовый ящик сессии: event.history

Единая запись недавних сообщений в сессии (со стороны пользователя и бота), используется как общая база контекста для AI, предотвращения повторов, анализа поведения — модули больше не хранят историю отдельно.

messages = await event.history(20)   # Последние 20 сообщений в сессии, в порядке возрастания времени
for m in messages:
    print(m["role"], ":", m["text"])  # role: "user" / "bot"

• Автоматически записывается: входящие сообщения (role=user) + исходящие сообщения бота (role=bot)
• Хранение: отдельная SQLite-таблица, политика хранения = лимит на сессию (по умолчанию 50) + глобальный TTL (по умолчанию 7 дней)
• Настройка: ErisPulse.transcript = {enabled = true, max_per_session = 50, ttl_hours = 168}
• API менеджера: sdk.transcript.append() / get() / clear()

Транзакции сообщений: message_tx

Все исходящие отправки в транзакции автоматически записываются; при исключительном выходе автоматически отменяются отправленные сообщения (если адаптер не реализует delete_message, пропускается, но журнал записей остаётся).

async with event.message_tx():
    await event.reply("Обрабатываю, подождите")
    result = await do_something()          # Здесь выбрасывается исключение →
    await event.reply(f"Готово: {result}")   # Сообщение "Обрабатываю..." автоматически удаляется

Отправка вне транзакции не записывается (нулевой накладной расход); get_send_receipts() позволяет просмотреть отправленные в текущей транзакции сообщения.

Трассировка цепочки: trace-id

Каждое входящее событие автоматически получает идентификатор трассировки (используется event["id"], если отсутствует — генерируется), который проходит через:

• Контекст обработчика (get_current_trace_id() для чтения)
• Исходящие отправки (строки лога [Send] с дополнением [trace:...], поле trace_id в хуках message.sending/sent)
• Данные жизненного цикла (dict автоматически дополняется _trace_id)
• Направленные события (lifecycle.emit(..., to=...)) и транзакционные подтверждения сообщений

При обработке сообщения несколькими модулями, один и тот же ID можно использовать для объединения всей цепочки (логи / медленные запросы / аудит).

Связь с другими системами

• Владение: ожидания / аренды / таймеры все записывают owner, при выгрузке — очищаются (Система владения)
• Область видимости: проверка прав при совпадении ответа + по модулю; аудит при межмодульных вызовах — по направлению отправки (Область видимости)
• Conversation: многошаговый диалог — это конечный автомат на уровне сессии взаимодействия (Conversation), его ожидания также имеют все отмены / проверки / владения, описанные на этой странице

Связанные документы

• Многошаговый диалог (Conversation) — конечный автомат, автоматические контрольные точки и восстановление после перезапуска
• Система владения (owner) — обзор и границы дизайна очистки владения
• Область видимости (scope) — конфигурация проверки прав и аудита при отправке
• Взаимодействие между модулями — межмодульные вызовы и направленные события