Workflows

# From grip/workflow/models.py:100-113
@dataclass(slots=True)
class WorkflowDef:
    """Complete workflow definition: a named DAG of steps.
    
    Steps are validated at load time to ensure:
      - Non-empty workflow name
      - At least one step
      - Valid step names (alphanumeric, underscore, hyphen)
      - Positive timeout values
      - No duplicate step names
      - All depends_on references point to existing steps
      - No circular dependencies
    """
    name: str
    description: str = ""
    steps: list[StepDef] = field(default_factory=list)

# From grip/workflow/models.py:32-44
@dataclass(slots=True)
class StepDef:
    """Definition of a single workflow step.
    
    depends_on lists step names that must complete before this step runs.
    prompt can contain {{step_name.output}} placeholders that are resolved
    at execution time from prior step results.
    """
    name: str              # Unique identifier (alphanumeric, underscore, hyphen)
    prompt: str            # Task description sent to the agent
    profile: str = "default"  # Agent profile to use
    depends_on: list[str] = field(default_factory=list)  # Step dependencies
    timeout_seconds: int = 300  # Per-step timeout

# From grip/workflow/models.py:159-179
def get_execution_order(self) -> list[list[str]]:
    """Return steps grouped into parallel execution layers.
    
    Each layer contains steps whose dependencies are all in earlier
    layers, so they can execute concurrently.
    """
    adj, in_degree = _build_graph(self.steps)
    layers: list[list[str]] = []
    queue = deque(n for n, d in in_degree.items() if d == 0)
    
    while queue:
        layer = sorted(queue)
        layers.append(layer)
        queue.clear()
        for node in layer:
            for neighbor in adj[node]:
                in_degree[neighbor] -= 1
                if in_degree[neighbor] == 0:
                    queue.append(neighbor)
    
    return layers

# For the parallel-pipeline workflow above:
Layer 1: ["fetch_data"]                              # No dependencies
Layer 2: ["analyze_frontend", "analyze_backend"]     # Both depend on fetch_data
Layer 3: ["merge_results"]                           # Depends on both analyses

# From grip/workflow/engine.py:76-93
for layer_idx, layer_names in enumerate(layers, 1):
    runnable = [
        name for name in layer_names
        if result.step_results[name].status != StepStatus.SKIPPED
    ]
    if not runnable:
        continue
    
    logger.info("Executing layer {}/{}: {}", layer_idx, len(layers), runnable)
    
    tasks = []
    for step_name in runnable:
        step_def = step_map[step_name]
        step_result = result.step_results[step_name]
        resolved_prompt = self._resolve_template(step_def.prompt, result.step_results)
        tasks.append(self._execute_step(step_def, step_result, resolved_prompt))
    
    await asyncio.gather(*tasks)

# From grip/workflow/engine.py:166-181
@staticmethod
def _resolve_template(prompt: str, step_results: dict[str, StepResult]) -> str:
    """Replace {{step_name.output}} placeholders with sanitized step outputs.
    
    Outputs are stripped of nested template patterns, truncated, and wrapped
    in delimiters to prevent injection and clarify boundaries for the LLM.
    """
    def replacer(match: re.Match) -> str:
        step_name = match.group(1)
        result = step_results.get(step_name)
        if result and result.status == StepStatus.COMPLETED:
            sanitized = WorkflowEngine._sanitize_output(result.output)
            return f"[output from {step_name}]\n{sanitized}\n[/output from {step_name}]"
        return match.group(0)
    
    return TEMPLATE_PATTERN.sub(replacer, prompt)

# From grip/workflow/models.py:219-259
@dataclass(slots=True)
class WorkflowRunResult:
    """Complete result of a workflow execution."""
    workflow_name: str
    status: str = "pending"  # pending, running, completed, failed, partial
    step_results: dict[str, StepResult] = field(default_factory=dict)
    started_at: str = ""
    completed_at: str = ""
    total_duration_seconds: float = 0.0
    
    def to_dict(self) -> dict[str, Any]:
        return {
            "workflow_name": self.workflow_name,
            "status": self.status,
            "started_at": self.started_at,
            "completed_at": self.completed_at,
            "total_duration_seconds": self.total_duration_seconds,
            "steps": {
                name: {
                    "status": r.status.value,
                    "output": r.output[:500] + "... [truncated]" if len(r.output) > 500 else r.output,
                    "error": r.error,
                    "iterations": r.iterations,
                    "duration_seconds": r.duration_seconds,
                }
                for name, r in self.step_results.items()
            },
        }

# From grip/workflow/store.py:31-42
def save(self, workflow: WorkflowDef) -> Path:
    """Save a workflow definition to disk."""
    self._validate_name(workflow.name)
    path = self._dir / f"{workflow.name}.json"
    tmp = path.with_suffix(".tmp")
    tmp.write_text(
        json.dumps(workflow.to_dict(), indent=2, ensure_ascii=False),
        encoding="utf-8",
    )
    tmp.rename(path)  # Atomic write
    logger.debug("Saved workflow: {}", workflow.name)
    return path

# List workflows
grip workflow list

# Run a workflow
grip workflow run feature-pipeline

# Create from template
grip workflow create my-workflow --template parallel

# Delete a workflow
grip workflow delete old-workflow