Tipos de pasos de pipeline

Un paso de pipeline tiene dos ejes independientes:

• Tipo (StepKind): cómo programa el motor el paso en relación con sus disparadores.
• Clave de cuerpo (BuiltInBodyKeys): qué clausura de cuerpo registrada ejecuta el trabajo del paso.

Tipos de pasos

Se definen en packages/cc_domain/lib/features/pipelines/domain/entities/step_kind.dart.

Tipo	Semántica de programación
trigger	Punto de entrada obligatorio: exactamente uno por plantilla, siempre el primer nodo. No hace trabajo por sí mismo; lo que inicia la ejecución (manual / evento / programación) se registra por separado como filas PipelineTrigger. Su cuerpo es el no-op pipeline.trigger.
listen	Se activa cuando todos los pasos de origen (en StepTrigger.sourceStepIds) completan.
join	Se activa cuando todos los pasos de waitForStepIds alcanzan un estado terminal.
router	Bifurcación condicional: el cuerpo devuelve StepResult.route(key), que solo activa la arista descendente cuyo routeKey coincide; las ramas no seleccionadas se marcan como omitidas.
forEach	Se programa como listen (se activa cuando todos los sourceStepIds completan). El cuerpo registrado flow.forEach es el que itera; el motor no genera un paso hijo por cada elemento.
terminal	Centinela del planificador sin cuerpo. La ejecución del pipeline termina cuando finaliza un paso terminal.

Cuerpos de paso

Clausuras de cuerpo registradas bajo las constantes de BuiltInBodyKeys (packages/cc_domain/lib/features/pipelines/domain/templates/builtin_template_seeds.dart). Hay 25, todas listadas a continuación. La configuración se lee de PipelineNodeConfig del paso (ver más abajo) salvo que se indique lo contrario; las «claves de estado» se leen del estado del pipeline o del payload del disparador en el momento de la ejecución.

Un paso cuyo bodyKey no está registrado produce una advertencia al cargar la plantilla y después nunca se ejecuta, pero no hace fallar la ejecución. Los pasos trigger y terminal están exentos de esa advertencia: el trigger igualmente ejecuta el cuerpo no-op pipeline.trigger; un paso terminal no tiene cuerpo.

pipeline.trigger

Cuerpo no-op del nodo trigger obligatorio: completa de inmediato para que el motor se ramifique en abanico hacia los pasos de escucha posteriores. Sin configuración.

pipeline.bashScript

Paso de shell sin agente (bash -c). Los marcadores {{key}} del script se sustituyen desde el estado y el payload del disparador con valores escapados para el shell.

Campo de configuración	Descripción
script	Comando de shell que se ejecutará (obligatorio)
outputKey	Clave de estado que recibe la salida estándar recortada cuando el código de salida es 0

Se ejecuta con el directorio de trabajo <cc_root>/pipelines/<pipelineRunId>/ (compartido por todos los pasos de bash de una ejecución) y GITHUB_TOKEN en el entorno. También escribe <stepId>_runDir en el estado. Una salida distinta de cero hace fallar el paso con la cola de la salida de error estándar.

conversation.promptAgent

Nodo genérico de prompt y despacho: renderiza el prompt, despacha al agente a una conversación y suspende hasta que la ejecución termina. El payload de submit_output del agente se recoge en el estado bajo outputKey.

Campo de configuración	Descripción
agentId	UUID del agente del espacio de trabajo que se despachará (obligatorio)
prompt	Texto de instrucción; admite sustitución {{key}} (obligatorio)
outputKey	Clave de estado para la salida recogida
outputSchema	Esquema JSON que debe cumplir la salida
label	Etiqueta legible para las personas que se muestra en el lienzo
extras.spaceId	Conversación existente en la que trabajar (predeterminado: una oculta)
extras.conversationTitle	Flujo con nombre dentro de ese espacio (opcional)
extras.mode	Se lee en el nodo que abre la sala (messaging.createSpace), no aquí. Predeterminado review.

pipeline.condition

Cuerpo de router: lee config.extras, evalúa una condición y devuelve StepResult.route(key). Las condiciones no son expresiones de JavaScript; se admiten tres formas de definirlas, en orden de prioridad:

• Árbol de predicados (extras.predicate): un árbol booleano que enruta "true" / "false". Cada nodo es un mapa con un type:
  • {"type": "fileExists", "paths": [...], "baseKey": "repo_local_path", "negate": false, "recursive": false}: verdadero cuando cualquiera de las rutas listadas existe en el disco; negate lo invierte; recursive también busca en subdirectorios (omitiendo .git, node_modules, build, .dart_tool). Las rutas relativas se resuelven contra state[baseKey] (predeterminado repo_local_path) y, si no, contra el directorio de trabajo por ejecución.
  • {"type": "comparison", "left": "{{score}}", "op": "gt", "right": 80}: operadores equals, notEquals, contains, exists, notExists, gt, lt; lee el estado del pipeline, no el sistema de archivos.
  • {"type": "and" | "or", "of": [<predicate>, ...]}: grupos booleanos.
  • {"type": "not", "of": <predicate>}: negación.
• Switch (extras.switchKey): {"switchKey": "pr_class", "cases": [...], "default": "standard"}; enruta al primer caso que el valor de estado contiene sin distinguir mayúsculas de minúsculas y, si no, a default.
• Comparación (los campos heredados de nivel superior extras.left / extras.op / extras.right): equivalente a un predicado comparison, se conserva para plantillas creadas antes que el árbol.

Las aristas que salen del nodo llevan un routeKey que debe coincidir con la clave devuelta.

prReview.comment

Publica los hallazgos consolidados como un comentario de revisión de PR a través del cliente de PR de GitHub.

Clave de estado	Descripción
repo_full_name	owner/repo (obligatorio)
pr_number	Número de la PR: entero o cadena numérica (obligatorio)
consolidated_findings	Cuerpo del comentario (obligatorio)

Devuelve StepResult.terminal con comment_review_id y commented_at en el estado.

messaging.postSpace

Publica un mensaje en un espacio de mensajería a través de MessagingPort.

Clave de estado	Descripción
space_id	Espacio de destino (obligatorio)
content	Cuerpo del mensaje, leído literalmente del estado (obligatorio)

Escribe posted_space_id y posted_at en el estado.

team.dispatch

Despacha un equipo completo en lugar de un solo agente y luego suspende hasta que terminan las ejecuciones de sus miembros.

Campo de configuración	Descripción
teamId	Equipo del espacio de trabajo que se despachará (obligatorio; mutuamente excluyente con agentId)
prompt	Texto del objetivo; admite sustitución {{key}} (obligatorio)
dispatchMode	allParallel (predeterminado: una tarea por miembro) o manager (despacha solo al líder para que coordine)
outputKey	Clave de estado para las salidas recogidas de los miembros
outputSchema, label, extras.spaceId	Igual que para conversation.promptAgent

human.gate

Puerta de aprobación. Despacha al agente aprobador a una conversación (modo review) con el prompt de la puerta y un contrato de salida {decision, reason}, y luego suspende hasta que el aprobador envía su resultado mediante submit_output.

Campo de configuración	Descripción
agentId	El agente aprobador (obligatorio)
prompt	Prompt de la puerta (opcional; se usa uno predeterminado cuando falta)
outputKey	Clave de estado para el payload {decision, reason} recogido
outputSchema	De forma predeterminada usa un esquema de aprobación (decision: approved / rejected, reason)
label, extras.spaceId	Igual que para conversation.promptAgent

La puerta se resuelve solo mediante submit_output. BuiltInBodyKeys.humanGate, la biblioteca de tipos de nodo y el texto de prompt de una plantilla integrada mencionan las herramientas MCP approve_step y reject_step; ninguna de las dos herramientas existe en ningún paquete ni registro.

repos.cleanup

Elimina worktrees aislados obsoletos; elige su modo a partir del payload del disparador. Respeta el indicador dryRun de la ejecución.

Clave de estado	Descripción
ticket_id	Libera los worktrees de ese ticket
repo_full_name + pr_number	Libera el worktree efímero del editor de PR
space_id	Desmonta los worktrees y la carpeta de esa conversación
(ninguna de las anteriores)	Barre el espacio de trabajo: directorios desaparecidos, espacios muertos y carpetas de conversación huérfanas

outputKey (opcional) recibe una cadena de resumen.

flow.forEach

Mapeo / reparto en abanico sobre una colección de estado: renderiza el prompt una vez por cada elemento y luego despacha al agente una sola vez con esos prompts concatenados (separadores ---), y suspende hasta que esa ejecución termina. Una colección vacía completa de inmediato con una lista vacía. No es un despacho paralelo por cada elemento.

Campo de configuración	Descripción
agentId	Agente que se despachará por elemento (obligatorio)
prompt	Prompt por elemento; admite sustitución {{key}} (obligatorio)
extras.iterableKey	Clave de estado que contiene la colección (obligatorio)
extras.itemKey	Clave a la que se enlaza el elemento actual en el prompt (predeterminado item)
outputKey	Clave de estado para la salida recogida (un solo despacho, no una lista)

flow.callPipeline

Ejecuta otra plantilla de pipeline como un subpaso anidado (el subpipeline se ejecuta como una ejecución hija con parentPipelineRunId establecido). El paso suspende hasta que el hijo alcanza un estado terminal; luego el estado final del hijo se fusiona bajo outputKey.

Campo de configuración	Descripción
extras.templateId	Plantilla hija que se ejecutará (obligatorio; un pipeline no puede llamarse a sí mismo)
inputKeys	Claves de estado que se copian en el payload del disparador del hijo
outputKey	Clave de estado para el estado final fusionado del hijo

code.index

Indexación en segundo plano con tree-sitter: recorre un repo, extrae símbolos y aristas en isolates de worker y los ingiere en el grafo de código del espacio de trabajo. Transmite el progreso a la fila de ejecución del paso; admite dry-run. Tree-sitter y sus gramáticas son nativos obligatorios: si falta una dylib, el servidor nunca arrancó, y una gramática que no se resuelve a mitad de la ejecución hace fallar el paso en lugar de omitirlo.

Clave de estado	Descripción
repo_id	Repo que se indexará (obligatorio)
repo_local_path	Ruta local del checkout (obligatorio)

meeting.diarize

Diarización de hablantes sin conexión. Lee el audio retenido de la reunión, lo agrupa en hablantes (Person 1, Person 2, …), reetiqueta los segmentos de la transcripción y reescribe el estado transcript; también emite diarization_spans para el paso paralelo meeting.updateTranscript. Deja pasar la transcripción sin cambios cuando no se retuvo audio o los modelos no están instalados.

Clave de estado	Descripción
meeting_id	Reunión que se diarizará (obligatorio)

meeting.saveNotes, meeting.addActionItems, meeting.addDecisions

Cuerpos deterministas que persisten el resumen de una reunión. Cada uno lee meeting_id y el payload estructurado meeting_outcome del paso de agente ({summary, enhancedNotes, actionItems[], decisions[]}) del estado y escribe su parte en su propia tabla: notas, elementos de acción o decisiones. Las filas guardadas antes nunca se borran en una ejecución degradada.

Otros cuerpos integrados

Clave de cuerpo	Descripción
meeting.identifySpeakers	Reconocimiento de hablantes entre reuniones: compara los hablantes diarizados con los perfiles de voz guardados y aplica automáticamente las coincidencias confiables. No-op sin perfiles ni embeddings.
meeting.updateTranscript	Aplica las etiquetas Person N de diarization_spans y fusiona los fragmentos por ventana en turnos por hablante; luego persiste la transcripción limpia. No-op cuando la diarización no produjo nada.
meeting.assemblePlayback	Ensambla la pista de reproducción mezclada de la reunión a partir de los WAV por canal retenidos. No-op cuando no se retuvo audio.
orchestration.markPhase	Al final del DAG de trabajo, escribe centinelas de fallo para los subtickets sin salida y cambia la orquestación a synthesizing.
orchestration.persistDeliverable	Escribe la salida de la síntesis en el ticket principal, lo completa, publica el entregable y marca la orquestación como completed.
orchestration.awaitApproval	Puerta de aprobación parcial: completa de inmediato cuando la clave de nodo del paso está en el conjunto aprobado de la orquestación; de lo contrario, suspende hasta que una aprobación posterior la reanude. Este es el único cuerpo que establece extras.awaitApproval, lo que lo exime del tiempo de espera de 24 horas de los pasos suspendidos.
messaging.createSpace	Abre la conversación en la que trabaja el resto de la plantilla, publicada bajo el outputKey del paso (los integrados usan pipeline_space_id). Una lista repoIds vacía no hace checkout de nada; extras['allRepos'] hace checkout de todos los repos del espacio de trabajo. extras['createConversation'] (desactivado de forma predeterminada) también abre un flujo con nombre. Un paso de agente SE UNE a una sala y nunca abre una, así que un reparto en abanico sin este nodo falla en el despacho; además, cablearlo una sola vez hace que todo el reparto en abanico comparta un único checkout en lugar de aprovisionar uno para cada uno.
prReview.finalize	El último paso de la plantilla de revisión de PR: ordena los nodos de revisión acumulados en grupos de consenso y calcula el veredicto por PR.
skills.analyze	Ejecuta el análisis de cadena de suministro de skills sobre las skills instaladas en un espacio de trabajo y registra cada veredicto.

Resultados de pasos

Un cuerpo de paso devuelve un StepResult (packages/cc_domain/lib/features/pipelines/domain/entities/step_result.dart). Todas las fábricas aceptan un mapa mutatedState opcional que se fusiona en el estado del pipeline, excepto failed.

Fábrica	Firma	Descripción
ok	StepResult.ok({mutatedState})	Finalización normal; se evalúan los pasos de escucha posteriores
route	StepResult.route(nextRouterKey, {mutatedState})	Finalización del router; selecciona la rama descendente cuyo routeKey coincide
suspendUntilEvent	StepResult.suspendUntilEvent(eventType, {mutatedState})	Pausa hasta que se dispara un tipo específico de evento de dominio
suspendUntilTasksComplete	StepResult.suspendUntilTasksComplete(taskIds, {mutatedState})	Pausa hasta que todas las tareas listadas alcanzan un estado terminal
terminal	StepResult.terminal({mutatedState})	La ejecución del pipeline ha terminado: no hay más pasos
failed	StepResult.failed(errorMessage)	El paso falló con un mensaje de error

StepRetryPolicy

Comportamiento de reintento para un cuerpo de nodo que falla (PipelineNodeConfig.retryPolicy; null desactiva los reintentos).

Campo	Tipo	Descripción
maxAttempts	int	Total de intentos incluido el primero (3 = 1 intento + 2 reintentos); debe ser ≥ 1
backoff	String	linear o exponential (predeterminado)
initialDelayMs	int	Retraso antes del primer reintento, en milisegundos; los retrasos posteriores crecen según backoff

PipelineNodeConfig

Configuración por nodo que se transporta dentro de la definición del paso (packages/cc_domain/lib/features/pipelines/domain/entities/pipeline_node_config.dart). Tanto los nodos integrados como los personalizados comparten esta forma; cada cuerpo lee solo los campos que le interesan.

Campo	Tipo	Descripción
prompt	String?	Plantilla de prompt con sustitución {{key}}
script	String?	Cuerpo del script de bash con sustitución {{key}}
agentId	String?	UUID del agente del espacio de trabajo que se despachará
inputKeys	List<String>	Claves de estado que este nodo consume como entrada
repoIds	List<String>	Repos de los que el nodo messaging.createSpace hace checkout (<repoId> o <repoId>@<branch>). Vacío = ninguno; extras['allRepos'] = todos los repos del espacio de trabajo. Se ignora en los pasos de agente.
outputKey	String?	Clave de estado bajo la que se escribe la salida de este nodo
label	String?	Etiqueta legible para las personas que se muestra en el lienzo (de forma predeterminada, el id del paso)
outputSchema	Map<String, dynamic>?	Esquema JSON (subconjunto) que debe cumplir el valor de salida
reducer	String?	Estrategia de fusión cuando ramas paralelas escriben el mismo outputKey: append, mergeLists, mergeMaps, sum u override (predeterminado)
retryPolicy	StepRetryPolicy?	Política de reintentos cuando falla el cuerpo; null = un solo intento
continueOnFail	bool	Cuando es verdadero, un fallo terminal no hace fallar la ejecución; el error se guarda en state['_stepErrors'][stepId]
timeoutMs	int?	Tiempo de espera de reloj de pared para el cuerpo del nodo, en milisegundos; null = sin tiempo de espera
teamId	String?	Equipo que se despachará (nodos team.dispatch); mutuamente excluyente con agentId
dispatchMode	String?	Modo de ejecución del equipo: allParallel o manager
extras	Map<String, dynamic>	Configuración libre específica del cuerpo (p. ej. iterableKey, templateId, árboles de condiciones)

Accesores derivados: spaceId (extras['spaceId']: conversación existente en la que trabajar), createsConversation (extras['createConversation']), conversationTitle (extras['conversationTitle']), modeName (extras['mode']: chat / review / plan, se lee en el nodo que abre el espacio) y runWhen (extras['runWhen']: omite el nodo cuando no se cumple).

Ver también

• Pipelines
• Crear un pipeline
• Disparadores de pipeline
• Ejecutar un pipeline