Распределённое исполнение в gigaevo-core

[GrigoryEvko]

1. Status quo

Текущая архитектура gigaevo-core опирается на допущение «один драйвер, один хост». Это допущение проявляется в трёх местах:

1. Исполнение программ. Пул подпроцессов loky.get_reusable_executor живёт на одной машине с драйвером и завершается вместе с ним.
2. Память. AmemGamMemory и CardStore используют локальный файловый кэш с записью через os.replace, без идемпотентности на стороне Memory API и без межпроцессной инвалидации.
3. Блокировка экземпляра. RedisInstanceLock выполняет взаимное исключение и явно запрещает двум драйверам использовать общий префикс ключей в Redis.

При этом в кодовой базе уже есть подсистемы, спроектированные с учётом распределённости: пул серверов LLM (EndpointPool с координацией через атомарный Lua-скрипт в Redis) и шина миграции островов (RedisStreamTransport с XADD/XREAD и захватом сообщений через SETNX). Эти подсистемы доказывают, что Redis-координация при корректной декомпозиции операций обеспечивает достаточную согласованность, и одновременно обнажают тот факт, что аналогичный подход на остальные слои пока не распространён.

Архитектурный фундамент для распределённой топологии поставляется в PR #20 «typed Redis coordination plane». Этот PR вводит единственный типизированный асинхронный координатор gigaevo.dataplane.DataPlane, через который проходят все взаимодействия с Redis. В его составе: восемь атомарных Lua-скриптов, линейные токены владения Token[Tag], типизированный возврат Result[T, E], обёртка потери аренды CrashEvent + wrap_lease и гибридный логический таймштамп HlcTimestamp. Этапы 3, 4 и 5 настоящего RFC прямо опираются на этот субстрат.

Дополнительный фактор - целевая нагрузка смежного проекта kernel-evo: оптимизация CUDA-ядер через эволюционный поиск. Эта нагрузка предъявляет требования, выходящие за рамки текущей поддержки gigaevo-core:

• Эксклюзивный доступ к GPU на время измерения: шум измерений недопустим.
• Привязка к типу GPU: бенчмарк должен выполняться на конкретной модели (H100, A100, L40S и т. д.).
• От сотен до тысяч вариантов ядра в каждом поколении.
• Гетерогенный парк: часть рабочих процессов выполняет только компиляцию (CPU), часть проводит измерения (GPU с эксклюзивным захватом).

Текущая локально-замкнутая модель не позволяет эффективно использовать парк из нескольких GPU-узлов: ни один штатный способ запуска драйвера не координирует работу за пределами одного хоста.

---

2. Текущая архитектура

2.1. Топология «один хост по умолчанию»

┌───────────────────────────────────────────────────────────────┐
│  Один хост                                                     
│                                                                
│  ┌───────────────────┐                                         
│  │     run.py        │                                         
│  │  ┌─────────────┐  │                                         
│  │  │EvolutionEng │  │  DataPlane (PR #20)                     
│  │  │DagRunner    │──┼─► dp.* ──EVALSHA─►┌──────────┐          
│  │  │Stages       │  │                   │  Redis   │          
│  │  └─────────────┘  │                   │ db /0    │ state   
│  │        │ loky      │                  │ db /15   │ LLM pool 
│  │        │ pipe + cloudpickle           └──────────┘         
│  │        ▼          │                                         
│  │  ┌─────────────┐  │                                         
│  │  │loky workers │  │                                         
│  │  │  (subproc)  │  │                                         
│  │  │   GPU       │  │ локальная CUDA                          
│  │  └─────────────┘  │                                         
│  └────────┬──────────┘                                         
│           │ requests (loopback)                                
│           ▼                                                    
│  ┌──────────────────┐                                          
│  │ Memory API :8000 │                                         
│  └──────────────────┘                                          
└──────────────┬─────────────────────────────────────────────────┘
               │ aiohttp / HTTPS
               ▼
       ┌────────────────────┐
       │  Внешний LLM       │  (OpenRouter или совместимый API)
       └────────────────────┘

2.2. Жёсткие ограничения для распределения

Ограничение	Где зафиксировано	К чему приводит при распределении	Статус после PR #20
Обработчики - локальные подпроцессы	gigaevo/programs/stages/python_executors/wrapper.py, asyncio.create_subprocess_exec	Невозможно запустить обработчики на отдельной машине	Без изменений; адресуется этапом 3
Переменные окружения передаются через наследование процесса	os.environ.copy() плюс белый список ключей	Удалённый обработчик не может получить конфигурацию из файловой системы родительского процесса	Без изменений; адресуется этапом 3
Идемпотентность записи в Memory API отсутствует	concept_api._request, POST /v1/memory-cards без ключа идемпотентности	Одновременно работающие драйверы создают дубликаты сущностей	IdempotencyToken + compute_content_hash_hex подготовлены в PR #20; интеграция в Memory API остаётся этапом 4
Один драйвер на префикс ключей	RedisInstanceLock.acquire в gigaevo/database/redis/locking.py	Несколько драйверов вынуждены работать с разными префиксами и не могут вести общий архив	Блокировка экземпляра смигрирована через LuaRegistry.evalsha + Token[InstanceLockTag] + wrap_lease в PR #20; расширение на лидерскую аренду роли производится фабричной функцией поверх существующих Lua-скриптов
CardStore - локальный файл, при конфликте сохраняется последняя запись	gigaevo/memory/shared_memory/card_store.py, os.replace	На общем диске несколько драйверов перезаписывают индекс друг друга	Относится к плоскости данных; вне круга ответственности DataPlane
MultiModelRouter._verify_models опрашивает серверы при инициализации	gigaevo/llm/models.py:_verify_models	N драйверов одновременно создают нагрузку на инференс-сервер при холодном старте	Без изменений

2.3. Что уже распределено

Подсистема	Координация	Готовность к множеству драйверов
Пул серверов LLM	Атомарный Lua в Redis; ключи llm_pool:{pool}:inflight, :cooldown:{hash}	Полная. Единственная проблема - счётчик inflight не уменьшается при аварийном завершении драйвера
Шина миграции островов	XADD/XREAD; SETNX-захват сообщения с TTL 120 с	Полная
Машина состояний программ	После PR #20: dp.transition_program_state + transition_state.lua с серверной валидацией FSM-таблицы PROGRAM_STATE_TRANSITIONS; идемпотентность через idempotency_token = compute_content_hash_hex(...); поток событий {prefix}:events:status	Согласована внутри одного префикса; межпрефиксная согласованность отнесена к фазе 2 substrate PR #20
Архив элит (MAP-Elites)	После PR #20: dp.try_replace_elite + archive_swap.lua с tiebreak-битом, обратным/прямым индексами и инвариантом биекции	Полная для редуцируемых селекторов (SumArchiveSelector); ParetoFrontSelector сохраняет WATCH-цикл с верхней границей 50 попыток
Реестр многоруких бандитов	После PR #20: dp.crdt_inc + counter_inc.lua с ActorIdentity(run_id, worker_id) для актор-партиционирования; per-arm trial counts и reward sums зеркалируются в Redis G-счётчики	Полная; кросс-процессное разделение состояния
Распределённая блокировка	После PR #20: dp.acquire_instance_lock / renew_instance_lock / release_instance_lock; типизированный Token[InstanceLockTag]; потеря аренды через wrap_lease + CrashEvent	Полная для блокировки префикса; расширение на лидерскую аренду роли и эксклюзивный захват GPU - фабричной функцией поверх существующих Lua-скриптов
Словарь подсказок (prompts/fetcher, coevolution/stats)	После PR #20: dp.crdt_inc для trials/successes/metrics; dp.bounded_list_push для fitness-окна; dp.set_add для каталога метрик; dp.raw_hash_get для опроса engine:total_generations	Полная; межзапусковая агрегация по списку dataplanes: list[DataPlane]

---

3. Предлагаемая архитектура

Архитектура опирается на четыре принципа:

1. Redis как единственная плоскость управления. Отдельные серверы очередей, диспетчеров и координаторов не вводятся. Через Redis проходят все формы координации: очереди задач, аренды, пульсы, захваты. Это согласуется с тем, как уже устроены шина миграции и пул серверов LLM.
2. Разделение плоскости управления и плоскости данных. Через Redis передаются метаданные и ссылки; код, артефакты сборки и результаты измерений хранятся в хранилище артефактов. Локальная файловая система, NFS и S3 выступают взаимозаменяемыми реализациями одной абстракции.
3. Обработчики как независимые процессы. gigaevo worker оформляется отдельной командой CLI. Жизненный цикл обработчика отвязан от жизненного цикла драйвера. Драйвер становится производителем заявок и потребителем результатов; управление вычислительными ресурсами выносится из драйвера.
4. Согласование задач и обработчиков через теги. Централизованный планировщик отсутствует; маршрутизация выполняется самими обработчиками через подписку на нужные очереди.

Все четыре принципа реализуются через типизированный субстрат gigaevo.dataplane.DataPlane, поставляемый PR #20. Прямые вызовы redis.asyncio блокированы линтером TID251; любая операция координации проходит через DataPlane.evalsha с заранее зарегистрированным Lua-скриптом; любой возврат имеет форму Result[T, E]; любая аренда оформляется линейным токеном Token[Tag] с дисциплиной однократного потребления.

3.1. Целевая топология (распределённый режим)

                  ┌───────────────────────────────────────┐
                  │           Координационный Redis        │
                  │                                        │
                  │  Потоки:                               │
                  │    tasks:cpu                           │
                  │    tasks:gpu-h100                      │
                  │    tasks:gpu-a100                      │
                  │    tasks:gpu-any                       │
                  │    results:{driver_id}                 │
                  │    evo:migrate    ← уже существует     │
                  │                                        │
                  │  Ключи:                                │
                  │    claim:{job_id}     SETNX TTL        │
                  │    heartbeat:{wid}    SETEX            │
                  │    gpu-lock:{node}:{idx}  SETNX TTL    │
                  │    lease:{role}      SETNX renewable   │
                  │    idem:{op_id}      опционально       │
                  └──┬───────────────┬───────────────┬─────┘
                     │               │               │
            DataPlane (PR #20) ──EVALSHA── через redis-py / TCP
                     │               │               │
        ┌────────────▼─────┐ ┌───────▼──────┐ ┌──────▼────────┐
        │ Драйвер 1        │ │ Драйвер 2     │ │  Драйвер N    │
        │ EvolutionEngine  │ │ EvolutionEng. │ │ EvolutionEng. │
        │ DagRunner        │ │ DagRunner     │ │ DagRunner     │
        │                  │ │               │ │               │
        │ XADD → tasks     │ │               │ │               │
        │ XREAD ← results: │ │               │ │               │
        │   {driver_1}     │ │               │ │               │
        └────────┬─────────┘ └───────────────┘ └───────────────┘
                 │
                 │ submission содержит ссылки на код в хранилище
                 ▼
        ┌────────────────────────────────────────────────────┐
        │     Хранилище артефактов (плоскость данных)         │
        │                                                     │
        │   file:///mnt/shared/gigaevo  (NFS / общий диск)    │
        │   s3://bucket/gigaevo         (облако)              │
        │   file:///tmp/gigaevo-{uid}   (один хост)           │
        │                                                     │
        │   Код, бинарники, результаты, отчёты профилирования │
        └─────────────┬───────────────────────────────────────┘
                        │
                        │ обработчики читают и пишут сюда
                        │
   ┌──────────────────┴───────────────────────────────────┐
   │  Парк обработчиков                                   │
   │                                                      │
   │  ┌────────────────────────┐  ┌─────────────────────┐ │
   │  │ gigaevo worker         │  │ gigaevo worker      │ │
   │  │ --tags=cpu,compile     │  │ --tags=gpu:h100,    │ │
   │  │                        │  │   bench,exclusive   │ │
   │  │ dp.consume(tasks:cpu)  │  │ dp.consume(tasks:   │ │
   │  │ dp.acquire_instance_   │  │   gpu-h100, gpu-any)│ │
   │  │   lock(claim:{id})     │  │ dp.acquire_instance │ │
   │  │ dp.renew_instance_lock │  │   _lock(gpu-lock:0) │ │
   │  └────────────────────────┘  └─────────────────────┘ │
   │                                                      │
   │  ┌────────────────────────┐  ┌─────────────────────┐ │
   │  │ gigaevo worker         │  │ gigaevo worker      │ │
   │  │ --tags=gpu:a100,bench  │  │ --tags=gpu:l40s,    │ │
   │  │   shareable            │  │   bench,exclusive   │ │
   │  │ dp.consume(tasks:      │  │ dp.consume(tasks:   │ │
   │  │   gpu-a100, gpu-any)   │  │   gpu-l40s, gpu-any)│ │
   │  │ Семафор cap=4          │  │ dp.acquire_instance │ │
   │  │                        │  │   _lock(gpu-lock:0) │ │
   │  └────────────────────────┘  └─────────────────────┘ │
   └──────────────────────────────────────────────────────┘
              │
              │ исходящий LLM-трафик (aiohttp после PR #17)
              ▼
      ┌──────────────────────┐
      │  LLM-кластер         │   через EndpointPool
      │  (несколько vLLM /   │   (уже работает)
      │   sglang серверов)   │
      └──────────────────────┘

3.2. Встроенный режим (для разработки и тестов)

Распределённая топология - обобщение; локальная топология сохраняется как частный случай. В режиме разработки достаточно одного процесса gigaevo, запущенного с флагом --embedded:

┌───────────────────────────────────────────────────────────────┐
│  Один процесс (gigaevo run --embedded)                        
│                                                               
│  ┌──────────────────┐    ┌─────────────────────────────────┐  
│  │ EvolutionEngine  │    │ Встроенный пул обработчиков     │  
│  │ DagRunner        │───►│(та же реализация ExecutorBackend│  
│  │                  │    │  что у gigaevo worker, но в том │  
│  │ XADD ──────────► │    │  же процессе и без сетевого     │  
│  │       in-memory  │◄───│  Redis: реализация на asyncio   │  
│  │       queue      │    │  Queue вместо XSTREAM)          │  
│  └──────────────────┘    └─────────────────────────────────┘  
│           │                                                   
│           ▼                                                   
│  ┌────────────────────────┐                                   
│  │ Хранилище артефактов:  │                                   
│  │ file:///tmp/...        │ (как сейчас спил-файлы loky)      
│  └────────────────────────┘                                   
└───────────────────────────────────────────────────────────────┘

Встроенный режим использует общий контракт ExecutorBackend, реализованный поверх asyncio.Queue вместо Redis Streams. Внешний Redis для разработки и тестирования при этом не требуется. Типизированный контракт обмена (Result[T, E] + Token[Tag] + Versioned[T] из PR #20) обеспечивает инвариантность семантики относительно физического размещения участников, и переключение между распределённой и встроенной топологиями выполняется фабрикой без правок прикладного кода.

3.3. Контракт обработчика

Каждый обработчик реализует короткий рабочий цикл:

async def worker_loop(config: WorkerConfig) -> None:
    await register_self(config)                         # объявляет теги и мощность

    while not config.should_exit:
        msg = await xread_filtered(
            streams=tag_to_streams(config.tags),
            block_seconds=10,
        )
        if msg is None:
            continue                                    # пустая итерация

        claim_holder = await try_claim(
            msg.job_id, worker_id=config.worker_id, ttl=msg.deadline_s,
        )
        if claim_holder != config.worker_id:
            continue                                    # захватил другой обработчик

        gpu_lock = None
        if msg.requires_gpu_exclusive:
            gpu_lock = await acquire_gpu_lock(
                node_id=config.node_id,
                gpu_idx=config.gpu_indices[0],
                ttl=msg.deadline_s,
            )
            if gpu_lock is None:
                await release_claim(msg.job_id)
                continue

        heartbeat_task = asyncio.create_task(
            renew_heartbeat(config.worker_id, ttl=30, interval=10),
        )
        renew_lock_task = asyncio.create_task(
            renew_lock(claim_key=f"claim:{msg.job_id}", ttl=msg.deadline_s, interval=msg.deadline_s // 3),
        )
        try:
            code = await artifact_store.read(msg.code_uri)
            result = await execute(code, msg)
            result_uri = await artifact_store.write(
                key=f"result/{msg.job_id}/{msg.attempt}",
                value=result,
            )
            await xadd(
                stream=f"results:{msg.driver_id}",
                fields={
                    "job_id": msg.job_id,
                    "attempt": msg.attempt,
                    "result_uri": result_uri,
                    "worker_id": config.worker_id,
                    "wall_time_s": elapsed,
                },
            )
        except Exception as exc:
            await xadd(
                stream=f"results:{msg.driver_id}",
                fields={
                    "job_id": msg.job_id,
                    "attempt": msg.attempt,
                    "error": serialize_error(exc),
                },
            )
        finally:
            heartbeat_task.cancel()
            renew_lock_task.cancel()
            if gpu_lock is not None:
                await release_gpu_lock(gpu_lock)
            await xack(msg.stream, msg.id)

• Захват продлевается из фоновой задачи. Это снимает противоречие между «короткий TTL, чтобы быстро освобождать после смерти обработчика» и «длинный TTL, чтобы пережить долгую работу». TTL остаётся коротким; обработчик его продлевает, пока работа идёт. После PR #20 реализуется как dp.renew_instance_lock(lease, ttl_s=...); потеря аренды (отказ token-CAS, исключение транспорта) видна типизированно через wrap_lease + CrashEvent, без out-of-band-флагов.
• Блокировка GPU действует на время работы. Освобождается в блоке finally. Если процесс убит без блока finally - TTL истечёт и блокировка снимется автоматически. После PR #20 реализуется фабричной функцией поверх dp.acquire_instance_lock с ключом gpu-lock:{node}:{idx}, без модификации Lua-субстрата.
• Запись результата строго идемпотентна. Ключ хранилища result/{job_id}/{attempt} уникален; повторная запись с тем же значением attempt либо переписывает байт-в-байт, либо отвергается на стороне хранилища. Координационная сторона идемпотентности обеспечивается IdempotencyToken + compute_content_hash_hex из PR #20.

3.4. Примитивы координации

Используются четыре примитива. Все четыре поставляются в рамках PR #20 как типизированные методы DataPlane поверх соответствующих атомарных Lua-скриптов.

Примитив 1: атомарный захват с TTL. Реализован в scripts/instance_lock_acquire.lua + dp.acquire_instance_lock(key, *, ttl_s, deadline_monotonic) -> Result[InstanceLease, LockHeld].

-- KEYS[1] = "claim:{job_id}", ARGV[1] = worker_id, ARGV[2] = ttl_ms
-- (PR #20 scripts/instance_lock_acquire.lua: SET key token NX PX ttl_ms)
if redis.call('SET', KEYS[1], ARGV[1], 'NX', 'PX', ARGV[2]) then
    return {'ok'}
else
    return {'held', redis.call('GET', KEYS[1])}
end

Возвращает либо линейный токен Token[InstanceLockTag] (через Ok(InstanceLease(...))), либо типизированную ошибку Err(LockHeld(key=..., holder=...)) с идентификатором текущего владельца. Поле holder помечено REDACT_META_KEY для защиты от утечки чувствительных значений через str(err). Используется для назначения задачи (claim:{job_id}), для блокировки GPU (gpu-lock:{node}:{idx}), для эксклюзивной аренды роли (lease:{role}).

Примитив 2: продлеваемый пульс. Реализован в scripts/instance_lock_renew.lua (token-CAS обновление TTL) + dp.renew_instance_lock(lease, ttl_s) -> Result[InstanceLease, LockLost] + dp.wrap_lease(lease) -> CrashWatchedHandle.

Обработчик записывает dp.renew_instance_lock(lease, ttl_s=30) каждые 10 секунд. Соотношение TTL и интервала обновления - 3 : 1 - даёт двойной запас на случай задержек GC, сетевых пауз или пропущенной итерации. Драйвер (или любой наблюдатель) перебирает захваты и проверяет наличие пульса у владельца. Если пульс отсутствует, захват снимается, и задача возвращается в очередь через повторное XADD от стороны, обнаружившей пропажу. Это гарантирует освобождение работы при сбое обработчика в течение не более 30 секунд. На стороне обработчика потеря аренды видна типизированно через CrashEvent без out-of-band-флагов.

Примитив 3: устойчивая очередь на Redis Streams. Словарь подготовлен в PR #20 (substrate фазы 2): NewType-идентификаторы StreamName, ConsumerGroup, ConsumerName, EventId, CausationId, CorrelationId плюс subspace streams/ под Lua-скрипты. Полная реализация dp.emit / dp.consume поверх XREADGROUP / XACK отнесена к фазе 2 общего плана программы изоляции.

Контракт: XADD при отправке, XREAD при чтении, XACK после успешной обработки. Поток имеет ограничение длины через MAXLEN ~. Каждая очередь имеет одну группу потребителей; в неё подписаны все обработчики с подходящими тегами. Это та же схема, что уже используется в RedisStreamTransport. XREAD с блокированием на ~10 секунд даёт реакцию обработчика на новое задание в пределах сетевой задержки, без активного опроса.

Примитив 4: идемпотентная запись. Реализован в PR #20 для FSM-переходов программ: NewType IdempotencyToken + поле idempotency_token = compute_content_hash_hex({"pid": pid, "from": from, "to": to, "patch": patch_dict}, schema_version=1) + ленивая запись через HSETNX в {prefix}:program:{pid}:idem с TTL 300 секунд. Lua-скрипт transition_state.lua проверяет наличие токена через HEXISTS до выполнения мутации; повторный вызов с тем же токеном возвращает кешированный результат предыдущего успешного перехода без побочного эффекта.

В Memory API клиент добавляет заголовок Idempotency-Key: <uuid> (фаза 4 настоящего RFC). Сервер сверяет ключ с короткоживущим журналом (например, Redis-хеш с TTL 24 часа): при совпадении возвращает ранее созданную сущность без повторной обработки, при отсутствии - выполняет операцию и фиксирует ключ. В хранилище артефактов идемпотентность обеспечивается уникальностью пути result/{job_id}/{attempt}: повторная запись с теми же координатами безопасна и не нарушает целостности.

3.5. Модель отказов

Сценарий отказа	Симптом	Восстановление
Аварийное завершение обработчика	Пульс не продлевается	Через ≤30 с TTL пульса истекает, наблюдатель снимает захват, задача возвращается в очередь, идемпотентная запись результата через IdempotencyToken (PR #20) защищает от двойного применения
Сетевая изоляция обработчика	Пульс не доходит до Redis, обработчик жив, но изолирован	То же, что аварийное завершение. При восстановлении dp.renew_instance_lock вернёт Err(LockLost); wrap_lease поднимет CrashEvent; обработчик через match тихо отказывается от записи результата
Аварийное завершение драйвера во время отправки	Часть пакета XADD прошла, часть - нет	Незавершённые задачи отсутствуют в очереди; перезапуск драйвера отправляет недостающие
Сбой первичного Redis	Координация останавливается	Sentinel переводит реплику в первичные; восстановление продолжается; задания в полёте могут выполниться дважды, но защищены IdempotencyToken PR #20
Зависание GPU	Пульс обработчика жив, прогресс отсутствует	Сторожевой таймер на стороне обработчика: если задание не продвинулось за N × ожидаемое время, обработчик принудительно завершает работу и записывает ошибку; внешний супервизор перезапускает обработчика, при необходимости - с nvidia-smi --gpu-reset
Медленный обработчик создаёт блокировку головы очереди	Одна задача удерживает захват, тогда как поколение должно завершиться	Дедлайн поколения короче TTL захвата; при истечении дедлайна драйвер активно снимает захваты повисших задач через dp.release_instance_lock и фиксирует их как неуспех
Двойная запись в Memory API	Два драйвера POST'ят одну логическую карту	Заголовок Idempotency-Key; сервер возвращает существующий entity_id
Повреждение результата в хранилище	Обработчик записал испорченные байты	Контрольная сумма проверяется при чтении; при несовпадении задача возвращается в очередь с увеличенным значением attempt
Недоступность хранилища артефактов	Обработчик не может записать результат	Попытка завершается неуспехом; задача возвращается в очередь с увеличенным attempt; при исчерпании лимита попыток драйвер видит окончательный неуспех
Удаление программы драйвером в процессе исполнения	Обработчик тратит ресурсы впустую	Перед записью результата обработчик проверяет состояние программы в Redis; если статус DISCARDED, запись отменяется

---

4. Сравнение с текущей моделью

4.1. Сводная таблица

Аспект	Сейчас	Предлагается
Жизненный цикл обработчика	Связан с драйвером, завершается вместе с ним	Независимый процесс, переживает драйвер
Передача переменных окружения	Наследование os.environ с фильтрацией	Сериализуемый WorkerConfig, доставляется как параметры запуска или через Redis
Передача кода	Внутрипроцессная, через WorkerCall (cloudpickle)	Через хранилище артефактов, по ссылке
Передача результата	Файл сброса данных на локальной FS, mmap у родителя	Через хранилище артефактов, по ссылке
Распределение нагрузки	Loky внутри одного хоста	Redis Streams + теги + захват через dp.acquire_instance_lock (PR #20)
Дисциплина координации	Прямые redis.asyncio вызовы во многих точках	Монопольный субстрат gigaevo.dataplane.DataPlane (PR #20); прямые импорты redis блокированы линтером TID251
Атомарность операций	Ручные циклы WATCH/MULTI/EXEC без верхней границы	Атомарные Lua-скрипты в составе EVALSHA с типизированным возвратом Result[T, E] (PR #20)
Гетерогенность парка	Не поддерживается	Через теги обработчиков и подписку на очереди
Эксклюзивный GPU	Не поддерживается явно	Через dp.acquire_instance_lock("gpu-lock:{node}:{idx}", ttl_s=...) с TTL и типизированной потерей через wrap_lease (PR #20)
Обнаружение отказа обработчика	По завершению процесса (только локально)	По истечении пульса (через сеть); типизированный CrashEvent на стороне обработчика (PR #20)
Откат при сбое обработчика	Задача теряется; повторная подача через таймаут	TTL захвата истекает → задача возвращается в очередь автоматически; защищена IdempotencyToken (PR #20)
Несколько драйверов на одном задании	Невозможно (instance lock)	Поддерживается; драйверы - производители и потребители; блокировка экземпляра смигрирована через DataPlane (PR #20)
Защита от двойного применения результата	Отсутствует	IdempotencyToken + compute_content_hash_hex в transition_state.lua (PR #20); заголовок Idempotency-Key в Memory API остаётся этапом 4
Точки отказа	Драйвер, Redis, filesystem, LLM-сервер	Redis, хранилище артефактов, LLM-сервер; драйвер становится восстановимым

4.2. Плюсы

• Горизонтальное масштабирование вычислений. Парк обработчиков расширяется по мере добавления хостов без изменений в драйвере.
• Изоляция отказов. Сбой одного обработчика не влияет на других. Сбой драйвера не теряет работу: выполненные результаты лежат в results:{driver_id}, незавершённые задачи возвращаются в очередь автоматически.
• Типизация по построению. Каждая операция координации возвращает Result[T, E] с исчерпывающим набором ошибочных вариантов; каждая аренда типизирована через линейный токен с дисциплиной однократного потребления; mypy в строгом режиме обнаруживает рассогласования до запуска (PR #20).
• Гетерогенный парк. На одном кластере одновременно живут обработчики разной мощности и назначения. Маршрутизация - через теги.
• Прямая поддержка измерений CUDA-ядер. Эксклюзивный GPU, привязка по типу, разделение между компиляцией (CPU) и измерением (GPU) - без частных случаев в коде.
• Многодрайверный режим без специальных условий. Несколько экспериментов работают одновременно, разделяя один парк обработчиков. Очереди и теги дают естественное разделение.
• Совместимость с существующими системами оркестрации. SLURM, Kubernetes, ssh - все одинаково подходят как способ предоставить хосты, на которых живут обработчики. gigaevo не знает о них; они не знают о gigaevo. Точка соединения - переменная окружения GIGAEVO_REDIS_URL.

4.3. Минусы

• Усложнённая локальная разработка. Сегодня разработчик запускает python run.py и получает работающий процесс. Новая модель требует поднятого Redis (он и сейчас обязателен) и как минимум одного обработчика в отдельном терминале, либо переключения на встроенный режим. Смягчение: режим --embedded (раздел 3.2) сохраняет однопроцессную модель для тестов и разработки.
• Зависимость от хранилища артефактов. Сегодня файлы сброса данных лежат локально. Новая модель требует общедоступного хранилища, видимого и драйверу, и обработчикам. Смягчение: реализация file:// для одного хоста сохраняет текущее поведение; nfs:// и s3:// остаются опциональными расширениями.
• Сериализация по сети вместо внутрипроцессного IPC. Cloudpickle через Redis имеет ограниченную пропускную способность, поэтому для задач с большими аргументами или результатами принципиально, чтобы тяжёлые данные шли через плоскость данных, а не через Redis. Смягчение: действует контракт «всё свыше 64 КБ переносится через хранилище артефактов»; в самом Redis остаются только метаданные и ссылки.
• Дополнительная эксплуатационная нагрузка. Redis потребует мониторинга, резервного копирования и, по необходимости, конфигурации высокой доступности (Sentinel или Cluster). Хранилище артефактов также нуждается во внимании. Смягчение: до уровня примерно 100 обработчиков и 50 тыс. операций в секунду одиночный экземпляр Redis работает надёжно и без дополнительных слоёв.
• Сложность отладки распределённых сценариев. Поток исполнения теперь пересекает сетевые границы, и интерактивный отладчик здесь бессилен. Смягчение: структурированные логи через loguru с тегами worker_id и node_id (задача L-D2); все ключевые события Redis (захват, освобождение, истечение TTL) пишутся в наблюдаемый поток events:audit.

---

5. Этапы внедрения

Архитектура внедряется инкрементально. Каждый этап оставляет систему в работающем состоянии и приносит измеримую пользу.

Этап 0. Готовое состояние (после PR #17).
Сетевой слой переписан, прямой httpx из кодовой базы исключён, опубликован стабильный фундамент для асинхронных HTTP-вызовов.

Этап 1. Подготовительная гигиена (выполнено в PR #14).

• WorkerConfig: явная сериализуемая конфигурация обработчика (frozen dataclass, slots). from_env распознаёт GIGAEVO_EXECUTOR_MAX_WORKERS, GIGAEVO_EXECUTOR_IDLE_TIMEOUT_S, GIGAEVO_EXECUTOR_SPILL_DIR, GIGAEVO_EXECUTOR_NODE_ID и GIGAEVO_EXECUTOR_DISABLE_PDEATHSIG.
• Идентификация обработчиков: node_id из socket.gethostname() (либо GIGAEVO_EXECUTOR_NODE_ID), worker_id через uuid4. Оба идентификатора пробрасываются в окружение обработчика (GIGAEVO_NODE_ID, GIGAEVO_WORKER_ID) и в каждую запись ExecutionMetrics, прикреплённую к результату вызова.
• Точки расширения жизненного цикла на уровне вызова: on_submit, on_complete, on_shutdown, вынесены в отдельный опциональный протокол LifecycleObservable. Хуки уровня процесса-обработчика (on_worker_start, on_worker_idle, on_worker_exit) отложены до Этапа 3, когда появится сетевой источник этих событий.

Этап 2. Абстракция исполнителя (выполнено в PR #14).

• Минимальный интерфейс ExecutorBackend с двумя методами: execute и shutdown. Наблюдение жизненного цикла вынесено в отдельный опциональный протокол LifecycleObservable; реализации без полезных событий не обязаны добавлять заглушки. recycle_worker отложен: вводится одновременно с механизмом принудительного перезапуска обработчика по счётчику вызовов.
• Реализация LokyBackend соответствует обоим протоколам и сохраняет существующее поведение loky (ленивое создание пула, изолированные spill-каталоги, PR_SET_PDEATHSIG на Linux, скрабинг окружения по белому списку + префиксам).
• Точки вызова (runtime_metrics.py, optimization/utils.py, optimization/optuna/stage.py, execution.py) продолжают использовать run_exec_runner без изменений; добавленный кварг backend= позволяет внедрить любой ExecutorBackend без правки вызывающего кода. Singleton по умолчанию (default_loky_backend()) сохраняет совместимость.

Этап 2.5. Координационный субстрат DataPlane (выполнено в PR #20).

• gigaevo.dataplane.DataPlane: единственный типизированный асинхронный координатор для всех взаимодействий с Redis.
• Восемь атомарных Lua-скриптов: transition_state, archive_swap, instance_lock_acquire/renew/release, counter_inc, lwwr_set, bounded_list_push.
• Линейные токены Token[Tag] с mint_root / mint_split / mint_split_n, EngineRoot для разметки подпространств.
• Типизированный возврат Result[T, E] с дискриминированной иерархией DataPlaneError (19 структурированных типов исключений).
• Обёртка потерянного владения wrap_lease + CrashWatchedHandle + OneShotFlag + CrashEvent: потеря аренды видна типизированно через match, без out-of-band-флагов.
• Гибридный логический таймштамп HlcTimestamp с лексикографически сравнимой hex-упаковкой.
• NewType-словарь для фазы 3: StreamName, ConsumerGroup, ConsumerName, EventId, CausationId, CorrelationId, IdempotencyToken, NodeId плюс пять других идентификаторов.
• Дисциплинарные ограждения: линтер ruff TID251 блокирует прямые импорты redis / redis.asyncio / redis.exceptions вне субстрата; AST-инспектор отвергает неаннотированные широкие except внутри координатора.
• Подсистемы, мигрированные через DataPlane: машина состояний программ (FSM-таблица в Redis, серверная валидация переходов), архив элит (try_replace_elite + archive_swap.lua), реестр многоруких бандитов (crdt_inc + actor-партиционирование), словарь подсказок (bounded_list_push, crdt_inc, set_add), распределённая блокировка экземпляра (acquire/renew/release_instance_lock + wrap_lease), миграционная шина (частично; полное закрытие отложено в фазу 2).

Этап 3. Сетевая очередь задач.

• Реализация RedisStreamWorkerBackend поверх dp.emit / dp.consume с использованием существующих в PR #20 NewType-идентификаторов (StreamName, ConsumerGroup, EventId); полная реализация consume поверх XREADGROUP / XACK отнесена к фазе 2 substrate настоящего RFC.
• Абстракция ArtifactStore с реализацией file:// (поведение совпадает с текущими файлами сброса данных).
• Команда CLI gigaevo worker.
• Встроенный режим --embedded с реализацией ExecutorBackend поверх asyncio.Queue; контракт обмена (Result[T, E] + Token[Tag] + Versioned[T] из PR feat(dataplane): typed Redis coordination plane with atomic Lua substrate, FSM migration, and Hydra hardening #20) инвариантен относительно физического размещения участников.

Этап 4. Усиление надёжности.

• Продлеваемый пульс с продлением аренды захвата через dp.renew_instance_lock + wrap_lease (примитив уже доступен через PR #20).
• Дедупликация записи в Memory API через заголовок Idempotency-Key (требует согласованных изменений на стороне сервера); координационная сторона уже располагает IdempotencyToken + compute_content_hash_hex из PR feat(dataplane): typed Redis coordination plane with atomic Lua substrate, FSM migration, and Hydra hardening #20.
• Ограничение глубины очереди (мягкий backpressure через периодический опрос XLEN).
• Сторожевой таймер для застрявших задач.
• Кросс-драйверная инвалидация кеша Memory API через Redis pub/sub.

Этап 5. Особенности гетерогенного парка.

• Теги обработчиков и подписка на очереди по тегам.
• Эксклюзивный захват GPU через фабричную функцию поверх dp.acquire_instance_lock с ключом gpu-lock:{node}:{idx}; без модификации Lua-субстрата (поставлено в PR #20).
• Альтернативная реализация ArtifactStore для NFS / S3.

После Этапа 2 (интерфейс ExecutorBackend) и Этапа 2.5 (DataPlane substrate) меняются следующие места:

Место	Что меняется
gigaevo/programs/stages/python_executors/wrapper.run_exec_runner	Превращается в тонкую обёртку над ExecutorBackend.submit. Сохраняет существующую сигнатуру для обратной совместимости
gigaevo/programs/stages/runtime_metrics.py	Без изменений: продолжает вызывать run_exec_runner
gigaevo/programs/stages/optimization/utils.py	Без изменений
gigaevo/programs/stages/optimization/optuna/stage.py	Без изменений
gigaevo/programs/stages/python_executors/execution.py	Без изменений
run.py	После PR #20: использует build_dataplane, build_engine_root, build_actor_identity плюс семейство wire_* для пост-instantiate-привязки координатора. Этап 3 добавляет инициализацию RedisStreamWorkerBackend на основе конфигурации Hydra
tests/conftest.py	Использует LokyBackend или встроенный backend в зависимости от теста

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

---

Возможное пространство ключей Redis

Ключ или поток	Тип	TTL	Назначение
tasks:{tag}	Stream	бессрочно (MAXLEN)	Очередь задач, сегментированная по типу обработчика
results:{driver_id}	Stream	бессрочно (MAXLEN)	Результаты, адресованные конкретному драйверу
evo:migrate	Stream	бессрочно (MAXLEN)	Миграция программ между островами (уже существует)
{prefix}:events:status	Stream	бессрочно (MAXLEN ~ 10000)	Поток событий FSM-переходов программ (через transition_state.lua PR #20)
events:audit	Stream	бессрочно (MAXLEN)	Аудит-лог операций координации
claim:{job_id}	String	равен deadline_s задачи	Захват задачи обработчиком (через dp.acquire_instance_lock PR #20), продлеваемый
heartbeat:{worker_id}	String	30 с	Пульс обработчика, продлеваемый каждые 10 с (через dp.renew_instance_lock)
gpu-lock:{node}:{idx}	String	равен deadline_s задачи	Эксклюзивный захват GPU (фабричная функция поверх dp.acquire_instance_lock)
lease:{role}	String	равен интервалу продления × 3	Лидерская аренда роли (например, lease:gam-rebuilder); фабричная функция поверх Lua-скриптов PR #20
idem:{op_id}	String	24 ч	Журнал идемпотентности Memory API (фаза 4)
{prefix}:program:{pid}:idem	Hash	300 с	Журнал идемпотентности FSM-переходов программ (через IdempotencyToken PR #20)
{prefix}:fsm:program_state	Hash	бессрочно	FSM-таблица, загружаемая в Redis на dp.startup() (PR #20)
{prefix}:program:{pid}	String	бессрочно	JSON-блоб программы, мутируемый только через dp.transition_program_state (PR #20)
{prefix}:status:{state}	Set	бессрочно	Множества программ по состоянию, обновляемые атомарно вместе с FSM-переходом (PR #20)
{prefix}:ts	String	бессрочно	Счётчик эпох, инкрементируемый атомарно при каждом FSM-переходе (PR #20)
{prefix}:archive / :archive:reverse / :archive:scores	Hash	бессрочно	MAP-Elites: прямой/обратный индексы плюс score; через dp.try_replace_elite + archive_swap.lua (PR #20)
{prefix}:counter:{key}	Hash	бессрочно	G-счётчики с актор-партиционированием через dp.crdt_inc + counter_inc.lua (PR #20)
{prefix}:lwwr:{key}	Hash	бессрочно	LWW-регистр с разрешением равенства по HLC через dp.lwwr_set + lwwr_set.lua (substrate фазы 2 PR #20)
instance-lock:{prefix}	String	продлевается	Блокировка экземпляра через dp.acquire_instance_lock (PR #20 §9.4)
llm_pool:{name}:inflight	Hash	бессрочно	Существующий счётчик пула LLM
llm_pool:{name}:cooldown:{hash}	String	cooldown_secs	Существующий cooldown сервера LLM

Все потоки ограничены по длине через MAXLEN ~ N, где N подбирается под ожидаемую глубину очереди для соответствующего тега.