Событийные хуки

VibeOS предоставляет три системы хуков, которые запускают собственный код в ключевых точках жизненного цикла:

Система	Регистрация	Где выполняется	Назначение
Шлюзовые хуки	HOOK.yaml + handler.py в ~/.vibeos/hooks/	Только шлюз	Логирование, оповещения, вебхуки
Плагинные хуки	ctx.register_hook() в плагине	CLI + Шлюз	Перехват инструментов, метрики, ограничения
Шелловые хуки	Блок hooks: в ~/.vibeos/config.yaml, указывающий на шелл-скрипты	CLI + Шлюз	Готовые скрипты для блокировки, автоформатирования, внедрения контекста

Все три системы неблокирующие — ошибки в любом хуке перехватываются и логируются, никогда не приводя к сбою агента.

Шлюзовые событийные хуки

Шлюзовые хуки срабатывают автоматически во время работы шлюза (Telegram, Discord, Slack, WhatsApp, Teams), не блокируя основной конвейер агента.

Создание хука

Каждый хук представляет собой директорию в ~/.vibeos/hooks/, содержащую два файла:

~/.vibeos/hooks/
└── my-hook/
    ├── HOOK.yaml      # Объявляет, какие события слушать
    └── handler.py     # Обработчик на Python

HOOK.yaml

name: my-hook
description: Логировать всю активность агента в файл
events:
  - agent:start
  - agent:end
  - agent:step

Список events определяет, какие события запускают ваш обработчик. Вы можете подписаться на любую комбинацию событий, включая шаблоны вроде command:*.

handler.py

import json
from datetime import datetime
from pathlib import Path

LOG_FILE = Path.home() / ".vibeos" / "hooks" / "my-hook" / "activity.log"

async def handle(event_type: str, context: dict):
    """Вызывается для каждого подписанного события. Должна называться 'handle'."""
    entry = {
        "timestamp": datetime.now().isoformat(),
        "event": event_type,
        **context,
    }
    with open(LOG_FILE, "a") as f:
        f.write(json.dumps(entry) + "\n")

Правила для обработчика:

• Должен называться handle
• Получает event_type (строка) и context (словарь)
• Может быть async def или обычным def — оба варианта работают
• Ошибки перехватываются и логируются, никогда не приводя к сбою агента

Доступные события

Событие	Когда срабатывает	Ключи контекста
gateway:startup	Запуск процесса шлюза	platforms (список активных платформ)
session:start	Создана новая сессия обмена сообщениями	platform, user_id, session_id, session_key
session:end	Сессия завершена (до сброса)	platform, user_id, session_key
session:reset	Пользователь выполнил /new или /reset	platform, user_id, session_key
agent:start	Агент начинает обработку сообщения	platform, user_id, session_id, message
agent:step	Каждая итерация цикла вызова инструментов	platform, user_id, session_id, iteration, tool_names
agent:end	Агент завершает обработку	platform, user_id, session_id, message, response
command:*	Выполнена любая slash-команда	platform, user_id, command, args

Сопоставление по шаблону

Обработчики, зарегистрированные для command:*, срабатывают для любого события command: (command:model, command:reset и т.д.). Отслеживайте все slash-команды с помощью одной подписки.

Примеры

Оповещение в Telegram о долгих задачах

Отправьте себе сообщение, когда агент выполняет более 10 шагов:

# ~/.vibeos/hooks/long-task-alert/HOOK.yaml
name: long-task-alert
description: Оповещение, когда агент делает много шагов
events:
  - agent:step

# ~/.vibeos/hooks/long-task-alert/handler.py
import os
import httpx

THRESHOLD = 10
BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN")
CHAT_ID = os.getenv("TELEGRAM_HOME_CHANNEL")

async def handle(event_type: str, context: dict):
    iteration = context.get("iteration", 0)
    if iteration == THRESHOLD and BOT_TOKEN and CHAT_ID:
        tools = ", ".join(context.get("tool_names", []))
        text = f"⚠️ Агент работает уже {iteration} шагов. Последние инструменты: {tools}"
        async with httpx.AsyncClient() as client:
            await client.post(
                f"https://api.telegram.org/bot{BOT_TOKEN}/sendMessage",
                json={"chat_id": CHAT_ID, "text": text},
            )

Логгер использования команд

Отслеживайте, какие slash-команды используются:

# ~/.vibeos/hooks/command-logger/HOOK.yaml
name: command-logger
description: Логировать использование slash-команд
events:
  - command:*

# ~/.vibeos/hooks/command-logger/handler.py
import json
from datetime import datetime
from pathlib import Path

LOG = Path.home() / ".vibeos" / "logs" / "command_usage.jsonl"

def handle(event_type: str, context: dict):
    LOG.parent.mkdir(parents=True, exist_ok=True)
    entry = {
        "ts": datetime.now().isoformat(),
        "command": context.get("command"),
        "args": context.get("args"),
        "platform": context.get("platform"),
        "user": context.get("user_id"),
    }
    with open(LOG, "a") as f:
        f.write(json.dumps(entry) + "\n")

Вебхук при старте сессии

Отправьте POST-запрос во внешний сервис при создании новых сессий:

# ~/.vibeos/hooks/session-webhook/HOOK.yaml
name: session-webhook
description: Уведомлять внешний сервис о новых сессиях
events:
  - session:start
  - session:reset

# ~/.vibeos/hooks/session-webhook/handler.py
import httpx

WEBHOOK_URL = "https://your-service.example.com/vibeos-events"

async def handle(event_type: str, context: dict):
    async with httpx.AsyncClient() as client:
        await client.post(WEBHOOK_URL, json={
            "event": event_type,
            **context,
        }, timeout=5)

Учебное пособие: BOOT.md — Запуск стартового чек-листа при каждой загрузке шлюза

Популярный паттерн из сообщества: поместите Markdown-чек-лист в ~/.vibeos/BOOT.md, и агент будет запускать его один раз при каждом старте шлюза. Полезно для сценариев вроде «при каждой загрузке проверять ночные сбои cron и пинговать меня в Discord, если что-то пошло не так» или «суммировать последние 24 часа deploy.log и публиковать в Slack #ops».

Этот учебный материал покажет, как создать такой хук самостоятельно. VibeOS не поставляется со встроенным хуком BOOT.md — вы настраиваете именно то поведение, которое вам нужно.

Что мы создаем

1. Файл ~/.vibeos/BOOT.md с инструкциями по запуску на естественном языке.
2. Шлюзовый хук, который срабатывает на gateway:startup, порождает одноразового агента с разрешенной моделью/учетными данными вашего шлюза и выполняет инструкции из BOOT.md.
3. Соглашение [SILENT], чтобы агент мог отказаться от отправки сообщения, когда нечего сообщать.

Шаг 1: Напишите свой чек-лист

Создайте ~/.vibeos/BOOT.md. Пишите так, как если бы давали инструкции человеку-ассистенту:

# Стартовый чек-лист

1. Выполни `vibeos cron list` и проверь, не было ли сбоев запланированных задач за ночь.
2. Если были сбои, отправь сводку в Discord #ops с помощью инструмента `send_message`.
3. Проверь, есть ли в `/opt/app/deploy.log` строки с ERROR за последние 24 часа. Если да, суммируй их и включи в то же сообщение в Discord.
4. Если ничего не пошло не так, ответь только `[SILENT]`, чтобы сообщение не отправлялось.

Агент видит это как часть своего промпта, поэтому работает всё, что можно описать на естественном языке — вызовы инструментов, шелловые команды, отправка сообщений, суммирование файлов.

Шаг 2: Создайте хук

~/.vibeos/hooks/boot-md/
├── HOOK.yaml
└── handler.py

~/.vibeos/hooks/boot-md/HOOK.yaml

name: boot-md
description: Запускать ~/.vibeos/BOOT.md при старте шлюза
events:
  - gateway:startup

~/.vibeos/hooks/boot-md/handler.py

"""Запускать ~/.vibeos/BOOT.md при каждом старте шлюза."""

import logging
import threading
from pathlib import Path

logger = logging.getLogger("hooks.boot-md")

BOOT_FILE = Path.home() / ".vibeos" / "BOOT.md"


def _build_prompt(content: str) -> str:
    return (
        "Вы выполняете стартовый чек-лист. Следуйте инструкциям "
        "ниже точно.\n\n"
        "---\n"
        f"{content}\n"
        "---\n\n"
        "Выполните каждую инструкцию. Используйте инструмент send_message для доставки "
        "сообщений на платформы вроде Discord или Slack.\n"
        "Если ничего не требует внимания и нечего сообщать, ответьте "
        "ТОЛЬКО: [SILENT]"
    )


def _run_boot_agent(content: str) -> None:
    """Порождает одноразового агента и выполняет чек-лист.

    Использует разрешенную модель шлюза и учетные данные времени выполнения,
    чтобы это работало с пользовательскими конечными точками, агрегаторами
    и OAuth-провайдерами.
    """
    try:
        from gateway.run import _resolve_gateway_model, _resolve_runtime_agent_kwargs
        from run_agent import AIAgent

        agent = AIAgent(
            model=_resolve_gateway_model(),
            **_resolve_runtime_agent_kwargs(),
            platform="gateway",
            quiet_mode=True,
            skip_context_files=True,
            skip_memory=True,
            max_iterations=20,
        )
        result = agent.run_conversation(_build_prompt(content))
        response = (result.get("final_response", "") or "").strip()
        if response.upper() not in {"[SILENT]", "SILENT", "NO_REPLY", "NO REPLY"}:
            logger.info("boot-md завершен: %s", response[:200])
        else:
            logger.info("boot-md завершен (нечего сообщать)")
    except Exception as e:
        logger.error("boot-md агент не удался: %s", e)


async def handle(event_type: str, context: dict) -> None:
    if not BOOT_FILE.exists():
        return
    content = BOOT_FILE.read_text(encoding="utf-8").strip()
    if not content:
        return

    logger.info("Запуск BOOT.md (%d символов)", len(content))

    # Фоновый поток, чтобы запуск шлюза не блокировался полным циклом агента.
    thread = threading.Thread(
        target=_run_boot_agent,
        args=(content,),
        name="boot-md",
        daemon=True,
    )
    thread.start()

Две ключевые строки:

• _resolve_gateway_model() читает текущую настроенную модель шлюза.
• _resolve_runtime_agent_kwargs() разрешает учетные данные провайдера так же, как это делает обычный цикл шлюза — включая API-ключи, базовые URL, OAuth-токены и пулы учетных данных.

Без них простой AIAgent() вернется к встроенным значениям по умолчанию и получит ошибку 401 при обращении к любой нестандартной конечной точке.

Шаг 3: Протестируйте

Перезапустите шлюз:

vibeos gateway restart

Следите за логами:

vibeos logs --follow --level INFO | grep boot-md

Вы должны увидеть Запуск BOOT.md (N символов), за которым следует либо boot-md завершен: ... (сводка того, что сделал агент), либо boot-md завершен (нечего сообщать), когда агент ответил точным токеном тишины, таким как [SILENT].

Удалите ~/.vibeos/BOOT.md, чтобы отключить чек-лист — хук останется загруженным, но будет молча пропускать выполнение, когда файла нет.

Расширение паттерна

• Чек-листы с учетом расписания: используйте datetime.now().weekday() внутри инструкций BOOT.md («если понедельник, также проверь еженедельный лог деплоя»). Инструкции — это свободный текст, поэтому агент может обработать всё, что способен понять.
• Несколько чек-листов: укажите хуку на другой файл (STARTUP.md, MORNING.md и т.д.) и зарегистрируйте отдельные директории хуков для каждого.
• Вариант без агента: если вам не нужен полный цикл агента, пропустите AIAgent и отправляйте фиксированное уведомление напрямую через httpx. Дешевле, быстрее и не зависит от провайдера.

Почему это не встроенная функция

Ранняя версия VibeOS поставляла это как встроенный хук и молча порождала агента с настройками по умолчанию при каждой загрузке шлюза. Это удивляло пользователей с пользовательскими конечными точками и делало функцию невидимой для тех, кто не знал о её существовании. Оставляя это как документированный паттерн — созданный вами, в вашей директории хуков — вы видите, что именно он делает, и включаете его, создавая файлы.

Как это работает

1. При запуске шлюза HookRegistry.discover_and_load() сканирует ~/.vibeos/hooks/
2. Каждая поддиректория с HOOK.yaml + handler.py загружается динамически
3. Обработчики регистрируются для объявленных событий
4. В каждой точке жизненного цикла hooks.emit() запускает все подходящие обработчики
5. Ошибки в любом обработчике перехватываются и логируются — сломанный хук никогда не приводит к сбою агента

к сведению

Шлюзовые хуки срабатывают только в шлюзе (Telegram, Discord, Slack, WhatsApp, Teams). CLI не загружает шлюзовые хуки. Для хуков, работающих везде, используйте плагинные хуки.

Плагинные хуки

Плагины могут регистрировать хуки, которые срабатывают в сессиях как CLI, так и шлюза. Они регистрируются программно через ctx.register_hook() в функции register() вашего плагина.

Подробнее об упаковке и регистрации плагинов см. в руководстве по плагинам.

def register(ctx):
    ctx.register_hook("pre_tool_call", my_tool_observer)
    ctx.register_hook("post_tool_call", my_tool_logger)
    ctx.register_hook("pre_llm_call", my_memory_callback)
    ctx.register_hook("post_llm_call", my_sync_callback)
    ctx.register_hook("on_session_start", my_init_callback)
    ctx.register_hook("on_session_end", my_cleanup_callback)

Общие правила для всех хуков:

• Колбэки получают именованные аргументы. Всегда принимайте **kwargs для обратной совместимости — в будущих версиях могут быть добавлены новые параметры без нарушения работы вашего плагина.
• Если колбэк аварийно завершается, он логируется и пропускается. Другие хуки и агент продолжают нормальную работу. Некорректный плагин никогда не может сломать агента.
• Возвращаемые значения двух хуков влияют на поведение: pre_tool_call может блокировать инструмент, а pre_llm_call может внедрять контекст в вызов LLM. Все остальные хуки — это наблюдатели типа «забыл и забыл».
• Колбэки-наблюдатели автоматически получают telemetry_schema_version. Когда он присутствует, turn_id, api_request_id, task_id, session_id и api_call_count являются отдельными полями корреляции. Относитесь к api_request_id как к непрозрачному идентификатору; не разбирайте его строковый формат.

Краткий справочник

Хук	Срабатывает когда	Возвращает
pre_tool_call	Перед выполнением любого инструмента	{"action": "block", "message": str} для блокировки вызова
post_tool_call	После возврата любого инструмента	игнорируется
pre_llm_call	Один раз за цикл, перед циклом вызова инструментов	{"context": str} для добавления контекста к сообщению пользователя
post_llm_call	Один раз за цикл, после цикла вызова инструментов	игнорируется
on_session_start	Создана новая сессия (только первый цикл)	игнорируется
on_session_end	Сессия завершается	игнорируется
on_session_finalize	CLI/шлюз завершает активную сессию (сброс, сохранение, статистика)	игнорируется
on_session_reset	Шлюз заменяет ключ сессии (например, /new, /reset)	игнорируется
subagent_start	Дочерний агент delegate_task создан и готов к запуску	игнорируется
subagent_stop	Дочерний агент delegate_task завершил работу	игнорируется
pre_gateway_dispatch	Шлюз получил сообщение пользователя, до аутентификации и диспетчеризации	{"action": "skip" | "rewrite" | "allow", ...} для управления потоком
pre_approval_request	Опасная команда требует одобрения пользователя, до отправки запроса/уведомления	игнорируется
post_approval_response	Пользователь ответил на запрос одобрения (или истекло время ожидания)	игнорируется
pre_api_request	Непосредственно перед исходящим HTTP/SDK-запросом к LLM	игнорируется (наблюдатель)
post_api_request	После успешного возврата запроса к LLM	игнорируется (наблюдатель)
api_request_error	После неудачного запроса к LLM (сеть/ошибка API)	игнорируется (наблюдатель)
transform_tool_result	После возврата любого инструмента, до передачи результата модели	str для замены результата, None для оставления без изменений
transform_terminal_output	Внутри инструмента terminal, до усечения/удаления ANSI/редактирования	str для замены сырого вывода, None для оставления без изменений
transform_llm_output	После завершения цикла вызова инструментов, до доставки финального ответа	str для замены текста ответа, None/пусто для оставления без изменений
kanban_task_claimed	Диспетчер назначил задачу канбана (до запуска воркера)	игнорируется (наблюдатель; процесс диспетчера)
kanban_task_completed	Задача канбана выполнена	игнорируется (наблюдатель; обычно процесс воркера)
kanban_task_blocked	Задача канбана заблокирована	игнорируется (наблюдатель; обычно процесс воркера)

---

pre_tool_call

Срабатывает непосредственно перед выполнением каждого инструмента — как встроенного, так и инструментов плагинов.

Сигнатура колбэка:

def my_callback(tool_name: str, args: dict, task_id: str, **kwargs):

Параметр	Тип	Описание
tool_name	str	Имя инструмента, который будет выполнен (например, "terminal", "web_search", "read_file")
args	dict	Аргументы, переданные модели инструменту
task_id	str	Идентификатор сессии/задачи. Пустая строка, если не задан.

Срабатывает: В model_tools.py, внутри handle_function_call(), до запуска обработчика инструмента. Срабатывает один раз на вызов инструмента — если модель вызывает 3 инструмента параллельно, срабатывает 3 раза.

Возвращаемое значение — блокировка вызова:

return {"action": "block", "message": "Причина блокировки вызова инструмента"}

Агент прерывает выполнение инструмента, возвращая message в качестве ошибки модели. Побеждает первая подходящая блокирующая директива (сначала регистрируются плагины Python, затем шелловые хуки). Любое другое возвращаемое значение игнорируется, поэтому существующие колбэки-наблюдатели продолжают работать без изменений.

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

Пример — журнал аудита вызовов инструментов:

import json, logging
from datetime import datetime

logger = logging.getLogger(__name__)

def audit_tool_call(tool_name, args, task_id, **kwargs):
    logger.info("TOOL_CALL session=%s tool=%s args=%s",
                task_id, tool_name, json.dumps(args)[:200])

def register(ctx):
    ctx.register_hook("pre_tool_call", audit_tool_call)

Пример — предупреждение об опасных инструментах:

DANGEROUS = {"terminal", "write_file", "patch"}

def warn_dangerous(tool_name, **kwargs):
    if tool_name in DANGEROUS:
        print(f"⚠ Выполняется потенциально опасный инструмент: {tool_name}")

def register(ctx):
    ctx.register_hook("pre_tool_call", warn_dangerous)

---

post_tool_call

Срабатывает непосредственно после возврата каждого выполненного инструмента.

Сигнатура колбэка:

def my_callback(tool_name: str, args: dict, result: str, task_id: str,
                duration_ms: int, **kwargs):

Параметр	Тип	Описание
tool_name	str	Имя только что выполненного инструмента
args	dict	Аргументы, переданные модели инструменту
result	str	Возвращаемое значение инструмента (всегда строка JSON)
task_id	str	Идентификатор сессии/задачи. Пустая строка, если не задан.
duration_ms	int	Время выполнения диспетчеризации инструмента в миллисекундах (измеряется с помощью time.monotonic() вокруг registry.dispatch()).

Срабатывает: В model_tools.py, внутри handle_function_call(), после возврата обработчика инструмента. Срабатывает один раз на вызов инструмента. Не срабатывает, если инструмент вызвал необработанное исключение (ошибка перехватывается и возвращается как строка JSON с ошибкой, а post_tool_call срабатывает с этой строкой ошибки в качестве result).

Возвращаемое значение: Игнорируется.

Варианты использования: Логирование результатов инструментов, сбор метрик, отслеживание успешности/неудач инструментов, панели задержек, оповещения о бюджете для каждого инструмента, отправка уведомлений при завершении определенных инструментов.

Пример — отслеживание метрик использования инструментов:

from collections import Counter, defaultdict
import json

_tool_counts = Counter()
_error_counts = Counter()
_latency_ms = defaultdict(list)

def track_metrics(tool_name, result, duration_ms=0, **kwargs):
    _tool_counts[tool_name] += 1
    _latency_ms[tool_name].append(duration_ms)
    try:
        parsed = json.loads(result)
        if "error" in parsed:
            _error_counts[tool_name] += 1
    except (json.JSONDecodeError, TypeError):
        pass

def register(ctx):
    ctx.register_hook("post_tool_call", track_metrics)

---

pre_llm_call

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

Сигнатура колбэка:

def my_callback(session_id: str, user_message: str, conversation_history: list,
                is_first_turn: bool, model: str, platform: str, **kwargs):

Параметр	Тип	Описание
session_id	str	Уникальный идентификатор текущей сессии
user_message	str	Исходное сообщение пользователя для этого цикла (до внедрения навыков)
conversation_history	list	Копия полного списка сообщений (формат OpenAI: [{"role": "user", "content": "..."}])
is_first_turn	bool	True, если это первый цикл новой сессии, False для последующих циклов
model	str	Идентификатор модели (например, "anthropic/claude-sonnet-4.6")
platform	str	Где выполняется сессия: "cli", "telegram", "discord" и т.д.

Срабатывает: В run_agent.py, внутри run_conversation(), после сжатия контекста, но до основного цикла while. Срабатывает один раз на вызов run_conversation() (т.е. один раз на цикл пользователя), а не один раз на вызов API в цикле инструментов.

Возвращаемое значение: Если колбэк возвращает словарь с ключом "context" или простую непустую строку, текст добавляется к текущему сообщению пользователя. Верните None для отсутствия внедрения.

# Внедрить контекст
return {"context": "Воспоминания:\n- Пользователь любит Python\n- Работает над vibeos-agent"}

# Простая строка (эквивалент)
return "Воспоминания:\n- Пользователь любит Python"

# Без внедрения
return None

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

Весь внедренный контекст эфемерен — добавляется только во время вызова API. Исходное сообщение пользователя в истории разговора никогда не изменяется, и ничего не сохраняется в базу данных сессии.

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

Варианты использования: Воспоминания, внедрение контекста RAG, ограничения, аналитика по циклам.

Пример — воспоминания:

import httpx

MEMORY_API = "https://your-memory-api.example.com"

def recall(session_id, user_message, is_first_turn, **kwargs):
    try:
        resp = httpx.post(f"{MEMORY_API}/recall", json={
            "session_id": session_id,
            "query": user_message,
        }, timeout=3)
        memories = resp.json().get("results", [])
        if not memories:
            return None
        text = "Вспомненный контекст:\n" + "\n".join(f"- {m['text']}" for m in memories)
        return {"context": text}
    except Exception:
        return None

def register(ctx):
    ctx.register_hook("pre_llm_call", recall)

Пример — ограничения:

POLICY = "Никогда не выполняй команды, удаляющие файлы, без явного подтверждения пользователя."

def guardrails(**kwargs):
    return {"context": POLICY}

def register(ctx):
    ctx.register_hook("pre_llm_call", guardrails)

---

post_llm_call

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

Сигнатура колбэка:

def my_callback(session_id: str, user_message: str, assistant_response: str,
                conversation_history: list, model: str, platform: str, **kwargs):

Параметр	Тип	Описание
session_id	str	Уникальный идентификатор текущей сессии
user_message	str	Исходное сообщение пользователя для этого цикла
assistant_response	str	Финальный текстовый ответ агента для этого цикла
conversation_history	list	Копия полного списка сообщений после завершения цикла
model	str	Идентификатор модели
platform	str	Где выполняется сессия

Срабатывает: В run_agent.py, внутри run_conversation(), после выхода из цикла инструментов с финальным ответом. Защищено условием if final_response and not interrupted — поэтому не срабатывает, когда пользователь прерывает цикл или агент достигает лимита итераций без ответа.

Возвращаемое значение: Игнорируется.

Варианты использования: Синхронизация данных разговора с внешней системой памяти, вычисление метрик качества ответов, логирование