Устойчивость и resume¶

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

Что сохраняется¶

SessionStore сохраняет Context под session_id — одним blob'ом для FileKVBackend/InMemoryKVBackend или снапшотом плюс журналом коммитов для AppendBackend:

артефакты (типизированные, версионированные) и связи между ними;
цепочку коммитов (Context.history()), так что провенанс и diff/ rollback/replay продолжают работать;
очередь отложенных триггеров — события, которые ещё не разослало ни одно поколение.

Payload несёт format_version: данные от более новой версии reactifact отклоняются явно, старые проходят зарегистрированные миграции (Context.from_dict).

С AppendBackend (SQLiteKVBackend, PostgreSQLKVBackend) хранилище не переписывает весь Context на каждый коммит: первое сохранение (или первое после compaction/rollback) пишет снапшот, а последующие дописывают только новые коммиты и текущую очередь событий. FileKVBackend и InMemoryKVBackend сохраняют поведение полного blob'а.

SessionStore(backend, compact_every=N) ограничивает append-журнал: примерно каждые N сохранений снапшот перезаписывается, а записи очищаются. По умолчанию — 100; compact_every=None оставляет журнал до явного context.compact() или rollback. BranchStore принимает ту же опцию для каждого ключа ветки.

При дефолтном session_save_policy="per_commit" рантайм сохраняется на каждой границе поколения, после того как батч триггеров этого поколения израсходован. Поэтому снапшот консистентен: уже израсходованный триггер не реплеится, а тот, чей consumer так и не запускался, — остаётся.

from reactifact import Runtime, SessionStore
from reactifact.checkpoints import SQLiteKVBackend

store = SessionStore(SQLiteKVBackend("sessions.sqlite3"))
session = await store.open("user-1")            # загружает или создаёт
runtime = Runtime(session.context, agents=[...], session=session)
await runtime.arun()

# после краха / рестарта — новый процесс, в памяти ничего нет
session = await store.open("user-1")
runtime = Runtime(session.context, agents=[...], session=session)
await runtime.arun()   # продолжает: отложенные триггеры на месте

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

Гарантия: at-least-once¶

Возобновление — at-least-once, а не exactly-once. Шаг, который уже закоммитился до краха, может выполниться снова. Если он не идемпотентен, это даст дубликат артефакта (или повторный внешний side effect).

Давайте produce stable id, чтобы повторный прогон делал upsert, а не дублировал (§42):

@step(produces=Summary, consumes=Msg)
async def summarize(call):
    doc = call.trigger
    # Stable id делает повтор идемпотентным: повторное создание вернёт тот же
    # артефакт и не породит второго события.
    call.effects.create(Summary(text=...), id=f"summary:{doc.data.url}")

effects.create_once_from(...) (stable id, выведенный из триггера) и Consume(..., debounce=True) — остальные инструменты для той же цели.

Что не покрывается¶

Внешние side effects внутри produce. Письмо, запись файла или вызов стороннего API, сделанные напрямую в produce, могут повториться при retry. Так делать не нужно — пишите намерение: effects.act(kind=..., key=..., payload=...) создаёт PendingAction, а Runtime(dispatcher=...) выполняет его один раз после коммита намерения (см. паттерн outbox в паттернах). Сам I/O фреймворк по-прежнему не может сделать exactly-once — диспетчер должен быть идемпотентен по idempotency_key (передаётся в action.data), и это закрывает остаточное окно at-least-once. (call.request доступен для correlation id; см. reactifact.request.)
Фонового relay нет. Фреймворк не запускает процесс, который сам бы бесконечно ретраил outbox: следующий arun() дренирует закоммиченные действия, а flush_pending_actions() дренирует их без поколения — вызывайте его на старте и периодически (своим таймером в приложении), чтобы действие, закоммиченное прямо перед крахом, не зависло. Retry-политика — на стороне приложения: падающий диспетчер помечает действие failed (увеличивая attempts), вызывает Runtime(on_dispatch_error=...), если задан, и пробрасывает исключение; context.retry_action(id) возвращает его в очередь.
Работа «в полёте». Шаг, который выполнялся в момент краха, просто перезапускается; durable — только закоммиченные поколения.
Бюджеты сбрасываются на прогон. Счётчики Budget (max_runs, max_seconds) живут в рамках одного вызова arun(); возобновление начинает свежий бюджет, а не остаток старого.
Гранулярность per_turn. При session_save_policy="per_turn" сессия сохраняется один раз в конце хода, поэтому крах посреди хода откатывает весь ход и перезапускает его. per_commit (дефолт) сохраняется на каждой границе поколения.

Outbox в проде¶

Outbox-примитив безопасен по построению, но это возможность библиотеки, а не сервис доставки: он не запускает relay и не придумывает retry-политику. Чтобы внешние эффекты (почта, webhook, платежи) работали в проде, на вас четыре вещи:

Идемпотентный диспетчер. Передавайте action.data.idempotency_key во внешнюю систему, чтобы окно at-least-once (крах между отправкой и коммитом dispatched) закрывалось на той стороне. Никогда не генерируйте новый ключ при повторе.
Дренаж. Вызывайте flush_pending_actions() на старте и периодически своим таймером, чтобы действие, закоммиченное прямо перед крахом, не зависло. Следующий arun() тоже дренирует — но только если ход действительно запускается.
Retry-политика. Падающий диспетчер помечает действие failed (увеличивая attempts) и пробрасывает исключение; оберните диспетчер своим retry/backoff и/или повесьте Runtime(on_dispatch_error=...) для алерта, а context.retry_action(id) — чтобы вернуть в очередь. Backoff намеренно не в рантайме.
Наблюдаемость. Смотрите context.list_artifacts(PendingAction) (или CLI replay) на failed и долго висящие действия и алертите по ним.

С этими четырьмя пунктами фича готова к проду; если хочется, чтобы фреймворк сам бесконечно ретраил без таймера приложения, — это брокер/relay, вне области по замыслу (§75). Обёртку-retry на уровне диспетчера реализуйте в приложении.

Эволюция схемы¶

Тип сохранённого артефакта хранится строкой (Artifact.data_type, data_type у скомпилированных Create/Update, Event.artifact_type) — по умолчанию это Python qualified name (module.Qualname). Регистрировать ничего не нужно: id выводится автоматически при сохранении, а при загрузке resolve сам импортирует qualified name, поэтому обычная модель уровня модуля перезагружается без бойлерплейта. reactifact.types нужен лишь чтобы отвязать эту идентичность от структуры модулей, когда вы этого хотите.

Регистрация важна в трёх случаях:

Переименование или перенос — стабильный type_id сохраняет старые сессии/записи загружаемыми после переименования или переноса модели.
Алиасы — payload'ы с прежним id продолжают резолвиться.
Миграция схемы — хук migrate переименовывает поля / подставляет дефолты до валидации.

from reactifact import register_type

register_type(
    Answer,
    type_id="answer",                                   # что пишется с этого момента
    aliases=["old.pkg.Answer"],                         # что несут старые payload'ы
    migrate=lambda d: {**d, "prose": d.pop("text", "")},  # переименование поля
)

Когда нужен только стабильный id (без алиасов/миграции), достаточно classvar TYPE_ID на модели — вызывать ничего не надо. Вызывайте register_type при импорте, рядом с моделью.

type_id_of(model) — id, который пишется при сохранении: зарегистрированный type_id, classvar TYPE_ID, иначе qualified name.
resolve(type_id) — класс при загрузке: реестр → алиас → импорт по qualified name, поэтому дореестровые payload'ы продолжают грузиться.
migrate выполняется над raw-dict до валидации (переименования полей, дефолты).

Единственная реальная оговорка: qualified name загружается, только если класс импортируем в новом процессе. Модель, определённая в __main__, в ячейке ноутбука, внутри функции (<locals>) или собранная через pydantic.create_model, — нет; её id не резолвится при resume/replay. Задайте такой модели явный type_id (и реальный модуль, где она живёт); ошибка resolve называет id, который не нашла, и указывает на register.

Это покрывает артефакты, операции коммитов и события — состояние всей сессии и очередь отложенных триггеров — так что апгрейд с переименованием/переносом модели больше не ломает resume, replay и golden-снапшоты.

Resume vs. replay¶

Две разные операции, которые часто путают:

Resume (session.open(...) + Runtime.arun()) продолжает прерванный прогон: рассылает отложенные триггеры заново и может перезапустить produce.
Replay (reactifact replay, ReplayLLM) восстанавливает почему завершённый прогон пришёл к своему состоянию — детерминированно, без перезапуска агентов; см. Replay.

Первое — исполнение, второе — аудит.

Дедлайны, graceful shutdown и активные прогоны¶

Budget.max_seconds работает как жёсткий per-turn дедлайн: runtime отменяет генерацию в полёте (зависший LLM-вызов действительно прерывается), а не только проверяет между генерациями. Батч триггеров отменённой генерации не консьюмится, поэтому следующий arun() её возобновит; ход завершается RunOutcome.BUDGET_TIME_EXCEEDED.

При остановке процесса Runtime.request_stop() просит прогон остановиться на границе генерации — текущая генерация доигрывается (её коммит ложится), новая не начинается, выставляется RunOutcome.STOPPED. await runtime.ashutdown(timeout=…) делает это и ждёт ход, форс-отменяя его по timeout. Общие RuntimeResources он не закрывает; закрывайте их в своём lifespan (resources.aclose() или ResourceScope).

Активные прогоны процесса видны только на чтение:

from reactifact.runtime import active_runs, cancel_run

len(active_runs())           # сигнал readiness
active_runs()[0].session_id  # что именно выполняется
cancel_run(run_id)           # отменить один ход

active_runs() — только внутри процесса: одна event loop, без персистентности и cross-process. Это ops/readiness-представление, а не очередь задач.

Ограничение долгоживущего контекста¶

Сессия, живущая много ходов, копит каждый коммит и каждую версию артефакта вечно — память (и to_dict) растут без границ. Context.compact ограничивает их так же, как git сжимает историю:

session.context.compact(keep_commits=50, keep_versions=2)

Он сворачивает все коммиты старше последних keep_commits в baseline-снапшот и (при keep_versions) обрезает хранимую историю версий каждого артефакта. Абсолютные version/head_id сохраняются, поэтому context_hash, номера версий коммитов и resume не меняются.

Две честные оговорки:

Это необратимо. checkout/diff ниже получившегося baseline бросают исключение (Context.compacted_at его показывает), а провенанс артефактов, последний раз записанных до baseline, теряется — операции, их создавшие, свёрнуты.
Артефакты не вытесняются. Чтобы удалять старые артефакты (например, сообщения за окном), используйте рецепты памяти (WindowPruner, RollingDigestSummarizer, Recipes); compact ограничивает историю, а не рабочий набор.

См. также¶

Модель исполнения — цикл поколений и его явные не-гарантии.
Концепции — Context, коммиты и очередь событий.
Replay — проверка context_hash завершённого прогона.