Passaggi delle pipeline

Un passaggio di pipeline ha due assi indipendenti:

• Tipo (StepKind) — come il motore pianifica il passaggio rispetto ai suoi trigger.
• Chiave del corpo (BuiltInBodyKeys) — quale closure di corpo registrata esegue il lavoro del passaggio.

Tipi di passaggio

Definito in packages/cc_domain/lib/features/pipelines/domain/entities/step_kind.dart.

Tipo	Semantica di pianificazione
trigger	Punto di ingresso obbligatorio — esattamente uno per modello, sempre il primo nodo. Non svolge alcun lavoro da sé; ciò che avvia l’esecuzione (manuale / evento / pianificazione) viene tracciato separatamente come righe PipelineTrigger. Il suo corpo è il no-op pipeline.trigger.
listen	Si attiva quando tutti i passaggi di origine (in StepTrigger.sourceStepIds) sono completati.
join	Si attiva quando tutti i passaggi in waitForStepIds raggiungono lo stato terminale.
router	Diramazione condizionale — il corpo restituisce StepResult.route(key), che attiva solo l’arco a valle il cui routeKey corrisponde; i rami non selezionati vengono contrassegnati come saltati.
forEach	Pianificato come listen (si attiva quando tutti i sourceStepIds sono completati). È il corpo registrato flow.forEach a iterare; il motore non genera un passaggio figlio per ogni elemento.
terminal	Sentinella dello scheduler senza corpo. L’esecuzione della pipeline termina quando un passaggio terminale finisce.

Corpi dei passaggi

Closure di corpo registrate, indicizzate dalle costanti in BuiltInBodyKeys (packages/cc_domain/lib/features/pipelines/domain/templates/builtin_template_seeds.dart). Sono 25, tutte elencate di seguito. La configurazione viene letta dal PipelineNodeConfig del passaggio (vedi sotto) salvo diversa indicazione; le “chiavi di stato” vengono lette dallo stato della pipeline / dal payload del trigger al momento dell’esecuzione.

Un passaggio la cui bodyKey non è registrata produce un avviso al caricamento del modello e poi non viene mai eseguito — non fa fallire l’esecuzione. I passaggi trigger e terminal sono esenti da quell’avviso: il trigger esegue comunque il corpo no-op pipeline.trigger; un passaggio terminal non ha corpo.

pipeline.trigger

Corpo no-op del nodo trigger obbligatorio — si completa immediatamente, così il motore si dirama verso i listener a valle. Nessuna configurazione.

pipeline.bashScript

Passaggio shell senza agente (bash -c). I segnaposto {{key}} nello script vengono sostituiti a partire dallo stato + dal payload del trigger con valori con escape per la shell.

Campo di configurazione	Descrizione
script	Comando shell da eseguire (obbligatorio)
outputKey	Chiave di stato che riceve lo stdout ripulito in caso di uscita 0

Viene eseguito con cwd <cc_root>/pipelines/<pipelineRunId>/ (condivisa da tutti i passaggi bash di un’esecuzione) e GITHUB_TOKEN nell’ambiente. Scrive inoltre <stepId>_runDir nello stato. Un’uscita diversa da zero fa fallire il passaggio con la coda dello stderr.

conversation.promptAgent

Nodo generico di prompt e invio: renderizza il prompt, invia l’agente in una conversazione e sospende finché l’esecuzione non termina. Il payload submit_output dell’agente viene raccolto nello stato sotto outputKey.

Campo di configurazione	Descrizione
agentId	UUID dell’agente con ambito nello spazio di lavoro da inviare (obbligatorio)
prompt	Testo dell’istruzione; supporta la sostituzione {{key}} (obbligatorio)
outputKey	Chiave di stato per l’output raccolto
outputSchema	JSON Schema che l’output deve soddisfare
label	Etichetta leggibile mostrata sul canvas
extras.spaceId	Conversazione esistente in cui lavorare (predefinita: una nascosta)
extras.conversationTitle	Flusso con nome all’interno di quello spazio (opzionale)
extras.mode	Letto sul nodo che apre la stanza (messaging.createSpace), non qui. Predefinito review.

pipeline.condition

Corpo del router — legge config.extras, valuta una condizione e restituisce StepResult.route(key). Le condizioni non sono espressioni JavaScript; sono supportate tre forme di scrittura, in ordine di priorità:

• Albero di predicati (extras.predicate) — un albero booleano che instrada "true" / "false". Ogni nodo è una map con un type:
  • {"type": "fileExists", "paths": [...], "baseKey": "repo_local_path", "negate": false, "recursive": false} — vero quando uno qualsiasi dei percorsi elencati esiste su disco; negate lo inverte; recursive cerca anche nelle sottodirectory (saltando .git, node_modules, build, .dart_tool). I percorsi relativi vengono risolti rispetto a state[baseKey] (predefinito repo_local_path), altrimenti rispetto alla directory dello spazio di lavoro dell’esecuzione.
  • {"type": "comparison", "left": "{{score}}", "op": "gt", "right": 80} — operatori equals, notEquals, contains, exists, notExists, gt, lt; legge lo stato della pipeline, non il filesystem.
  • {"type": "and" | "or", "of": [<predicate>, ...]} — gruppi booleani.
  • {"type": "not", "of": <predicate>} — negazione.
• Switch (extras.switchKey) — {"switchKey": "pr_class", "cases": [...], "default": "standard"}; instrada verso il primo caso che il valore di stato contiene senza distinguere maiuscole e minuscole, altrimenti verso default.
• Confronto (legacy, a livello superiore extras.left / extras.op / extras.right) — equivalente a un predicato comparison, mantenuto per i modelli creati prima dell’albero.

Gli archi in uscita dal nodo portano un routeKey che deve corrispondere alla chiave restituita.

prReview.comment

Pubblica le segnalazioni consolidate come commento di revisione della PR tramite il client PR di GitHub.

Chiave di stato	Descrizione
repo_full_name	owner/repo (obbligatorio)
pr_number	Numero della PR — int o stringa numerica (obbligatorio)
consolidated_findings	Corpo del commento (obbligatorio)

Restituisce StepResult.terminal con comment_review_id e commented_at nello stato.

messaging.postSpace

Pubblica un messaggio in uno spazio di messaggistica tramite MessagingPort.

Chiave di stato	Descrizione
space_id	Spazio di destinazione (obbligatorio)
content	Corpo del messaggio, letto alla lettera dallo stato (obbligatorio)

Scrive posted_space_id e posted_at nello stato.

team.dispatch

Invia un intero team invece di un singolo agente, poi sospende finché le esecuzioni dei membri non terminano.

Campo di configurazione	Descrizione
teamId	Team con ambito nello spazio di lavoro da inviare (obbligatorio; mutuamente esclusivo con agentId)
prompt	Testo dell’obiettivo; supporta la sostituzione {{key}} (obbligatorio)
dispatchMode	allParallel (predefinito — un’attività per membro) oppure manager (invia solo il leader a coordinare)
outputKey	Chiave di stato per gli output raccolti dei membri
outputSchema, label, extras.spaceId	Come per conversation.promptAgent

human.gate

Cancello di approvazione. Invia l’agente approvatore in una conversazione (modalità review) con il prompt del cancello e un contratto di output {decision, reason}, poi sospende finché l’approvatore non invia tramite submit_output.

Campo di configurazione	Descrizione
agentId	L’agente approvatore (obbligatorio)
prompt	Prompt del cancello (opzionale; se assente ne viene usato uno predefinito)
outputKey	Chiave di stato per il payload {decision, reason} raccolto
outputSchema	Predefinito a uno schema di approvazione (decision: approved / rejected, reason)
label, extras.spaceId	Come per conversation.promptAgent

Il cancello si risolve solo tramite submit_output. BuiltInBodyKeys.humanGate, la libreria dei tipi di nodo e il testo del prompt di un modello integrato nominano tutti gli strumenti MCP approve_step e reject_step; nessuno dei due strumenti esiste in alcun pacchetto o registro.

repos.cleanup

Rimuove i worktree isolati obsoleti; sceglie la propria modalità dal payload del trigger. Rispetta il flag dryRun dell’esecuzione.

Chiave di stato	Descrizione
ticket_id	Rilascia i worktree di quel ticket
repo_full_name + pr_number	Rilascia il worktree effimero dell’editor di PR
space_id	Smantella i worktree e la cartella di quella conversazione
(nessuna delle precedenti)	Pulisci lo spazio di lavoro: directory scomparse, spazi morti, cartelle di conversazioni orfane

outputKey (opzionale) riceve una stringa di riepilogo.

flow.forEach

Map / fan-out su una collezione dello stato: renderizza il prompt una volta per elemento, poi invia l’agente una sola volta con quei prompt concatenati (separatori ---) e sospende finché quell’esecuzione non termina. Una collezione vuota si completa immediatamente con una lista vuota. Non è un invio parallelo per elemento.

Campo di configurazione	Descrizione
agentId	Agente da inviare per ogni elemento (obbligatorio)
prompt	Prompt per elemento; supporta la sostituzione {{key}} (obbligatorio)
extras.iterableKey	Chiave di stato che contiene la collezione (obbligatorio)
extras.itemKey	Chiave a cui è associato l’elemento corrente nel prompt (predefinito item)
outputKey	Chiave di stato per l’output raccolto (un solo invio, non una lista)

flow.callPipeline

Esegue un altro modello di pipeline come sotto-passaggio annidato (la sotto-pipeline viene eseguita come esecuzione figlia con parentPipelineRunId impostato). Il passaggio sospende finché il figlio non raggiunge uno stato terminale, poi lo stato finale del figlio viene unito sotto outputKey.

Campo di configurazione	Descrizione
extras.templateId	Modello figlio da eseguire (obbligatorio; una pipeline non può chiamare se stessa)
inputKeys	Chiavi di stato copiate nel payload del trigger del figlio
outputKey	Chiave di stato per lo stato finale unito del figlio

code.index

Indicizzazione tree-sitter in background: percorre un repo, estrae simboli/archi in isolate worker e li inserisce nel grafo del codice dello spazio di lavoro. Trasmette l’avanzamento in streaming nella riga di esecuzione del passaggio; supporta il dry-run. Tree-sitter e le sue grammatiche sono native richieste — una dylib mancante significa che il server non è mai stato avviato, e una grammatica che non si risolve a metà esecuzione fa fallire il passaggio invece di saltarlo.

Chiave di stato	Descrizione
repo_id	Repo da indicizzare (obbligatorio)
repo_local_path	Percorso locale del checkout (obbligatorio)

meeting.diarize

Diarizzazione offline dei parlanti. Legge l’audio conservato della riunione, lo raggruppa in parlanti (Person 1, Person 2, …), rietichetta i segmenti della trascrizione e riscrive lo stato transcript; emette anche diarization_spans per il passaggio parallelo meeting.updateTranscript. Restituisce la trascrizione invariata quando non è stato conservato alcun audio o i modelli non sono installati.

Chiave di stato	Descrizione
meeting_id	Riunione da diarizzare (obbligatorio)

meeting.saveNotes, meeting.addActionItems, meeting.addDecisions

Corpi deterministici di persistenza del riepilogo della riunione. Ognuno legge meeting_id e il payload strutturato meeting_outcome del passaggio agente ({summary, enhancedNotes, actionItems[], decisions[]}) dallo stato e scrive la propria parte nella propria tabella — note, attività da fare o decisioni. Le righe salvate in precedenza non vengono mai cancellate in un’esecuzione degradata.

Altri corpi integrati

Chiave del corpo	Descrizione
meeting.identifySpeakers	Riconoscimento dei parlanti tra riunioni diverse: confronta i parlanti diarizzati con i profili vocali salvati e applica automaticamente le corrispondenze sicure. No-op senza profili/embedding.
meeting.updateTranscript	Applica le etichette Person N da diarization_spans e unisce i frammenti per finestra in turni per parlante, poi persiste la trascrizione ripulita. No-op quando la diarizzazione non ha prodotto nulla.
meeting.assemblePlayback	Assembla la traccia di riproduzione mixata della riunione dai WAV per canale conservati. No-op quando non è stato conservato alcun audio.
orchestration.markPhase	Alla fine del DAG di lavoro, scrive sentinelle di errore per i sotto-ticket senza output e porta l’orchestrazione a synthesizing.
orchestration.persistDeliverable	Scrive l’output di sintesi nel ticket padre, lo completa, pubblica il deliverable e contrassegna l’orchestrazione come completed.
orchestration.awaitApproval	Cancello di approvazione parziale: si completa immediatamente quando la chiave del nodo del passaggio è nell’insieme approvato dell’orchestrazione, altrimenti sospende finché un’approvazione successiva non lo riprende. È l’unico corpo che imposta extras.awaitApproval, il che lo esenta dal timeout di 24 ore dei passaggi sospesi.
messaging.createSpace	Apre la conversazione in cui lavora il resto del modello, pubblicata sotto l’outputKey del passaggio (i corpi integrati usano pipeline_space_id). Una lista repoIds vuota non fa il checkout di nulla; extras['allRepos'] fa il checkout di ogni repo dello spazio di lavoro. extras['createConversation'] (disattivato per impostazione predefinita) apre anche un flusso con nome. Un passaggio agente ADERISCE a una stanza e non ne apre mai una, quindi un fan-out senza questo nodo fallisce all’invio — e collegarlo una volta significa che l’intero fan-out condivide un unico checkout invece di procurarne uno per ciascuno.
prReview.finalize	L’ultimo passaggio del modello di revisione delle PR: ordina i nodi di revisione accumulati in bucket di consenso e calcola il verdetto per PR.
skills.analyze	Esegue la scansione della supply chain delle skill sulle skill installate di uno spazio di lavoro e registra ogni verdetto.

Risultati dei passaggi

Il corpo di un passaggio restituisce un StepResult (packages/cc_domain/lib/features/pipelines/domain/entities/step_result.dart). Tutte le factory accettano una map opzionale mutatedState unita allo stato della pipeline, tranne failed.

Factory	Firma	Descrizione
ok	StepResult.ok({mutatedState})	Completamento normale; i listener a valle vengono valutati
route	StepResult.route(nextRouterKey, {mutatedState})	Completamento del router; seleziona il ramo a valle il cui routeKey corrisponde
suspendUntilEvent	StepResult.suspendUntilEvent(eventType, {mutatedState})	Sospende finché non si attiva un tipo specifico di evento di dominio
suspendUntilTasksComplete	StepResult.suspendUntilTasksComplete(taskIds, {mutatedState})	Sospende finché tutte le attività elencate non raggiungono lo stato terminale
terminal	StepResult.terminal({mutatedState})	L’esecuzione della pipeline è terminata — nessun altro passaggio
failed	StepResult.failed(errorMessage)	Il passaggio è fallito con un messaggio di errore

StepRetryPolicy

Comportamento di retry per il corpo di un nodo che fallisce (PipelineNodeConfig.retryPolicy; null disabilita i retry).

Campo	Tipo	Descrizione
maxAttempts	int	Tentativi totali, incluso il primo (3 = 1 tentativo + 2 retry); deve essere ≥ 1
backoff	String	linear o exponential (predefinito)
initialDelayMs	int	Ritardo prima del primo retry, in millisecondi; i ritardi successivi crescono in base a backoff

PipelineNodeConfig

Configurazione per nodo contenuta nella definizione del passaggio (packages/cc_domain/lib/features/pipelines/domain/entities/pipeline_node_config.dart). I nodi integrati e quelli personalizzati condividono questa forma; ogni corpo legge solo i campi che gli interessano.

Campo	Tipo	Descrizione
prompt	String?	Modello di prompt con sostituzione {{key}}
script	String?	Corpo dello script Bash con sostituzione {{key}}
agentId	String?	UUID dell’agente con ambito nello spazio di lavoro da inviare
inputKeys	List<String>	Chiavi di stato che questo nodo consuma come input
repoIds	List<String>	Repo di cui il nodo messaging.createSpace fa il checkout (<repoId> o <repoId>@<branch>). Vuoto = nessuno; extras['allRepos'] = ogni repo dello spazio di lavoro. Ignorato sui passaggi agente.
outputKey	String?	Chiave di stato sotto cui viene scritto l’output di questo nodo
label	String?	Etichetta leggibile mostrata sul canvas (predefinita: l’ID del passaggio)
outputSchema	Map<String, dynamic>?	JSON Schema (sottoinsieme) che il valore di output deve soddisfare
reducer	String?	Strategia di unione quando rami paralleli scrivono lo stesso outputKey: append, mergeLists, mergeMaps, sum o override (predefinito)
retryPolicy	StepRetryPolicy?	Politica di retry in caso di fallimento del corpo; null = un solo tentativo
continueOnFail	bool	Se vero, un fallimento terminale non fa fallire l’esecuzione; l’errore viene archiviato sotto state['_stepErrors'][stepId]
timeoutMs	int?	Timeout in tempo reale per il corpo del nodo, in millisecondi; null = nessun timeout
teamId	String?	Team da inviare (nodi team.dispatch); mutuamente esclusivo con agentId
dispatchMode	String?	Modalità di esecuzione del team: allParallel o manager
extras	Map<String, dynamic>	Configurazione libera specifica del corpo (ad es. iterableKey, templateId, alberi di condizioni)

Accessor derivati: spaceId (extras['spaceId'] — conversazione esistente in cui lavorare), createsConversation (extras['createConversation']), conversationTitle (extras['conversationTitle']), modeName (extras['mode'] — chat / review / plan, letto sul nodo che apre lo spazio) e runWhen (extras['runWhen'] — salta il nodo quando non è soddisfatto).

Vedi anche

• Pipeline
• Crea una pipeline
• Trigger delle pipeline
• Esegui una pipeline