Типы шагов конвейера

У шага конвейера две независимые оси:

• Вид (StepKind) — как движок планирует шаг относительно его триггеров.
• Ключ тела (BuiltInBodyKeys) — какое зарегистрированное замыкание тела выполняет работу шага.

Виды шагов

Определены в packages/cc_domain/lib/features/pipelines/domain/entities/step_kind.dart.

Вид	Семантика планирования
trigger	Обязательная точка входа — ровно одна на шаблон, всегда первый узел. Сам никакой работы не делает; то, что инициирует запуск (вручную / по событию / по расписанию), отслеживается отдельно в строках PipelineTrigger. Его тело — pipeline.trigger, не выполняющее ничего.
listen	Срабатывает, когда завершаются все исходные шаги (в StepTrigger.sourceStepIds).
join	Срабатывает, когда все шаги в waitForStepIds достигают терминального состояния.
router	Условное ветвление — тело возвращает StepResult.route(key), что активирует только то нижестоящее ребро, чей routeKey совпадает; невыбранные ветви помечаются пропущенными.
forEach	Планируется как listen (срабатывает, когда завершаются все sourceStepIds). Именно зарегистрированное тело flow.forEach выполняет перебор; движок не порождает по одному дочернему шагу на элемент.
terminal	Сторожевой элемент планировщика без тела. Запуск конвейера завершается, когда завершается терминальный шаг.

Тела шагов

Замыкания тел, зарегистрированные под константами из BuiltInBodyKeys (packages/cc_domain/lib/features/pipelines/domain/templates/builtin_template_seeds.dart). Всего их 25, и все перечислены ниже. Конфигурация читается из PipelineNodeConfig шага (см. ниже), если не указано иное; «ключи состояния» читаются из состояния конвейера / payload триггера во время выполнения.

Шаг, чей bodyKey не зарегистрирован, выдаёт предупреждение при загрузке шаблона и затем никогда не выполняется — запуск из-за этого не проваливается. Шаги trigger и terminal от этого предупреждения освобождены: триггер всё равно выполняет тело pipeline.trigger, которое ничего не делает; у шага terminal тела нет.

pipeline.trigger

Тело обязательного узла trigger, не выполняющее никакой работы, — завершается немедленно, чтобы движок разошёлся по нижестоящим слушателям. Конфигурации нет.

pipeline.bashScript

Шаг оболочки без агента (bash -c). Заполнители {{key}} в скрипте подставляются из состояния и payload триггера экранированными для оболочки значениями.

Поле конфигурации	Описание
script	Команда оболочки для выполнения (обязательно)
outputKey	Ключ состояния, получающий обрезанный stdout при завершении с кодом 0

Выполняется с cwd <cc_root>/pipelines/<pipelineRunId>/ (общим для всех bash-шагов запуска) и с GITHUB_TOKEN в окружении. Также записывает <stepId>_runDir в состояние. Завершение с ненулевым кодом проваливает шаг, а сообщением об ошибке становится хвост stderr.

conversation.promptAgent

Обобщённый узел «промпт и отправка»: рендерит промпт, отправляет агента в разговор и приостанавливается до завершения запуска. Payload submit_output агента собирается в состояние под outputKey.

Поле конфигурации	Описание
agentId	UUID агента в пределах рабочего пространства для отправки (обязательно)
prompt	Текст инструкции; поддерживает подстановку {{key}} (обязательно)
outputKey	Ключ состояния для собранного вывода
outputSchema	JSON Schema, которой должен соответствовать вывод
label	Человекочитаемая метка на холсте
extras.spaceId	Существующий разговор для работы (по умолчанию — скрытый)
extras.conversationTitle	Именованный поток внутри этого пространства (необязательно)
extras.mode	Читается на узле, который открывает комнату (messaging.createSpace), а не здесь. По умолчанию review.

pipeline.condition

Тело роутера — читает config.extras, вычисляет условие и возвращает StepResult.route(key). Условия — не выражения JavaScript; поддерживаются три формы записи, в порядке приоритета:

• Дерево предикатов (extras.predicate) — булево дерево, которое маршрутизирует "true" / "false". Каждый узел — это отображение с полем type:
  • {"type": "fileExists", "paths": [...], "baseKey": "repo_local_path", "negate": false, "recursive": false} — истина, когда любой из перечисленных путей существует на диске; negate инвертирует результат; recursive дополнительно ищет в подкаталогах (пропуская .git, node_modules, build, .dart_tool). Относительные пути разрешаются относительно state[baseKey] (по умолчанию repo_local_path), иначе — относительно каталога рабочего пространства для данного запуска.
  • {"type": "comparison", "left": "{{score}}", "op": "gt", "right": 80} — операторы equals, notEquals, contains, exists, notExists, gt, lt; читает состояние конвейера, а не файловую систему.
  • {"type": "and" | "or", "of": [<predicate>, ...]} — булевы группы.
  • {"type": "not", "of": <predicate>} — отрицание.
• Switch (extras.switchKey) — {"switchKey": "pr_class", "cases": [...], "default": "standard"}; переходит к первому варианту, который значение состояния содержит без учёта регистра, иначе — к default.
• Сравнение (устаревшие поля верхнего уровня extras.left / extras.op / extras.right) — эквивалент предиката comparison, сохранённый для шаблонов, написанных до появления дерева.

Рёбра, выходящие из узла, несут routeKey, который должен совпадать с возвращённым ключом.

prReview.comment

Публикует консолидированные находки как комментарий к ревью PR через клиент GitHub PR.

Ключ состояния	Описание
repo_full_name	owner/repo (обязательно)
pr_number	Номер PR — int или числовая строка (обязательно)
consolidated_findings	Тело комментария (обязательно)

Возвращает StepResult.terminal с comment_review_id и commented_at в состоянии.

messaging.postSpace

Публикует сообщение в пространство обмена сообщениями через MessagingPort.

Ключ состояния	Описание
space_id	Целевое пространство (обязательно)
content	Тело сообщения, читаемое из состояния дословно (обязательно)

Записывает posted_space_id и posted_at в состояние.

team.dispatch

Отправляет целую команду вместо отдельного агента, затем приостанавливается до завершения запусков участников.

Поле конфигурации	Описание
teamId	Команда в пределах рабочего пространства для отправки (обязательно; взаимоисключающе с agentId)
prompt	Текст цели; поддерживает подстановку {{key}} (обязательно)
dispatchMode	allParallel (по умолчанию — одна задача на участника) или manager (отправляется только лидер для координации)
outputKey	Ключ состояния для собранных выводов участников
outputSchema, label, extras.spaceId	Как в conversation.promptAgent

human.gate

Ворота одобрения. Отправляет одобряющего агента в разговор (в режиме ревью) с промптом ворот и контрактом вывода {decision, reason}, затем приостанавливается, пока одобряющий не отправит результат через submit_output.

Поле конфигурации	Описание
agentId	Одобряющий агент (обязательно)
prompt	Промпт ворот (необязательно; при отсутствии используется значение по умолчанию)
outputKey	Ключ состояния для собранного payload {decision, reason}
outputSchema	По умолчанию — схема одобрения (decision: approved / rejected, reason)
label, extras.spaceId	Как в conversation.promptAgent

Ворота завершаются только через submit_output. BuiltInBodyKeys.humanGate, библиотека типов узлов и текст промпта одного встроенного шаблона — все называют MCP-инструменты approve_step и reject_step; ни одного из этих инструментов не существует ни в одном пакете или реестре.

repos.cleanup

Удаляет устаревшие изолированные worktree; режим выбирается из payload триггера. Учитывает флаг dryRun запуска.

Ключ состояния	Описание
ticket_id	Освободить worktree этого тикета
repo_full_name + pr_number	Освободить эфемерный worktree редактора PR
space_id	Снести worktree и папку этого разговора
(ничего из перечисленного)	Прочесать рабочее пространство: исчезнувшие каталоги, мёртвые пространства, папки осиротевших разговоров

outputKey (необязательно) получает строку со сводкой.

flow.forEach

Map / веерный запуск по коллекции состояния: рендерит промпт по одному разу на элемент, затем отправляет агента один раз с этими промптами, соединёнными разделителями ---, и приостанавливается до завершения этого запуска. Пустая коллекция завершается немедленно с пустым списком. Это не по одной параллельной отправке на элемент.

Поле конфигурации	Описание
agentId	Агент для отправки на каждый элемент (обязательно)
prompt	Промпт на элемент; поддерживает подстановку {{key}} (обязательно)
extras.iterableKey	Ключ состояния, хранящий коллекцию (обязательно)
extras.itemKey	Ключ, к которому привязан текущий элемент в промпте (по умолчанию item)
outputKey	Ключ состояния для собранного вывода (одна отправка, не список)

flow.callPipeline

Запускает другой шаблон конвейера как вложенный подшаг (подконвейер выполняется как дочерний запуск с установленным parentPipelineRunId). Шаг приостанавливается, пока дочерний запуск не достигнет терминального состояния, после чего итоговое состояние дочернего запуска сливается под outputKey.

Поле конфигурации	Описание
extras.templateId	Дочерний шаблон для запуска (обязательно; конвейер не может вызвать сам себя)
inputKeys	Ключи состояния, копируемые в payload триггера дочернего запуска
outputKey	Ключ состояния для слитого итогового состояния дочернего запуска

code.index

Фоновое индексирование через tree-sitter: обходит репозиторий, извлекает символы и рёбра в изолятах воркеров и загружает их в граф кода рабочего пространства. Передаёт прогресс в строку запуска шага; поддерживает dry-run. Tree-sitter и его грамматики — обязательные нативные зависимости: отсутствующая dylib означает, что сервер так и не запустился, а грамматика, которая не разрешается во время выполнения, проваливает шаг, а не пропускается.

Ключ состояния	Описание
repo_id	Репозиторий для индексации (обязательно)
repo_local_path	Локальный путь checkout (обязательно)

meeting.diarize

Офлайн-диаризация спикеров. Читает сохранённое аудио встречи, кластеризует его по спикерам (Person 1, Person 2, …), переразмечает сегменты транскрипта и перезаписывает состояние transcript; также выдаёт diarization_spans для параллельного шага meeting.updateTranscript. Пропускает транскрипт без изменений, если аудио не сохранилось или модели не установлены.

Ключ состояния	Описание
meeting_id	Встреча для диаризации (обязательно)

meeting.saveNotes, meeting.addActionItems, meeting.addDecisions

Детерминированные тела сохранения сводки встречи. Каждое читает meeting_id и структурированный payload meeting_outcome агентского шага ({summary, enhancedNotes, actionItems[], decisions[]}) из состояния и записывает свою часть в собственную таблицу — заметки, пункты действий или решения. Ранее сохранённые строки никогда не стираются, если запуск прошёл неполно.

Прочие встроенные тела

Ключ тела	Описание
meeting.identifySpeakers	Межвстречное распознавание спикеров: сопоставляет диаризованных спикеров с сохранёнными голосовыми профилями и автоматически применяет уверенные совпадения. Ничего не делает без профилей и эмбеддингов.
meeting.updateTranscript	Применяет метки Person N из diarization_spans и сливает фрагменты по окнам в реплики спикеров, затем сохраняет очищенный транскрипт. Ничего не делает, если диаризация ничего не дала.
meeting.assemblePlayback	Собирает сведённую дорожку воспроизведения встречи из сохранённых поканальных WAV. Ничего не делает, если аудио не сохранилось.
orchestration.markPhase	В конце рабочего DAG записывает маркеры сбоя для подтикетов без вывода и переключает оркестрацию в synthesizing.
orchestration.persistDeliverable	Записывает вывод синтеза в родительский тикет, завершает его, публикует результат и помечает оркестрацию как completed.
orchestration.awaitApproval	Ворота частичного одобрения: завершаются немедленно, если ключ узла шага входит в одобренное множество оркестрации, иначе приостанавливаются до более позднего одобрения, которое их возобновит. Это единственное тело, задающее extras.awaitApproval, что освобождает его от 24-часового тайм-аута приостановленного шага.
messaging.createSpace	Открывает разговор, в котором работает остальная часть шаблона, публикуя его под outputKey шага (встроенные шаблоны используют pipeline_space_id). Пустой список repoIds ничего не берёт в checkout; extras['allRepos'] берёт в checkout все репозитории рабочего пространства. extras['createConversation'] (по умолчанию выключен) также открывает именованный поток. Агентский шаг ПРИСОЕДИНЯЕТСЯ к комнате и никогда её не открывает, поэтому веерный запуск без этого узла проваливается при отправке — а если подключить его один раз, весь веерный запуск делит один checkout вместо выделения по одному на каждого.
prReview.finalize	Последний шаг шаблона ревью PR: раскладывает накопленные узлы ревью по корзинам консенсуса и вычисляет вердикт по каждому PR.
skills.analyze	Запускает проверку цепочки поставок навыков по установленным в рабочем пространстве навыкам и записывает каждый вердикт.

Результаты шагов

Тело шага возвращает StepResult (packages/cc_domain/lib/features/pipelines/domain/entities/step_result.dart). Все фабрики принимают необязательную карту mutatedState, которая сливается в состояние конвейера, кроме failed.

Фабрика	Сигнатура	Описание
ok	StepResult.ok({mutatedState})	Обычное завершение; нижестоящие слушатели вычисляются
route	StepResult.route(nextRouterKey, {mutatedState})	Завершение роутера; выбирает нижестоящую ветвь, чей routeKey совпадает
suspendUntilEvent	StepResult.suspendUntilEvent(eventType, {mutatedState})	Пауза до срабатывания доменного события указанного типа
suspendUntilTasksComplete	StepResult.suspendUntilTasksComplete(taskIds, {mutatedState})	Пауза, пока все перечисленные задачи не достигнут терминального состояния
terminal	StepResult.terminal({mutatedState})	Запуск конвейера завершён — шагов больше нет
failed	StepResult.failed(errorMessage)	Шаг завершился сбоем с сообщением об ошибке

StepRetryPolicy

Поведение повторов при сбое тела узла (PipelineNodeConfig.retryPolicy; null отключает повторы).

Поле	Тип	Описание
maxAttempts	int	Всего попыток, включая первую (3 = 1 попытка + 2 повтора); должно быть ≥ 1
backoff	String	linear или exponential (по умолчанию)
initialDelayMs	int	Задержка перед первым повтором в миллисекундах; последующие задержки растут согласно backoff

PipelineNodeConfig

Конфигурация отдельного узла, хранимая внутри определения шага (packages/cc_domain/lib/features/pipelines/domain/entities/pipeline_node_config.dart). Встроенные и пользовательские узлы имеют одну и ту же форму; каждое тело читает только нужные ему поля.

Поле	Тип	Описание
prompt	String?	Шаблон промпта с подстановкой {{key}}
script	String?	Тело bash-скрипта с подстановкой {{key}}
agentId	String?	UUID агента в пределах рабочего пространства для отправки
inputKeys	List<String>	Ключи состояния, которые этот узел потребляет на вход
repoIds	List<String>	Репозитории, которые берёт в checkout узел messaging.createSpace (<repoId> или <repoId>@<branch>). Пусто = ничего; extras['allRepos'] = все репозитории рабочего пространства. Игнорируется на агентских шагах.
outputKey	String?	Ключ состояния, под которым записывается вывод этого узла
label	String?	Человекочитаемая метка на холсте (по умолчанию — ID шага)
outputSchema	Map<String, dynamic>?	JSON Schema (подмножество), которой должно соответствовать значение вывода
reducer	String?	Стратегия слияния, когда параллельные ветви пишут в один outputKey: append, mergeLists, mergeMaps, sum или override (по умолчанию)
retryPolicy	StepRetryPolicy?	Политика повторов при сбое тела; null = одна попытка
continueOnFail	bool	Если true, терминальный сбой не проваливает запуск; ошибка складывается в state['_stepErrors'][stepId]
timeoutMs	int?	Тайм-аут тела узла по реальному времени в миллисекундах; null = без тайм-аута
teamId	String?	Команда для отправки (узлы team.dispatch); взаимоисключающе с agentId
dispatchMode	String?	Режим выполнения команды: allParallel или manager
extras	Map<String, dynamic>	Свободная конфигурация, специфичная для тела (например, iterableKey, templateId, деревья условий)

Производные методы доступа: spaceId (extras['spaceId'] — существующий разговор для работы), createsConversation (extras['createConversation']), conversationTitle (extras['conversationTitle']), modeName (extras['mode'] — chat / review / plan, читается на узле, открывающем пространство) и runWhen (extras['runWhen'] — пропустить узел, если условие не выполнено).

См. также

• Конвейеры
• Создать конвейер
• Триггеры конвейера
• Запустить конвейер