Pipeline Mode — Workflows YAML Multi-Step

El Pipeline Mode ejecuta workflows definidos en YAML con pasos secuenciales. Cada paso puede tener su propio agente, modelo, prompt, checks, condiciones y variables.

---

Concepto

Un pipeline define una secuencia de tareas donde cada paso puede depender de los anteriores. Los pasos se comunican entre sí mediante variables ({{nombre}}) y se pueden condicionar, verificar con checks, y proteger con checkpoints git.

name: feature-pipeline
steps:
  - name: analyze
    agent: plan
    prompt: "Analiza los requisitos de {{feature}}"
    output_var: analysis

  - name: implement
    agent: build
    prompt: "Implementa según este análisis: {{analysis}}"
    checks:
      - "pytest tests/ -q"
    checkpoint: true

  - name: review
    agent: review
    prompt: "Revisa la implementación de {{feature}}"
    condition: "run_review == 'true'"

---

Uso básico

# Ejecutar un pipeline
architect pipeline workflow.yaml --var feature="user auth"

# Ver qué haría sin ejecutar (dry-run)
architect pipeline workflow.yaml --var feature="user auth" --dry-run

# Reanudar desde un paso específico
architect pipeline workflow.yaml --var feature="user auth" --from-step implement

# Con variables múltiples
architect pipeline workflow.yaml \
  --var feature="payment gateway" \
  --var env=staging \
  --var run_review=true

---

Opciones del comando

Opción	Default	Descripción
PIPELINE_FILE	(requerido)	Archivo YAML con la definición del pipeline
--var KEY=VALUE	—	Variable para el pipeline (repetible)
--from-step NAME	—	Reanudar desde un step específico (salta anteriores)
--dry-run	false	Mostrar plan sin ejecutar agentes
-c, --config PATH	—	Archivo de configuración YAML de architect
--quiet	false	Solo resultado final

---

Formato YAML del pipeline

Estructura completa

name: mi-pipeline                    # Nombre identificativo
variables:                           # Variables iniciales (opcional)
  key: value
steps:
  - name: step-id                    # Identificador único del paso
    agent: build                     # Agente: build, plan, review, resume, o custom
    prompt: "Prompt con {{var}}"     # Prompt con sustitución de variables
    model: gpt-4o                    # Modelo LLM (opcional, override)
    condition: "var == 'true'"       # Condición para ejecutar (opcional)
    output_var: result               # Guardar output como variable (opcional)
    checks:                          # Comandos de verificación post-step (opcional)
      - "pytest tests/"
      - "ruff check src/"
    checkpoint: true                 # Crear git checkpoint (opcional)
    timeout: 300                     # Timeout en segundos (opcional)

Campos de cada step

Campo	Tipo	Default	Descripción
name	str	(requerido)	Identificador del paso
agent	str	"build"	Agente a usar
prompt	str	""	Prompt con soporte para {{variables}}
model	str|null	null	Modelo LLM (null = usar default del config)
condition	str|null	null	Expresión condicional. Si evalúa a falsy, el step se salta
output_var	str|null	null	Nombre de variable donde guardar el output del agente
checks	list[str]	[]	Comandos shell post-step (exit 0 = pass)
checkpoint	bool	false	Crear git checkpoint al completar el step
timeout	int|null	null	Timeout en segundos

---

Características

Variables ({{nombre}})

Las variables se sustituyen en los prompts antes de la ejecución. Se definen en tres fuentes (menor a mayor prioridad):

1. Sección variables del YAML
2. Flag --var KEY=VALUE del CLI (sobreescribe YAML)
3. output_var de steps anteriores (se añaden dinámicamente)

name: var-demo
variables:
  project: myapp
  lang: python
steps:
  - name: analyze
    agent: plan
    prompt: "Analiza el proyecto {{project}} escrito en {{lang}}"
    output_var: analysis

  - name: implement
    agent: build
    prompt: |
      Implementa las mejoras sugeridas:
      {{analysis}}

La sustitución usa regex \{\{(.+?)\}\}. Variables no definidas se dejan como {{nombre}} (no se eliminan).

Condiciones (condition)

Un step con condition se evalúa antes de ejecutarse. Si la condición es falsa, el step se salta con status "skipped".

La evaluación es simple:

• "true", "yes", "1" → True
• "false", "no", "0", "" → False
• Cualquier otro string no vacío → True

Las variables se resuelven en la condición antes de evaluar:

steps:
  - name: setup
    prompt: "..."

  - name: deploy
    prompt: "Deploy a producción"
    condition: "deploy_enabled == 'true'"
    # Si --var deploy_enabled=true → se ejecuta
    # Si --var deploy_enabled=false → se salta

Output variables (output_var)

Captura la salida final del agente y la almacena como variable para steps posteriores:

steps:
  - name: analyze
    agent: plan
    prompt: "Analiza el código y lista las 3 mejoras más importantes"
    output_var: improvements

  - name: implement
    agent: build
    prompt: "Implementa estas mejoras: {{improvements}}"

El valor capturado es el final_output del AgentState — el texto que el agente produce como respuesta final.

Checks

Los checks son comandos shell que se ejecutan después de cada step:

steps:
  - name: implement
    prompt: "Implementa la feature"
    checks:
      - "pytest tests/ -q"
      - "ruff check src/"

• Cada check se ejecuta como subprocess.run(cmd, shell=True, timeout=120)
• Exit 0 = check pasó
• El resultado se almacena en PipelineStepResult.checks_passed
• Los checks no bloquean la ejecución del pipeline — el siguiente step se ejecuta igualmente

Checkpoints

Con checkpoint: true, se crea un git commit automático al completar el step:

steps:
  - name: implement
    prompt: "Implementa la feature"
    checkpoint: true
    # → git add -A && git commit -m "architect:checkpoint:implement"

El commit usa el prefijo architect:checkpoint:<step_name>. Esto permite:

• Ver qué cambió en cada step: git log --oneline --grep="architect:checkpoint"
• Rollback a un step específico con CheckpointManager.rollback()

Dry-run

Con --dry-run, el pipeline muestra el plan sin ejecutar agentes:

architect pipeline workflow.yaml --var feature="auth" --dry-run

Output:

Pipeline: feature-pipeline
  Step 1: analyze (plan) — "Analiza los requisitos de auth"
  Step 2: implement (build) — "Implementa según este análisis: {{analysis}}"
    Checks: pytest tests/ -q, ruff check src/
    Checkpoint: sí
  Step 3: review (review) — "Revisa la implementación de auth"
    Condition: run_review == 'true'

From-step (resume)

Con --from-step, el pipeline salta los steps anteriores y empieza desde el indicado:

# El step "analyze" ya se ejecutó. Reanudar desde "implement"
architect pipeline workflow.yaml --from-step implement

---

Validación del YAML (v1.1.0)

Antes de ejecutar, el pipeline valida el YAML completamente. Si hay errores, se muestra un mensaje descriptivo y se sale con exit code 3 (CONFIG_ERROR) sin ejecutar ningún step.

Reglas de validación

Regla	Descripción
prompt requerido	Cada step debe tener un campo prompt no vacío
Campos válidos	Solo se permiten: name, agent, prompt, model, checkpoint, condition, output_var, checks, timeout
Campos desconocidos	Se rechazan con error. El campo task incluye hint: “¿quisiste decir prompt?”
Al menos 1 step	El pipeline debe tener al menos un step definido
Formato de step	Cada step debe ser un objeto YAML (dict), no un string u otro tipo
Prompt no whitespace-only	Un prompt con solo espacios o saltos de línea se rechaza

Ejemplo de error

# pipeline-malo.yaml
name: bad-pipeline
steps:
  - name: analyze
    task: "Analiza el proyecto"    # ← campo incorrecto
  - name: implement
    prompt: ""                      # ← prompt vacío
  - name: deploy
    prompt: "Deploy"
    priority: high                  # ← campo desconocido

$ architect pipeline pipeline-malo.yaml
Error de validación: Pipeline 'pipeline-malo.yaml' tiene errores de validación:
  analyze: campo desconocido 'task' (¿quisiste decir 'prompt'?)
  analyze: falta 'prompt' (el campo 'task' no es válido, usa 'prompt')
  implement: falta 'prompt' o está vacío
  deploy: campo desconocido 'priority'

Todos los errores se recopilan y se muestran juntos en un solo mensaje, facilitando la corrección.

API: PipelineValidationError

from architect.features.pipelines import PipelineValidationError

try:
    runner = PipelineRunner.from_yaml(path, variables, agent_factory)
except PipelineValidationError as e:
    print(f"YAML inválido: {e}")

PipelineValidationError hereda de ValueError para backward compatibility.

---

Salida visual (HUMAN logging)

A partir de v1.1.0, el Pipeline Mode emite eventos de nivel HUMAN que producen una salida visual clara en stderr. El usuario puede ver en tiempo real el progreso de cada step sin necesidad de flags -v.

━ Pipeline step 1/3: analyze (agent: plan) ━━━━━━━━━━━━━━━━━━
   ✓ Step 'analyze' → success ($0.0234, 12.5s)

━ Pipeline step 2/3: implement (agent: build) ━━━━━━━━━━━━━━━
   ✓ Step 'implement' → success ($0.0456, 89.3s)

   ⏭️  Step 'deploy' omitido (condición no cumplida)

Eventos emitidos

Evento	Cuándo	Datos
pipeline.step_start	Al inicio de cada step	step, agent, index, total
pipeline.step_skipped	Cuando la condición no se cumple	step
pipeline.step_done	Al completar cada step	step, status, cost, duration

Se desactiva con --quiet o --json. Ver logging.md para detalles del sistema HUMAN.

---

Flujo interno

architect pipeline workflow.yaml --var feature="auth"
  │
  ├── 1. PipelineRunner.from_yaml(path, variables)
  │       ├── yaml.safe_load(file)
  │       ├── _validate_steps(steps_data) → PipelineValidationError si hay errores
  │       ├── Merge variables YAML + CLI
  │       └── Construir PipelineConfig con steps
  │
  ├── 2. runner.run(from_step=None, dry_run=False)
  │       │
  │       ├── Para cada step:
  │       │   ├── 2a. _eval_condition(condition) → skip si False
  │       │   ├── 2b. _resolve_vars(prompt) → sustituir {{variables}}
  │       │   ├── 2c. agent_factory(agent=step.agent, model=step.model)
  │       │   │       └── AgentLoop fresco con ContextBuilder, CostTracker, etc.
  │       │   ├── 2d. agent.run(resolved_prompt) → AgentState
  │       │   ├── 2e. Si output_var: variables[output_var] = state.final_output
  │       │   ├── 2f. Si checks: _run_checks(checks) → checks_passed
  │       │   ├── 2g. Si checkpoint: _create_checkpoint(step_name)
  │       │   └── 2h. Registrar PipelineStepResult
  │       │
  │       └── Retornar list[PipelineStepResult]
  │
  └── 3. Mostrar resultados

---

API Python

PipelineConfig

@dataclass
class PipelineConfig:
    name: str                          # Nombre del pipeline
    steps: list[PipelineStep]          # Pasos secuenciales
    variables: dict[str, str]          # Variables iniciales

PipelineStep

@dataclass
class PipelineStep:
    name: str                          # Identificador
    agent: str = "build"               # Agente a usar
    prompt: str = ""                   # Prompt (soporta {{variables}})
    model: str | None = None           # Modelo LLM override
    checkpoint: bool = False           # Crear git checkpoint
    condition: str | None = None       # Condición para ejecutar
    output_var: str | None = None      # Variable donde guardar output
    checks: list[str] = []            # Comandos de verificación
    timeout: int | None = None         # Timeout en segundos

PipelineRunner

class PipelineRunner:
    def __init__(
        self,
        config: PipelineConfig,
        agent_factory: Callable[..., Any],
        workspace_root: str | None = None,
    ) -> None: ...

    def run(
        self,
        from_step: str | None = None,
        dry_run: bool = False,
    ) -> list[PipelineStepResult]: ...

    def get_plan_summary(self) -> str: ...

    @classmethod
    def from_yaml(
        cls,
        path: str,
        variables: dict[str, str],
        agent_factory: Callable[..., Any],
        workspace_root: str | None = None,
    ) -> "PipelineRunner": ...

PipelineStepResult

@dataclass
class PipelineStepResult:
    step_name: str                     # Identificador del paso
    status: str                        # "success", "partial", "failed", "skipped", "dry_run"
    cost: float = 0.0                  # Coste USD
    duration: float = 0.0              # Segundos
    checks_passed: bool = True         # True si todos los checks pasaron
    error: str | None = None           # Mensaje de error

---

Ejemplos

Pipeline de feature completa

name: feature-pipeline
variables:
  branch: feature/auth
steps:
  - name: plan
    agent: plan
    prompt: |
      Analiza el proyecto y planifica cómo implementar
      autenticación JWT. Lista los archivos a modificar
      y el orden de los cambios.
    output_var: plan

  - name: implement
    agent: build
    prompt: |
      Ejecuta este plan paso a paso:
      {{plan}}
    model: gpt-4o
    checks:
      - "pytest tests/ -q"
      - "ruff check src/"
    checkpoint: true

  - name: docs
    agent: build
    prompt: "Actualiza la documentación para reflejar los cambios de autenticación"
    checkpoint: true

Pipeline de CI/CD

name: ci-review
variables:
  base_branch: origin/main
steps:
  - name: review
    agent: review
    prompt: "Revisa los cambios de este PR respecto a {{base_branch}}"
    output_var: review_result

  - name: fix
    agent: build
    prompt: "Corrige estos problemas encontrados en la review: {{review_result}}"
    condition: "auto_fix == 'true'"
    checks:
      - "pytest tests/ -q"

architect pipeline ci-review.yaml \
  --var base_branch=origin/main \
  --var auto_fix=true

Pipeline con múltiples modelos

name: multi-model
steps:
  - name: draft
    agent: build
    model: gpt-4o-mini          # Modelo rápido para el draft
    prompt: "Genera un primer borrador de tests para auth.py"
    output_var: draft

  - name: refine
    agent: build
    model: claude-sonnet-4-6    # Modelo más capaz para refinar
    prompt: "Mejora y completa estos tests: {{draft}}"
    checks:
      - "pytest tests/test_auth.py -v"