Ловим ошибку в момент передачи задачи между агентами

Почему финальный лог появляется слишком поздно

Handoff — передача задачи, контекста и ответственности от одного агента другому. На этой границе состояние обычно преобразуется: сообщения сокращаются, поля переименовываются, результаты инструментов сериализуются, а внутренние объекты превращаются в транспортный формат.

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

Обычного журнала «агент A завершён, агент B запущен» недостаточно. Для локализации нужны три вещи:

• канонический снимок состояния непосредственно перед экспортом;
• снимок сразу после импорта, до первого действия принимающего агента;
• автоматическая проверка обязательных полей и допустимых преобразований.

Сначала задаём контракт границы

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

Ниже приведён пример контракта, а не универсальный стандарт:

{
  "schema_version": 1,
  "handoff_id": "h-0042",
  "task": {
    "id": "task-17",
    "goal": "Подготовить сводку",
    "constraints": [
      "Не отправлять данные во внешние системы"
    ]
  },
  "artifacts": [
    {
      "id": "notes",
      "revision": 3,
      "sha256": "..."
    }
  ],
  "decisions": [
    {
      "id": "scope",
      "value": "Только локальные материалы"
    }
  ]
}

Для границы полезно разделить поля на категории:

Идентичные

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

Преобразуемые

Например, история сообщений может быть заменена структурированной сводкой, если это явно разрешено контрактом.

Локальные

Кеши, метрики процесса и внутренние указатели не передаются и не участвуют в сравнении.

Секретные

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

Воспроизводимый минимальный стенд

Следующий пример использует только стандартную библиотеку Python. Он намеренно моделирует ошибку: экспортёр забывает перенести ограничения задачи. Команды создают лишь локальные файлы в текущем каталоге и не выполняют сетевых запросов.

1. Опишите снимок и канонизацию

from __future__ import annotations

import copy
import hashlib
import json
from pathlib import Path
from typing import Any

SNAPSHOT_DIR = Path("handoff-snapshots")

VOLATILE_KEYS = {
    "created_at",
    "local_cache",
    "runtime_metrics",
}

REQUIRED_PATHS = (
    ("schema_version",),
    ("handoff_id",),
    ("task", "id"),
    ("task", "goal"),
    ("task", "constraints"),
    ("artifacts",),
    ("decisions",),
)


def canonical(value: Any) -> Any:
    if isinstance(value, dict):
        return {
            key: canonical(item)
            for key, item in sorted(value.items())
            if key not in VOLATILE_KEYS
        }
    if isinstance(value, list):
        return [canonical(item) for item in value]
    return value


def encoded(value: Any) -> bytes:
    return json.dumps(
        canonical(value),
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
    ).encode("utf-8")


def digest(value: Any) -> str:
    return hashlib.sha256(encoded(value)).hexdigest()


def write_snapshot(stage: str, state: dict[str, Any]) -> Path:
    SNAPSHOT_DIR.mkdir(mode=0o700, exist_ok=True)
    path = SNAPSHOT_DIR / f"{state['handoff_id']}.{stage}.json"
    path.write_text(
        json.dumps(canonical(state), ensure_ascii=False, indent=2),
        encoding="utf-8",
    )
    return path

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

2. Добавьте проверку структуры и расхождений

class HandoffDivergence(RuntimeError):
    pass


def read_path(state: dict[str, Any], path: tuple[str, ...]) -> Any:
    current: Any = state
    for part in path:
        if not isinstance(current, dict) or part not in current:
            joined = ".".join(path)
            raise HandoffDivergence(f"Отсутствует обязательное поле: {joined}")
        current = current[part]
    return current


def compare_handoff(
    before: dict[str, Any],
    after: dict[str, Any],
) -> None:
    for path in REQUIRED_PATHS:
        left = read_path(before, path)
        right = read_path(after, path)
        if canonical(left) != canonical(right):
            joined = ".".join(path)
            raise HandoffDivergence(
                f"Расхождение в {joined}: "
                f"before={left!r}, after={right!r}"
            )

    if digest(before) != digest(after):
        raise HandoffDivergence(
            "Состояние различается вне проверенных путей: "
            f"before_sha256={digest(before)}, "
            f"after_sha256={digest(after)}"
        )

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

3. Оберните саму передачу

def instrumented_handoff(
    source_state: dict[str, Any],
    export_state,
    import_state,
) -> dict[str, Any]:
    before = canonical(copy.deepcopy(source_state))
    write_snapshot("before", before)

    payload = export_state(copy.deepcopy(source_state))
    after = canonical(import_state(copy.deepcopy(payload)))
    write_snapshot("after", after)

    compare_handoff(before, after)
    return after

Критическая деталь: compare_handoff вызывается до запуска принимающего агента. При расхождении функция возвращать управление не должна. Так повреждённое состояние не успеет породить новые действия.

4. Воспроизведите потерю поля

def broken_export(state: dict[str, Any]) -> dict[str, Any]:
    return {
        "schema_version": state["schema_version"],
        "handoff_id": state["handoff_id"],
        "task": {
            "id": state["task"]["id"],
            "goal": state["task"]["goal"],
            # Ошибка примера: constraints не перенесены.
        },
        "artifacts": state["artifacts"],
        "decisions": state["decisions"],
    }


def import_payload(payload: dict[str, Any]) -> dict[str, Any]:
    imported = copy.deepcopy(payload)
    imported["task"].setdefault("constraints", [])
    return imported


state = {
    "schema_version": 1,
    "handoff_id": "h-0042",
    "task": {
        "id": "task-17",
        "goal": "Подготовить сводку",
        "constraints": [
            "Не отправлять данные во внешние системы"
        ],
    },
    "artifacts": [],
    "decisions": [],
}

instrumented_handoff(
    state,
    export_state=broken_export,
    import_state=import_payload,
)

Сохраните объединённый пример как handoff_check.py и выполните:

python3 handoff_check.py

Ожидаемый результат именно для этого примера — исключение до запуска второго агента:

HandoffDivergence: Расхождение в task.constraints:
before=['Не отправлять данные во внешние системы'], after=[]

Исправляем экспорт и проверяем результат

Исправленный экспортёр переносит весь контракт задачи:

def correct_export(state: dict[str, Any]) -> dict[str, Any]:
    return {
        "schema_version": state["schema_version"],
        "handoff_id": state["handoff_id"],
        "task": copy.deepcopy(state["task"]),
        "artifacts": copy.deepcopy(state["artifacts"]),
        "decisions": copy.deepcopy(state["decisions"]),
    }

Замените broken_export на correct_export в вызове и снова запустите файл. Успешное завершение процесса без исключения означает только то, что два снимка совпали по текущему контракту. Оно не доказывает правильность цели или качество работы агентов.

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

python3 -m json.tool handoff-snapshots/h-0042.before.json
python3 -m json.tool handoff-snapshots/h-0042.after.json
sha256sum handoff-snapshots/h-0042.before.json \
  handoff-snapshots/h-0042.after.json

После исправления ожидаются два одинаковых SHA-256 для канонических файлов. В рабочей системе дополнительно проверьте:

• при расхождении принимающий агент не делает ни одного вызова инструмента;
• в ошибке присутствуют handoff_id, стадия и путь изменённого поля;
• снимки связываются с одним запуском, но не содержат секретов;
• повтор одной передачи создаёт предсказуемый результат или отдельную попытку с явным номером;
• изменение версии схемы обрабатывается миграцией, а не молчаливым заполнением полей.

Перенос в рабочий контур

В реальном оркестраторе оформите границу как отдельную операцию со стадиями capture_before, serialize, deserialize, validate и только затем start_receiver. Каждая стадия должна использовать один handoff_id.

handoff:
  snapshot:
    enabled: true
    directory: "./handoff-snapshots"
    redact:
      - "credentials"
      - "headers.authorization"
      - "tool_inputs.raw_private_data"
  validation:
    fail_closed: true
    schema_version: 1
    exact_paths:
      - "handoff_id"
      - "task.id"
      - "task.goal"
      - "task.constraints"
      - "artifacts"
      - "decisions"
    ignore_paths:
      - "created_at"
      - "runtime_metrics"
      - "local_cache"

Это пример конфигурации, а не синтаксис конкретного фреймворка. Значение fail_closed: true выражает правило: при невозможности проверить состояние передача блокируется. Для некритичных процессов допустим карантин с ручным разбором, но продолжение работы с непроверенным состоянием должно быть осознанной политикой, а не обработчиком исключения по умолчанию.

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

Типовые ошибки

Снимок создаётся слишком рано

Если между снимком и сериализацией состояние ещё изменяется, проверка сравнивает не настоящую границу. Снимайте before после завершения всех действий отправителя и непосредственно перед экспортом.

Снимок после импорта создаётся слишком поздно

Первый шаг принимающего агента может дополнить историю или нормализовать задачу. Тогда невозможно отличить дефект передачи от допустимого изменения. Снимок after нужен до планирования и вызовов инструментов.

Сравниваются только хеши

Хеш быстро подтверждает различие, но не показывает его причину. Храните структурный diff или хотя бы путь первого несовпавшего обязательного поля.

Списки бездумно сортируются

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

Отсутствующее значение приравнивается к пустому

constraints: [] и отсутствие constraints могут означать разные вещи: «ограничений нет» и «ограничения потеряны». Не применяйте setdefault до проверки обязательности поля.

Ошибка только записывается в лог

Если после сообщения об ошибке оркестратор всё равно запускает следующего агента, граница остаётся незащищённой. Проверка должна управлять переходом состояния, а не быть декоративной телеметрией.

Ограничения метода

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

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

Для больших артефактов снимок должен содержать идентификатор версии, размер и криптографический отпечаток, а не копию содержимого. Это уменьшает объём журнала, но требует гарантировать неизменяемость хранилища артефактов.

Наконец, снимки повышают наблюдаемость ценой дискового пространства и риска накопления чувствительных данных. Редактирование, контроль доступа и срок хранения являются частью реализации, а не последующей оптимизацией.

Критерий готовности

Инструментированную границу можно считать рабочей, когда намеренная потеря каждого обязательного поля воспроизводимо останавливает передачу до первого действия следующего агента, а диагностическое событие однозначно указывает запуск, handoff и изменённый путь.

Дальше имеет смысл распространить тот же контракт на остальные переходы оркестратора и свериться с материалами в разделе практических руководств. Термины, используемые при описании агентов, состояния и инструментов, собраны в глоссарии Agent Lab Journal.