Миграция состояния AI-агента без потери незавершённых задач

Почему обычного обновления схемы недостаточно

Рассмотрим чекпоинт — сохранённый снимок состояния, из которого агент продолжает выполнение. Новая версия приложения может прочитать его JSON, но всё равно неправильно интерпретировать данные.

Например, старая версия хранила вызов инструмента как {"tool":"send","args":{...}}, а новая ожидает tool_call.name, tool_call.input и статус выполнения. Если просто переименовать поля, останется неизвестным главное: был ли вызов уже отправлен во внешнюю систему. Повторный запуск способен продублировать необратимое действие.

Поэтому миграция должна сохранять три разных аспекта:

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

1. Определите контракт состояния

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

Ниже приведён пример целевой оболочки состояния. Это иллюстрация контракта, а не готовый формат конкретного фреймворка.

{
  "schema_version": 3,
  "agent_version": "2.4.0",
  "checkpoint_id": "cp_example_001",
  "revision": 17,
  "status": "suspended",
  "messages": [],
  "memory": {},
  "tasks": [],
  "tool_calls": [],
  "resume": {
    "next_step_id": "step_8",
    "reason": "awaiting_approval"
  },
  "migration": {
    "source_version": 2,
    "applied": ["v2_to_v3"],
    "migrated_at": "2026-01-01T00:00:00Z"
  }
}

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

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

2. Зафиксируйте инварианты до написания преобразователя

Инварианты описывают не форму JSON, а свойства, которые нельзя потерять. Для незавершённых задач полезен следующий минимальный набор:

1. Каждая задача имеет стабильный task_id.
2. Каждый шаг имеет стабильный step_id и явный статус.
3. Статусы принадлежат закрытому набору: pending, running, waiting, succeeded, failed, cancelled, unknown.
4. Для внешнего действия хранится ключ идемпотентности.
5. Результат инструмента связан с конкретным вызовом через call_id.
6. Неоднозначное состояние никогда автоматически не считается невыполненным.
7. После миграции сохраняется исходная копия, пригодная для отката.

Особенно важен статус unknown. Если процесс остановился после отправки запроса, но до записи ответа, нельзя безопасно выбирать между pending и succeeded. Такое действие следует отправить на сверку или ручное разрешение.

3. Используйте цепочку малых миграций

Преобразователь должен поддерживать переход только между соседними версиями: v1 → v2 → v3. Это упрощает проверку, повторное использование и расследование ошибок.

from copy import deepcopy

CURRENT_SCHEMA = 3

def migrate(state: dict) -> dict:
    result = deepcopy(state)
    version = result.get("schema_version", 1)

    while version < CURRENT_SCHEMA:
        fn = MIGRATIONS.get(version)
        if fn is None:
            raise ValueError(f"Нет миграции из версии {version}")
        result = fn(result)
        next_version = result.get("schema_version")
        if next_version != version + 1:
            raise ValueError("Миграция нарушила последовательность версий")
        version = next_version

    if version != CURRENT_SCHEMA:
        raise ValueError(f"Неподдерживаемая версия {version}")

    validate_state(result)
    return result

MIGRATIONS = {
    1: migrate_v1_to_v2,
    2: migrate_v2_to_v3,
}

Функция получает копию объекта и не изменяет исходную запись. Запись результата в хранилище выполняется отдельно — только после полной проверки.

4. Преобразуйте незавершённую работу явно

Предположим, версия 2 хранила плоский план и индекс следующего шага:

{
  "schema_version": 2,
  "plan": ["получить данные", "подготовить ответ", "отправить ответ"],
  "next_step": 1
}

Пример преобразования в явные задачи:

import hashlib

def stable_step_id(checkpoint_id: str, index: int, text: str) -> str:
    raw = f"{checkpoint_id}:{index}:{text}".encode("utf-8")
    return "step_" + hashlib.sha256(raw).hexdigest()[:16]

def migrate_v2_to_v3(state: dict) -> dict:
    checkpoint_id = state["checkpoint_id"]
    plan = state.pop("plan", [])
    next_step = state.pop("next_step", 0)

    steps = []
    for index, text in enumerate(plan):
        if index < next_step:
            status = "succeeded"
        elif index == next_step:
            status = "pending"
        else:
            status = "waiting"

        steps.append({
            "step_id": stable_step_id(checkpoint_id, index, text),
            "position": index,
            "instruction": text,
            "status": status
        })

    state["tasks"] = [{
        "task_id": f"task_{checkpoint_id}",
        "status": "pending" if steps else "succeeded",
        "steps": steps
    }]
    state["resume"] = {
        "next_step_id": steps[next_step]["step_id"]
        if next_step < len(steps) else None,
        "reason": "migrated_checkpoint"
    }
    state["schema_version"] = 3
    state.setdefault("migration", {}).setdefault("applied", []).append(
        "v2_to_v3"
    )
    return state

Такое правило допустимо только при известной семантике старого next_step. Если индекс означал «последний запущенный шаг», а не «следующий шаг», алгоритм должен быть другим. Не угадывайте смысл по названию поля: зафиксируйте его по коду старой версии и имеющимся внутренним спецификациям.

5. Отделите миграцию данных от совместимости инструментов

Старое имя инструмента нельзя безусловно заменить новым. Сначала классифицируйте изменение:

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

TOOL_ALIASES = {
    "knowledge.lookup": {
        "new_name": "search_internal",
        "safe_to_resume": True
    },
    "message.send": {
        "new_name": "deliver_message",
        "safe_to_resume": False
    }
}

def migrate_tool_call(call: dict) -> dict:
    mapping = TOOL_ALIASES.get(call["name"])
    if mapping is None:
        return {**call, "status": "blocked",
                "block_reason": "tool_not_available"}

    migrated = {**call, "name": mapping["new_name"]}

    if not mapping["safe_to_resume"] and call["status"] == "running":
        migrated["status"] = "unknown"
        migrated["requires_reconciliation"] = True

    return migrated

Для операции с внешним эффектом сохраняйте исходный call_id и ключ идемпотентности. Новый рантайм не должен генерировать их заново при возобновлении.

6. Не переписывайте историю сообщений без необходимости

Сообщения часто участвуют в аудите и воспроизведении решений. Безопаснее сохранить исходное содержимое и добавить нормализованное представление рядом:

{
  "message_id": "msg_example_14",
  "original": {
    "role": "function",
    "name": "knowledge.lookup",
    "content": "{\"items\":[]}"
  },
  "normalized": {
    "role": "tool",
    "tool_call_id": "call_example_7",
    "content": {"items": []}
  }
}

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

7. Выполните миграцию через теневую запись

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

BEGIN;

SELECT checkpoint_id, revision, payload
FROM agent_checkpoints
WHERE checkpoint_id = :checkpoint_id
FOR UPDATE;

INSERT INTO agent_checkpoint_versions (
    checkpoint_id,
    source_revision,
    schema_version,
    payload,
    migration_status
) VALUES (
    :checkpoint_id,
    :expected_revision,
    :target_schema,
    :migrated_payload,
    'validated'
);

UPDATE agent_checkpoints
SET active_revision = :new_revision,
    revision = revision + 1
WHERE checkpoint_id = :checkpoint_id
  AND revision = :expected_revision;

-- Приложение обязано проверить, что UPDATE изменил ровно одну строку.
COMMIT;

Это пример транзакционной последовательности. Названия таблиц и синтаксис блокировки необходимо адаптировать к используемой СУБД.

Безопасный порядок развёртывания:

1. Развернуть код, который читает старую и новую схемы, но пишет старую.
2. Запустить пробное преобразование без переключения активных записей.
3. Проверить отчёт и изолировать неоднозначные состояния.
4. Включить запись новой схемы для ограниченной доли новых чекпоинтов.
5. Мигрировать сохранённые состояния пакетами с ограничением нагрузки.
6. Переключить чтение на новую схему.
7. Удалять поддержку старого формата только после завершения окна отката.

8. Запускайте преобразователь в безопасном режиме

Команды ниже показывают рекомендуемый интерфейс условной утилиты agent-state-migrate. Это пример, а не указание на существующий пакет.

# Только анализ: без записи и возобновления задач
agent-state-migrate plan \
  --from-schema 2 \
  --to-schema 3 \
  --read-only \
  --report ./migration-report.json

# Ограниченный пакет с сохранением исходных версий
agent-state-migrate apply \
  --from-schema 2 \
  --to-schema 3 \
  --batch-size 100 \
  --preserve-source \
  --no-resume \
  --report ./migration-applied.json

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

Проверка результата

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

Структурная проверка

• У каждого состояния задана ожидаемая версия схемы.
• Обязательные поля присутствуют и имеют корректные типы.
• Все ссылки task_id, step_id и call_id разрешаются.
• Нет неизвестных статусов и инструментов без явной блокировки.

Проверка инвариантов

def validate_progress(before: dict, after: dict) -> None:
    old_completed = count_completed_steps(before)
    new_completed = count_completed_steps(after)

    if new_completed < old_completed:
        raise ValueError("Потеряна информация о завершённых шагах")

    if duplicate_ids(after["tasks"]):
        raise ValueError("Обнаружены повторяющиеся идентификаторы")

    for call in after.get("tool_calls", []):
        if call["status"] == "running" and not call.get("idempotency_key"):
            raise ValueError("Нельзя возобновить вызов без ключа идемпотентности")

Сверка количеств

Отчёт должен сопоставлять количество входных и выходных объектов:

прочитано состояний:      N
создано новых версий:     N
переключено указателей:   N - blocked
заблокировано:            blocked
потеряно задач:           0
потеряно сообщений:       0
неоднозначных вызовов:    unknown

Здесь буквы обозначают вычисляемые значения, а не фактические результаты запуска. Успех определяется не заранее заданным числом, а равенствами и нулевыми потерями.

Проверка возобновления

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

Безопасный откат

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

До возобновления задач откат выполняется переключением активного указателя на сохранённую исходную версию:

BEGIN;

UPDATE agent_checkpoints
SET active_revision = :source_revision,
    revision = revision + 1
WHERE checkpoint_id = :checkpoint_id
  AND active_revision = :migrated_revision
  AND revision = :expected_revision;

-- Продолжить только при изменении ровно одной строки.
COMMIT;

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

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

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

Версия определяется по наличию поля

Эвристика ломается на частично записанных и промежуточных объектах. Храните явный номер схемы и отклоняйте неподдерживаемые значения.

Все состояния running превращаются в pending

Это провоцирует повтор инструментального вызова. Используйте unknown и процедуру сверки.

Идентификаторы генерируются случайно при каждой попытке

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

Миграция одновременно преобразует и возобновляет задачу

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

Обновление выполняется на месте

Частичный сбой оставляет единственную копию непригодной. Используйте теневую версию и атомарное переключение указателя.

Старые сообщения безусловно отправляются новой модели

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

Откат проверяется только на схеме

После продолжения выполнения старый чекпоинт уже не отражает реальный мир. Нужна сверка внешних действий и журнала вызовов.

Ограничения

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

Итоговый контрольный список

1. Версия схемы хранится явно и отдельно от версии приложения.
2. Для задач, шагов и вызовов заданы стабильные идентификаторы.
3. Инварианты прогресса проверяются программно.
4. Миграции последовательны, детерминированы и не изменяют исходную запись.
5. Неоднозначные внешние вызовы получают статус unknown.
6. Преобразование, переключение состояния и возобновление разделены.
7. Запись защищена транзакцией и проверкой ревизии.
8. Исходный чекпоинт сохраняется на всё окно отката.
9. Отчёты не раскрывают содержимое памяти и аргументы инструментов.
10. Откат после возобновления включает сверку уже выполненных действий.

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