Temporal — durable-движок, на котором крутится пайплайн

Temporal — это что вообще

Когда в коде встречаются @workflow.signal, workflow.execute_activity и подобные — это не наш код, а API библиотеки Temporal. У Temporal две половины, и важно не путать их:

🖥️ Temporal Server

отдельный сервис (запущен в docker-compose). Хранит состояние каждого workflow в базе (Postgres), отдаёт задачи воркеру, переживает перезапуски.

📦 temporalio (Python-SDK)

библиотека, которую наш код импортирует. Даёт декораторы и функции: workflow.defn, workflow.signal, workflow.execute_activity и т.д.

Пайплайн (класс DevPipeline) — это workflow, написанный на Temporal API. Воркер (worker.py) подключается к серверу и выполняет эти workflow. Сервер «думает», а воркер «делает» — так и получается живучесть: даже если воркер упадёт, сервер помнит всё и передаст работу другому.

Ключевая идея: код workflow детерминирован и не делает I/O напрямую. Любая «настоящая» работа (HTTP, Nomad, БД) выносится в activity — а Temporal умеет переигрывать (replay) историю workflow, чтобы восстановиться после сбоя.

Словарь: что встречается в нашем коде

Ниже — каждая Temporal-конструкция из workflows.py, простыми словами.

Декораторы: объявление частей workflow

`@workflow.defn`

«Это — workflow». Вешает метку на класс DevPipeline, чтобы Temporal знал: вот сценарий, который можно запускать.

`@workflow.run`

«А вот точка входа». Помечает метод run() — именно его Temporal вызывает, когда стартует workflow.

`@workflow.signal`

«Сюда можно постучать снаружи». Помечает обработчики входящих событий (agent_done, human_verdict) — их вызывает не сам workflow, а внешний мир (агент, человек).

`@workflow.query`

«Сюда можно заглянуть, ничего не меняя». Помечает метод-«вопрос» (get_rework_count): спрашиваем состояние workflow снаружи, а он ничего не трогает.

Функции: действия workflow

`workflow.execute_activity(...)`

«Выполни шаг с I/O». Единственный способ для workflow сделать «настоящую» работу (диспатч в Nomad, HTTP к артефактам). Temporal записывает вызов и результат в историю.

`workflow.wait_condition(pred, timeout)`

«Спи, пока условие не станет true, но не дольше timeout». Так workflow ждёт сигнал агента — и делает это durably.

`workflow.info()`

«Расскажи, кто я». Возвращает метаданные текущего workflow: его id, task_queue и т.д. — нужны, чтобы собрать корреляцию шага.

`workflow.logger`

«Пиши в лог workflow». Логирование, привязанное к конкретному прогону — видно в Temporal UI.

`workflow.start_child_workflow(...)`

«Запусти другой workflow и не жди его». Так пайплайн запускает проверку (VerificationWorkflow) «в стороне».

`workflow.unsafe.imports_passed_through()`

«Разреши импорт тяжёлых модулей». Внутри workflow нельзя просто импортировать что угодно (ломает детерминизм); этот блок говорит Temporal: эти импорты безопасны, пропусти их.

Соседние типы

`RetryPolicy(maximum_attempts=3)`

«Если activity упала — пробуй ещё, до 3 раз». Temporal сам делает ретраи, нам не нужно писать циклы.

`ActivityError`

исключение, которое Temporal поднимает в workflow, когда activity упала — ловим и решаем, что делать.

`ApplicationError`

«ошибка, которую выбросил сам activity» — лежит «под» ActivityError, читаем из него причину.

`timedelta(...)`

таймауты и сроки. В workflow нельзя использовать «реальные часы» (time.time()) — только это, чтобы replay был детерминированным.

Как это выглядит в нашем коде

Фрагмент из workflows.py с «переводом» каждой строки:

@workflow.defn                     # «это workflow»
class DevPipeline:

    @workflow.signal               # «сюда стучится агент, когда закончил шаг»
    async def agent_done(self, exit_code: str, result: dict, step_id: str = ""):
        ...

    @workflow.signal               # «сюда стучится человек со своим вердиктом»
    async def human_verdict(self, artifact_id="", step_id="", feedback=""):
        ...

    @workflow.query                # «сюда можно заглянуть снаружи»
    def get_rework_count(self) -> int:
        return self._rework_count

    @workflow.run                  # «точка входа — с этого метода всё начинается»
    async def run(self, spec_id: str, task_id: str = "") -> dict:
        ...
        info = workflow.info()     # «кто я?» → нужен id workflow для корреляции

        await workflow.execute_activity(          # «сделай работу (I/O)»
            "prepare_agent_run", ...,             # имя activity + аргументы
            start_to_close_timeout=timedelta(seconds=30),   # потолок времени
            retry_policy=RetryPolicy(maximum_attempts=3),   # до 3 попыток
        )

        await workflow.wait_condition(            # «спи, пока не придёт результат агента»
            lambda: self._agent_result is not None,
            timeout=_STEP_TIMEOUT,                # но не дольше 30 минут
        )

Заметили: в workflow нет ни одного прямого HTTP-вызова или похода в Nomad. Всё это спрятано в activities (их вызывает execute_activity). Так Temporal может «переиграть» историю и восстановиться — детерминированный код это умеет, а сетевые походы — нет.

Кто что делает: сервер против воркера

Часть	Что делает	Где живёт
Temporal Server	хранит состояние workflow, «будит» их по сигналам и таймерам, отдаёт activity воркеру	docker-compose (сервис `temporal`)
Temporal Postgres	база истории workflow — источник durable-памяти	docker-compose
Worker	выполняет workflow-код и activities; поллит очередь	`worker.py` (наш код)
Temporal UI	веб-интерфейс: видно каждый прогон, его таймлайн и сигналы	http://localhost:8081 (прод: :8081 на 10.1.2.100)

Почему именно это даёт «живучесть»

Workflow делает шаг → Temporal записывает это в историю (Postgres).
Workflow «засыпает» (ждёт сигнал или таймер) — не занимая воркер.
Воркер может упасть, перезапуститься — сервер это переживёт.
Когда приходит сигнал (агент готов, человек решил) — сервер будит workflow.
Workflow переигрывает историю с начала (replay) и продолжает с того же места — ничего не потеряно и не задвоено.

Именно поэтому пайплайн не боится, что агент работает полчаса, а воркер перезапустят: «память» хранит сервер, а не процесс.