# Рабочие процессы

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

## Что такое рабочие процессы?

Рабочий процесс — это упорядоченная последовательность шагов. Каждый шаг — это класс PHP, который получает накопленное состояние всех предыдущих шагов и возвращает новые данные для объединения с состоянием. Рабочий процесс продвигается по одному шагу за раз, а каждый переход выполняется в рамках атомарной транзакции MySQL.

```php
use Queuety\Queuety;

$workflow_id = Queuety::workflow( 'generate_report' )
    ->then( FetchDataHandler::class )
    ->then( CallLLMHandler::class )
    ->then( FormatOutputHandler::class )
    ->dispatch( [ 'user_id' => 42 ] );
```

## Накопление состояния

Начальная полезная нагрузка, переданная в `dispatch()`, становится исходным состоянием. Возвращенные каждым шагом ключи объединяются с состоянием. Ключи, начинающиеся с `_`, зарезервированы для управляющих значений, таких как `_next_step`, и не объединяются с общедоступным состоянием:

```
dispatch(['user_id' => 42])

Step 0: FetchDataHandler   -> returns ['user_name' => 'Alice']
  state: {user_id: 42, user_name: 'Alice'}

Step 1: CallLLMHandler     -> returns ['llm_response' => '...']
  state: {user_id: 42, user_name: 'Alice', llm_response: '...'}

Step 2: FormatOutputHandler -> returns ['report_url' => '/reports/42.pdf']
  state: {user_id: 42, user_name: 'Alice', llm_response: '...', report_url: '/reports/42.pdf'}
```

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

## Возможность возобновления

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

Граница шага — это одна транзакция MySQL, которая атомарно:
1. Обновляет состояние рабочего процесса результатом шага
2. Помечает задание текущего шага как завершенное
3. Ставит в очередь задание следующего шага

Это гарантирует, что состояние никогда не будет потеряно, даже при сбоях, тайм-аутах или фатальных ошибках PHP.

## Жизненный цикл рабочего процесса

```
dispatch() -> running -> step 0 -> step 1 -> ... -> step N -> completed
                  |                   |
                  |                   +-> failed -> retry -> running (from failed step)
                  |
                  +-> paused -> resumed -> running
                  |
                  +-> waiting_for_signal -> signal received -> running
                  |
                  +-> waiting_for_workflows -> dependency settles -> running
                  |
                  +-> cancelled (via cancel_workflow)
```

| Статус | Описание |
|---|---|
| `running` | Активно выполняются шаги |
| `completed` | Все шаги успешно завершены |
| `failed` | Шаг завершился ошибкой после исчерпания повторных попыток |
| `paused` | Приостановлен вручную. Текущая выполняемая задача может завершиться, но последующие шаги не ставятся в очередь до возобновления |
| `waiting_for_signal` | Ожидание внешнего сигнала через `Queuety::signal()` |
| `waiting_for_workflows` | Ожидание завершения одного или нескольких других рабочих процессов через `wait_for_workflow()` или `wait_for_workflows()` |
| `cancelled` | Явная отмена через `Queuety::cancel_workflow()`. Ожидающие задания скрываются, а при наличии выполняется обработчик очистки. |

## Управление рабочими процессами

Проверка статуса:

```php
$state = Queuety::workflow_status( $workflow_id );
echo $state->status->value;   // 'running'
echo $state->current_step;    // 1
echo $state->total_steps;     // 3
print_r( $state->waiting_for ); // ['42'] while waiting on another workflow
print_r( $state->wait_details ); // matched/remaining blockers, result key, correlation info
```

Приостановка и возобновление:

```php
Queuety::pause_workflow( $workflow_id );
Queuety::resume_workflow( $workflow_id );
```

Повторный запуск рабочего процесса с шага, завершившегося ошибкой:

```php
Queuety::retry_workflow( $workflow_id );
```

Отмена рабочего процесса и запуск обработчиков очистки:

```php
Queuety::cancel_workflow( $workflow_id );
```

Подробности об обработчиках очистки и о том, что происходит при отмене рабочего процесса, см. в разделе [Отмена](https://queuety.inline0.dev/ru/docs/workflows/cancellation).

## Параметры рабочего процесса

Настройка очереди, приоритета и поведения при повторных попытках для всех шагов рабочего процесса:

```php
use Queuety\Enums\Priority;

Queuety::workflow( 'important_pipeline' )
    ->on_queue( 'pipelines' )
    ->with_priority( Priority::High )
    ->max_attempts( 5 )
    ->on_cancel( PipelineCleanupHandler::class )
    ->prune_state_after( 3 )
    ->then( StepA::class )
    ->then( StepB::class )
    ->dispatch( $payload );
```

Метод `on_cancel()` регистрирует обработчик очистки, запускаемый при отмене рабочего процесса. См. раздел [Отмена](https://queuety.inline0.dev/ru/docs/workflows/cancellation).

Метод `prune_state_after()` включает автоматическое удаление результатов старых шагов, чтобы ограничить размер состояния рабочего процесса. См. раздел [Очистка состояния](https://queuety.inline0.dev/ru/docs/features/state-pruning).

Используйте `compensate_with()`, чтобы назначить обработчик компенсации для отдельного шага. Используйте `compensate_on_failure()`, если нужно автоматически запускать компенсацию завершенных шагов после сбоя рабочего процесса, а не только при явной отмене. См. раздел [Компенсация](https://queuety.inline0.dev/ru/docs/workflows/compensation).

Используйте ограничения рабочего процесса, чтобы безопаснее управлять длительной оркестрацией: `version()` помечает определение версией на уровне приложения, Queuety также сохраняет детерминированный хеш определения для каждого запуска, `idempotency_key()` обеспечивает надежную отправку при повторных попытках или дублирующих запросах, а `max_transitions()`, `max_for_each_items()` и `max_state_bytes()` завершают вышедшие из-под контроля запуски до того, как ситуация усугубится. См. раздел [Ограничения](https://queuety.inline0.dev/ru/docs/workflows/guardrails).

## Типы шагов

Queuety поддерживает одиннадцать типов шагов в рабочем процессе:

- [Последовательные цепочки](https://queuety.inline0.dev/ru/docs/workflows/chains) с помощью `->then()` для выполнения по одному шагу за раз
- [Параллельные группы](https://queuety.inline0.dev/ru/docs/workflows/parallel) с помощью `->parallel()` для одновременного выполнения
- [Динамический перебор элементов](https://queuety.inline0.dev/ru/docs/workflows/for-each) с помощью `->for_each()` для ветвлений, обнаруженных во время выполнения, и явного завершения
- [Асинхронная передача управления](https://queuety.inline0.dev/ru/docs/workflows/async-handoffs) с помощью `->start_workflows()` или `->start_agents()` для запуска независимых дочерних рабочих процессов по элементам, обнаруженным во время выполнения
- [Условное ветвление](https://queuety.inline0.dev/ru/docs/workflows/conditional) с помощью именованных шагов и `_next_step`
- [Повторения](https://queuety.inline0.dev/ru/docs/workflows/repeats) с помощью `->repeat_until()` и `->repeat_while()` для надежного возврата к предыдущим именованным шагам
- [Запуск рабочих процессов](https://queuety.inline0.dev/ru/docs/workflows/run-workflows) с помощью `->run_workflow()` для вложенной композиции
- [Надежные задержки](https://queuety.inline0.dev/ru/docs/workflows/delays) с помощью `->delay()` для временных задержек
- [Сигналы](https://queuety.inline0.dev/ru/docs/workflows/signals) с помощью `->wait_for_signal()`, `->wait_for_signals()`, `->wait_for_approval()`, `->wait_for_input()` и `->wait_for_decision()` для внешних событийных триггеров
- [Зависимости рабочих процессов](https://queuety.inline0.dev/ru/docs/workflows/dependencies) с помощью `->wait_for_workflow()`, `->wait_for_workflows()` и `->wait_for_agents()` для координации между рабочими процессами
- [Потоковые шаги](https://queuety.inline0.dev/ru/docs/workflows/streaming) с помощью интерфейса `StreamingStep` для надежного вывода по частям

## Типовые сценарии

Если вы создаете автономные или полуавтономные системы, начните с раздела [Оркестрация агентов](https://queuety.inline0.dev/ru/docs/workflows/agent-orchestration). В нем показано, как `for_each()`, `start_agents()`, `wait_for_agents()`, `wait_for_workflow()` и `wait_for_decision()` взаимодействуют в реальных сценариях планирования и выполнения задач, а также проверки человеком.