ТЕХНИЧЕСКОЕ ЗАДАНИЕ: Реализовать rollback delegation

1. Цель задачи

Разработать механизм безопасного делегирования задач между агентами в multi-agent системе, который гарантирует консистентность состояния при ошибках исполнителя. Когда агент A допускает ошибку, его частичные действия должны быть откачены (rollback), а задача передана агенту B для повторного выполнения. Необходимо реализовать протокол, обеспечивающий атомарность и idempotency операций, чтобы система оставалась в консистентном состоянии независимо от точки сбоя.

Ключевой результат Функциональный прототип (или детальный дизайн + код), в котором при имитации ошибки агента A все его изменения отменяются, и агент B успешно завершает задачу без остаточных артефактов.

2. Исходные данные

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

Если нет реального multi-agent фреймворка — симулируем:

1. Создаём двух простых агентов (классы Python), которые могут выполнять последовательность действий (например, обновить записи в SQLite, отправить HTTP-запрос).
2. Один агент (A) намеренно выбрасывает исключение на последнем шаге.
3. Пишем координатор (DelegationManager), который перехватывает ошибку, вызывает rollback-функции для агента A и передаёт задачу агенту B.

3. Технологический стек

4. Этапы выполнения

Этап 1: Проектирование протокола rollback delegation (30 минут)

Действия

1. Определить жизненный цикл задачи:
   • Статусы: PENDING, IN_PROGRESS, COMMITTED, FAILED, ROLLED_BACK, DELEGATED, COMPLETED.
2. Спроектировать интерфейсы агента:

   class AgentInterface(ABC):
       @abstractmethod
       def execute_task(self, task: dict) -> bool: ...
       @abstractmethod
       def rollback_task(self, task_id: str) -> None: ...
   

3. Описать протокол координатора:
   • Вызвать agent_A.execute_task(task).
   • При успехе — зафиксировать commit.
   • При исключении — немедленно вызвать agent_A.rollback_task().
   • После отката — вызвать agent_B.execute_task(task).
4. Гарантировать идемпотентность: каждый шаг должен быть повторяем (использовать идемпотентные операции, например UPDATE ... SET status = 'COMPLETED' WHERE status = 'IN_PROGRESS').

Ожидаемый результат этапа Документ (Markdown) или диаграмма последовательности, описывающая протокол.

Этап 2: Реализация координатора и базовых агентов (1.5 часа)

Действия

1. Создать базовые классы AgentA и AgentB, наследующие AgentInterface.
2. Каждый агент имитирует работу с БД:
   • Создаёт временные записи (например, INSERT INTO actions (agent, step) VALUES ('A', 1)).
   • rollback_task удаляет эти записи (DELETE FROM actions WHERE agent = 'A' AND task_id = ?).
3. Реализовать DelegationManager:

   class DelegationManager:
       def __init__(self, agent_a: AgentInterface, agent_b: AgentInterface):
           ...
       def run(self, task: dict) -> str:
           try:
               self.agent_a.execute_task(task)
               return "SUCCESS"
           except Exception as e:
               logger.error(f"Agent A failed: {e}")
               self.agent_a.rollback_task(task["id"])
               logger.info("Rollback completed, delegating to Agent B")
               self.agent_b.execute_task(task)
               return "DELEGATED_SUCCESS"
   

4. Добавить детальное логирование каждого шага (время, статус, стек ошибок).

Ожидаемый результат этапа Работающий скрипт, который при запуске показывает:

• Если Agent A успешен — только его действия.
• Если Agent A бросает исключение — сначала откат A, затем успешное выполнение B.

Этап 3: Обработка граничных случаев и сбоев при откате (1 час)

Действия

1. Рассмотреть сценарии:
   • Rollback сам упал (например, БД недоступна).
   • Agent B тоже упал (тогда нужно повторить или делегировать дальше — фоллбэк).
   • Частичный откат (один шаг откатился, другой нет — нужно компенсировать).
2. Реализовать стратегию повторных попыток для rollback (retry с exponential backoff).
3. Добавить проверку консистентности после каждого rollback:
   • Проверить, что все записи агента A удалены или аннулированы.
   • Использовать отдельный статус ROLLING_BACK во избежание двойного отката.
4. Написать unit-тесты на каждый граничный случай (pytest).

Ожидаемый результат этапа Расширенный координатор, обрабатывающий сбои при откате и делегировании. Тесты покрывают не менее 5 сценариев.

Этап 4: Интеграционное тестирование и валидация консистентности (1 час)

Действия

1. Создать сквозной тест, который:
   • Инициализирует БД (SQLite in-memory).
   • Запускает задачу с Agent A, который падает на 2-м шаге.
   • Проверяет, что после отката в БД нет записей от A.
   • Проверяет, что Agent B выполнил задачу, и его записи присутствуют.
2. Написать скрипт, измеряющий консистентность:

   def test_consistency_after_failure():
       state_before = snapshot_db()
       manager.run(some_task)
       state_after = snapshot_db()
       # Agent A records should be absent, Agent B records present
       assert "agent_a" not in state_after
       assert state_after["agent_b"]["task_id"] == some_task["id"]
   

3. Добавить метрику успеха: время выполнения, количество вызовов rollback, частота отказов Agent B.

Ожидаемый результат этапа Набор интеграционных тестов (pytest), подтверждающих консистентность состояния при любом сценарии сбоя.

Этап 5: Документирование и демонстрация (30 минут)

Действия

1. Написать краткое README с описанием архитектуры, инструкцией по запуску и примером вывода.
2. Создать файл rollback_delegation_demo.py, который можно запустить и увидеть пошаговый лог.
3. Зафиксировать в отчёте:
   • Количество протестированных сценариев.
   • Время выполнения в нормальном и падёжном режиме.
   • Любые ограничения (например, нет поддержки distributed transactions).

Ожидаемый результат этапа README.md и демо-скрипт, готовые к презентации.

5. Критерии приемки (Definition of Done)

• Реализован протокол rollback delegation, при котором ошибка агента A приводит к откату всех его действий.
• Агент B получает задачу в том же состоянии, что и до начала работы агента A (чистая доска).
• Состояние системы после успешного делегирования полностью консистентно (нет остаточных данных от A, все шаги B завершены).
• Координатор корректно обрабатывает сбой во время rollback (повторяет попытку, логирует, не зацикливается).
• Все операции идемпотентны: повторный вызов execute или rollback не портит состояние.
• Написаны минимум 5 unit-тестов и 2 integration-теста, покрывающих основные сценарии (успех, ошибка A, ошибка при rollback, ошибка B).
• Код проходит pylint / black без ошибок, уровни логирования настроены (DEBUG для отладки, INFO для основных событий).
• Есть README с инструкцией по запуску и примерами вывода.

6. Ожидаемый результат

Основной артефакт Папка проекта с файлами:

• agent.py — классы AgentA, AgentB.
• manager.py — DelegationManager.
• exceptions.py — кастомные исключения (RollbackError, DelegateError).
• test_rollback.py — тесты.
• demo.py — демо-скрипт.
• README.md — документация.

Содержание README

• Архитектура (диаграмма или текстовое описание).
• Шаги запуска: pip install -r requirements.txt && python demo.py.
• Пример вывода:

2025-12-31 10:00:00 | INFO | Delegating task 'T-001' to Agent A
2025-12-31 10:00:01 | INFO | Agent A: step 1 complete
2025-12-31 10:00:02 | ERROR | Agent A failed: Simulated exception on step 2
2025-12-31 10:00:02 | INFO | Rolling back Agent A actions...
2025-12-31 10:00:02 | INFO | Rollback successful
2025-12-31 10:00:02 | INFO | Delegating task 'T-001' to Agent B
2025-12-31 10:00:03 | INFO | Agent B: task completed
Final state: consistency check passed.

Дополнительные результаты

• Метрики успеха (таблица с временами отката).
• Отчёт о покрытии тестов (coverage report).

7. Возможные сложности и их решение

8. Бюджет времени (оценка)

Примечание Если вы впервые знакомитесь с паттерном Saga или компенсирующими транзакциями, заложите дополнительно 1 час на изучение.

9. Связанные вопросы из базы знаний

10. Чек-лист самопроверки

• Я проверил, что при ошибке агента A его изменения действительно откатываются (а не просто игнорируются).
• Я убедился, что после отката агент B получает задачу в том же контексте, что и A (те же параметры, не испорченные).
• Я протестировал сценарий, где rollback сам выбрасывает исключение — система не зависла и записала предупреждение.
• Я проверил, что повторный запуск того же task_id (идемпотентность) не создаёт дублирующих записей.
• Я прочитал README и убедился, что другой разработчик сможет запустить демо без дополнительных пояснений.
• Я подключил хотя бы базовое логирование с разными уровнями, чтобы видеть хронологию.