diff --git a/.gitignore b/.gitignore
index d0d72309..dc4314e3 100644
--- a/.gitignore
+++ b/.gitignore
@@ -52,3 +52,8 @@ internal/web/ui/dist/
 # goreleaser before-hook output (release.extra_files)
 ty-chrome.zip
 .superpowers/
+.qa-run/
+.qa-agents/
+.pr-body.md
+.pr-comment.md
+.qacfg/
diff --git a/README.md b/README.md
index d1e6c58f..5c433c27 100644
--- a/README.md
+++ b/README.md
@@ -76,6 +76,7 @@ The same UI is also served in your browser at `http://localhost:8484` whenever `
 - **Kanban Board** - Visual task management with 4 columns (Backlog, In Progress, Blocked, Done)
 - **Git Worktrees** - Each task runs in an isolated worktree, no conflicts between parallel tasks
 - **Pluggable Executors** - Choose between Claude Code, OpenAI Codex, Gemini, Pi, OpenClaw, or OpenCode per task
+- **Workflows** - Turn one goal into a multi-step DAG (e.g. plan → code → parallel review → collect), each step on its own executor/model, advancing automatically (see [Workflows](#workflows))
 - **Event Hooks** - Run scripts when tasks change state (see [Event Hooks](#event-hooks))
 - **Ghost Text Autocomplete** - LLM-powered suggestions for task titles and descriptions as you type
 - **VS Code-style Fuzzy Search** - Quick task navigation with smart matching (e.g., "dsno" matches "diseno website")
@@ -88,6 +89,66 @@ The same UI is also served in your browser at `http://localhost:8484` whenever `
 - **Project Context Caching** - AI agents automatically cache codebase exploration results and reuse them across tasks, eliminating redundant exploration (see [Project Context](#project-context))
 - **Shell Completion** - Tab completion for commands, task IDs, projects, statuses, and flags in bash, zsh, fish, and PowerShell (see [Shell Completion](#shell-completion))
 
+## Workflows
+
+A **workflow** turns a single goal into a small DAG of step tasks that run on one shared git branch, each routed to its own executor and model, advancing automatically. Steps are sequential where they depend on each other and **parallel** where they don't.
+
+The built-in `plan-code-review` workflow:
+
+```
+Plan ──▶ Code ──▶ Review A ─┐
+                  Review B ─┴─▶ Collect ──▶ PR
+```
+
+| Step | Default | Job |
+|------|---------|-----|
+| **Plan** | Claude / Opus | Explore, write `PLAN.md`, push. No code. |
+| **Code** | Claude / Sonnet | Implement the plan, push. |
+| **Review A**, **Review B** | Claude / Opus, Claude / Sonnet | Two **independent** reviewers in parallel — different models + independent context catch different issues and avoid self-review bias. |
+| **Collect** | Claude / Sonnet | Read both reviews, apply the fixes worth applying, open the PR. |
+
+Steps advance with no human in the loop; a workflow only pauses when a step genuinely needs one — the final step opens a PR (landing in `blocked` for a human merge) or a step asks for input.
+
+### Running a workflow
+
+```bash
+# CLI
+ty pipeline "Add rate limiting to the API" --project myapp
+ty pipeline --list                 # show available workflows
+ty pipeline "..." --no-execute     # stage without starting
+
+# TUI: in the new-task form (n), pick a workflow in the "Workflow" selector.
+```
+
+On the board, a workflow shows as a **single card** (`⇄ goal · Review ∥ · 3/5`) instead of one card per step. It needs a project that uses git worktrees and has a remote to push to.
+
+### Custom workflows
+
+Workflows are defined in plain **YAML files** — one per workflow — in `~/.config/task/workflows/*.yaml` (override with `$TY_WORKFLOWS_DIR`), or per-project in `.taskyou/workflows/`. A file shadows a built-in of the same name. You write only *what* each step does and its `deps`; the git handoff (which branch to push to, when to open the PR) is derived from the step's position in the DAG.
+
+```yaml
+name: build-and-qa
+description: Plan, build, then security review and QA in parallel, then finalize.
+steps:
+  - {name: Plan,     model: opus,   prompt: "Design a plan for {{goal}}; write PLAN.md."}
+  - {name: Build,    deps: [Plan],  prompt: "Implement the plan."}
+  - {name: Security, deps: [Build], prompt: "Security review; write findings to security.md."}
+  - {name: QA,       deps: [Build], prompt: "Exercise the change; write results to qa.md."}
+  - {name: Finalize, deps: [Security, QA], prompt: "Address the findings and finalize."}
+```
+
+Two steps with the same `deps` run in parallel; a step depending on several joins them; multiple root steps (no `deps`) are parallel entry points (e.g. try 3 approaches at once).
+
+```bash
+# Author a workflow from a plain-English description (LLM → YAML you can edit)
+ty pipeline new "spike three approaches, pick the best, build it, review and test in parallel"
+
+# Eject the built-in to a YAML file to tweak its models / prompts / steps
+ty pipeline edit                   # writes ~/.config/task/workflows/plan-code-review.yaml
+```
+
+Custom workflows appear in `ty pipeline --list`, the `--definition` flag, and the TUI new-task selector automatically. Configuration lives entirely in these files — edit them by hand any time.
+
 ## Project Context
 
 TaskYou implements **intelligent codebase caching** to make AI agents dramatically more efficient across multiple tasks in the same project.
diff --git a/cmd/task/main.go b/cmd/task/main.go
index 8b8570f7..2d4d5dfd 100644
--- a/cmd/task/main.go
+++ b/cmd/task/main.go
@@ -32,6 +32,7 @@ import (
 	"github.com/bborn/workflow/internal/github"
 	"github.com/bborn/workflow/internal/hooks"
 	"github.com/bborn/workflow/internal/mcp"
+	"github.com/bborn/workflow/internal/pipeline"
 	"github.com/bborn/workflow/internal/ui"
 	"github.com/bborn/workflow/internal/web"
 )
@@ -810,6 +811,322 @@ Examples:
 	})
 	rootCmd.AddCommand(createCmd)
 
+	// Pipeline subcommand - create a multi-phase pipeline task
+	pipelineCmd := &cobra.Command{
+		Use:   "pipeline [goal]",
+		Short: "Create a multi-model plan → code → review pipeline for a goal",
+		Long: `Create a workflow: one goal broken into a small DAG of step tasks, each routed
+to its own executor and model, that hand work forward on a single shared branch
+and advance automatically — sequential where steps depend on each other, parallel
+where they don't.
+
+The default 'plan-code-review' workflow runs five steps on one branch:
+  Plan     (Claude / Opus)   — writes PLAN.md, no code
+  Code     (Claude / Sonnet) — implements the plan
+  Review A (Claude / Opus)   \
+  Review B (Claude / Sonnet)  } two independent reviewers, in parallel
+  Collect  (Claude / Sonnet) — merges reviews, applies fixes, opens the PR
+
+Workflows are defined in YAML files you can edit; define your own (add a QA step,
+a different shape) with 'task pipeline new "<describe it>"' or 'task pipeline edit'.
+
+Each step commits and pushes the shared branch, so the workflow needs a project
+that uses git worktrees and has a remote to push to.
+
+Examples:
+  task pipeline "Add rate limiting to the API" --project myapp
+  task pipeline "Refactor the auth module" -p myapp --definition my-workflow
+  task pipeline new "plan, build, security review + QA in parallel, then finalize"
+  task pipeline --list  # show available workflows`,
+		Args: cobra.MaximumNArgs(1),
+		Run: func(cmd *cobra.Command, args []string) {
+			listDefs, _ := cmd.Flags().GetBool("list")
+			if listDefs {
+				for _, d := range pipeline.Definitions() {
+					stepLabels := make([]string, 0, len(d.Steps))
+					for _, s := range d.Steps {
+						label := s.Name + " (" + s.Executor
+						if s.Model != "" {
+							label += "/" + s.Model
+						}
+						label += ")"
+						if len(s.Deps) > 0 {
+							label += " ← " + strings.Join(s.Deps, "+")
+						}
+						stepLabels = append(stepLabels, label)
+					}
+					origin := "built-in"
+					if d.Custom {
+						origin = "custom"
+					}
+					fmt.Printf("%s %s\n  %s\n  steps: %s\n", successStyle.Render(d.Name), dimStyle.Render("("+origin+")"), d.Description, strings.Join(stepLabels, " · "))
+				}
+				fmt.Println(dimStyle.Render("Custom workflows: " + pipeline.WorkflowsDir() + "/*.yaml  ·  create with: task pipeline new \"<describe it>\""))
+				return
+			}
+
+			var goal string
+			if len(args) > 0 {
+				goal = args[0]
+			}
+			body, _ := cmd.Flags().GetString("body")
+			body = unescapeNewlines(body)
+			if strings.TrimSpace(goal) == "" {
+				goal = body
+			}
+			if strings.TrimSpace(goal) == "" {
+				fmt.Fprintln(os.Stderr, errorStyle.Render("Error: a goal (positional argument or --body) is required"))
+				os.Exit(1)
+			}
+
+			project, _ := cmd.Flags().GetString("project")
+			definition, _ := cmd.Flags().GetString("definition")
+			permissionModeFlag, _ := cmd.Flags().GetString("permission-mode")
+			createDangerous, _ := cmd.Flags().GetBool("dangerous")
+			noExecute, _ := cmd.Flags().GetBool("no-execute")
+			outputJSON, _ := cmd.Flags().GetBool("json")
+
+			dbPath := db.DefaultPath()
+			database, err := openTaskDB(dbPath)
+			if err != nil {
+				fmt.Fprintln(os.Stderr, errorStyle.Render("Error: "+err.Error()))
+				os.Exit(1)
+			}
+			defer database.Close()
+
+			// Detect project from cwd if not specified.
+			if project == "" {
+				if cwd, err := os.Getwd(); err == nil {
+					if p, err := database.GetProjectByPath(cwd); err == nil && p != nil {
+						project = p.Name
+					}
+				}
+			}
+			if project == "" {
+				fmt.Fprintln(os.Stderr, errorStyle.Render("Error: could not determine project; pass --project"))
+				os.Exit(1)
+			}
+
+			permMode := db.NormalizePermissionMode(permissionModeFlag)
+			if permMode == "" && createDangerous {
+				permMode = db.PermissionModeDangerous
+			}
+
+			result, err := pipeline.Create(database, pipeline.Options{
+				Goal:           goal,
+				Project:        project,
+				Definition:     definition,
+				PermissionMode: permMode,
+				Execute:        !noExecute,
+			})
+			if err != nil {
+				fmt.Fprintln(os.Stderr, errorStyle.Render("Error: "+err.Error()))
+				os.Exit(1)
+			}
+
+			if outputJSON {
+				steps := make([]map[string]interface{}, 0, len(result.Tasks))
+				for i, t := range result.Tasks {
+					steps = append(steps, map[string]interface{}{
+						"step":     result.Definition.Steps[i].Name,
+						"deps":     result.Definition.Steps[i].Deps,
+						"id":       t.ID,
+						"executor": t.Executor,
+						"model":    t.Model,
+						"status":   t.Status,
+					})
+				}
+				out := map[string]interface{}{
+					"definition": result.Definition.Name,
+					"branch":     result.Branch,
+					"project":    project,
+					"steps":      steps,
+				}
+				jsonBytes, _ := json.Marshal(out)
+				fmt.Println(string(jsonBytes))
+				return
+			}
+
+			fmt.Println(successStyle.Render(fmt.Sprintf("Created %s workflow on branch %s", result.Definition.Name, result.Branch)))
+			for i, t := range result.Tasks {
+				s := result.Definition.Steps[i]
+				model := s.Model
+				if model == "" {
+					model = "default"
+				}
+				dep := ""
+				if len(s.Deps) > 0 {
+					dep = "  ← " + strings.Join(s.Deps, "+")
+				}
+				fmt.Printf("  #%d  %-9s %s/%s  (%s)%s\n", t.ID, s.Name, t.Executor, model, t.Status, dep)
+			}
+			if noExecute {
+				fmt.Println(dimStyle.Render("Staged but not started — queue the root step to run it."))
+			} else {
+				fmt.Println(dimStyle.Render("Running — steps advance automatically; parallel reviewers run at once."))
+			}
+		},
+	}
+	pipelineCmd.Flags().String("body", "", "Goal text (alternative to the positional argument)")
+	pipelineCmd.Flags().StringP("project", "p", "", "Project name (auto-detected from cwd if not specified)")
+	pipelineCmd.Flags().StringP("definition", "d", pipeline.DefaultDefinition, "Pipeline definition to use")
+	pipelineCmd.Flags().String("permission-mode", "", "Permission mode for every phase: default, accept-edits, auto, dangerous (defaults to the project's setting)")
+	pipelineCmd.Flags().Bool("dangerous", false, "Run every phase in dangerous mode (alias for --permission-mode dangerous)")
+	pipelineCmd.Flags().Bool("no-execute", false, "Stage the workflow without queuing the root step")
+	pipelineCmd.Flags().Bool("list", false, "List available workflow definitions and exit")
+	pipelineCmd.Flags().Bool("json", false, "Output in JSON format")
+	pipelineCmd.RegisterFlagCompletionFunc("project", completeFlagProjects)
+	pipelineCmd.RegisterFlagCompletionFunc("definition", func(cmd *cobra.Command, args []string, toComplete string) ([]string, cobra.ShellCompDirective) {
+		return pipeline.DefinitionNames(), cobra.ShellCompDirectiveNoFileComp
+	})
+
+	// Pipeline new subcommand - author a custom workflow from a free-text description
+	pipelineNewCmd := &cobra.Command{
+		Use:   "new [description]",
+		Short: "Create a custom workflow from a free-text description (LLM → YAML)",
+		Long: `Describe a workflow in plain English and get a ready-to-edit YAML file.
+The steps, dependencies (the DAG), and per-step prompts are generated; the git
+handoff between steps is added automatically at run time, so prompts describe
+only the work.
+
+Custom workflows live in ` + "`~/.config/task/workflows/*.yaml`" + ` (override with
+$TY_WORKFLOWS_DIR) and can also be dropped in a project's .taskyou/workflows/.
+Edit the file by hand any time — it's just YAML.
+
+Examples:
+  task pipeline new "spike three approaches, build the best, then write tests"
+  task pipeline new "plan, implement, security review, then QA in a browser" --name secure-flow
+  task pipeline new "..." --print        # print the YAML without saving`,
+		Args: cobra.MaximumNArgs(1),
+		Run: func(cmd *cobra.Command, args []string) {
+			desc, _ := cmd.Flags().GetString("body")
+			if len(args) > 0 {
+				desc = strings.TrimSpace(args[0] + " " + desc)
+			}
+			desc = strings.TrimSpace(unescapeNewlines(desc))
+			if desc == "" {
+				fmt.Fprintln(os.Stderr, errorStyle.Render("Error: describe the workflow (positional arg or --body)"))
+				os.Exit(1)
+			}
+			nameOverride, _ := cmd.Flags().GetString("name")
+			printOnly, _ := cmd.Flags().GetBool("print")
+
+			database, err := openTaskDB(db.DefaultPath())
+			if err != nil {
+				fmt.Fprintln(os.Stderr, errorStyle.Render("Error: "+err.Error()))
+				os.Exit(1)
+			}
+			defer database.Close()
+			apiKey, _ := database.GetSetting("anthropic_api_key")
+
+			fmt.Fprintln(os.Stderr, dimStyle.Render("Designing workflow…"))
+			ctx, cancel := context.WithTimeout(context.Background(), 70*time.Second)
+			defer cancel()
+			def, yamlBytes, err := pipeline.GenerateDefinition(ctx, apiKey, desc)
+			if err != nil {
+				fmt.Fprintln(os.Stderr, errorStyle.Render("Error: "+err.Error()))
+				os.Exit(1)
+			}
+			if nameOverride != "" {
+				def.Name = nameOverride
+				yamlBytes, _ = pipeline.Marshal(def)
+			}
+
+			if printOnly {
+				fmt.Print(string(yamlBytes))
+				return
+			}
+
+			dir := pipeline.WorkflowsDir()
+			if err := os.MkdirAll(dir, 0o755); err != nil {
+				fmt.Fprintln(os.Stderr, errorStyle.Render("Error: "+err.Error()))
+				os.Exit(1)
+			}
+			path := filepath.Join(dir, def.Name+".yaml")
+			if err := os.WriteFile(path, yamlBytes, 0o644); err != nil {
+				fmt.Fprintln(os.Stderr, errorStyle.Render("Error: "+err.Error()))
+				os.Exit(1)
+			}
+
+			fmt.Println(successStyle.Render("Created workflow '" + def.Name + "'"))
+			for _, s := range def.Steps {
+				dep := ""
+				if len(s.Deps) > 0 {
+					dep = "  ← " + strings.Join(s.Deps, "+")
+				}
+				model := s.Model
+				if model == "" {
+					model = "default"
+				}
+				fmt.Printf("  %-12s %s/%s%s\n", s.Name, s.Executor, model, dep)
+			}
+			fmt.Println(dimStyle.Render("Saved to " + path + " — edit it any time."))
+			fmt.Println(dimStyle.Render("Run it with: task pipeline \"<goal>\" --definition " + def.Name))
+		},
+	}
+	pipelineNewCmd.Flags().String("body", "", "Workflow description (alternative to the positional argument)")
+	pipelineNewCmd.Flags().String("name", "", "Override the generated workflow name")
+	pipelineNewCmd.Flags().Bool("print", false, "Print the YAML without saving")
+	pipelineCmd.AddCommand(pipelineNewCmd)
+
+	// Pipeline edit subcommand - write a workflow to a YAML file for editing
+	pipelineEditCmd := &cobra.Command{
+		Use:   "edit [name]",
+		Short: "Write a workflow to a YAML file you can edit (ejects the built-in)",
+		Long: `Workflows are configured by editing their YAML file. This writes the named
+workflow (default: ` + pipeline.DefaultDefinition + `) to the workflows directory so you can
+change models, prompts, or steps by hand. A custom file shadows the built-in of
+the same name.
+
+Examples:
+  task pipeline edit                       # eject the default workflow to edit
+  task pipeline edit plan-code-review      # same, explicit
+  task pipeline edit --print               # print the YAML instead of writing it`,
+		Args: cobra.MaximumNArgs(1),
+		Run: func(cmd *cobra.Command, args []string) {
+			name := pipeline.DefaultDefinition
+			if len(args) > 0 {
+				name = args[0]
+			}
+			printOnly, _ := cmd.Flags().GetBool("print")
+
+			def, ok := pipeline.Get(name)
+			if !ok {
+				fmt.Fprintln(os.Stderr, errorStyle.Render("Error: unknown workflow "+name))
+				os.Exit(1)
+			}
+			yamlBytes, err := pipeline.Marshal(def)
+			if err != nil {
+				fmt.Fprintln(os.Stderr, errorStyle.Render("Error: "+err.Error()))
+				os.Exit(1)
+			}
+			if printOnly {
+				fmt.Print(string(yamlBytes))
+				return
+			}
+			dir := pipeline.WorkflowsDir()
+			if err := os.MkdirAll(dir, 0o755); err != nil {
+				fmt.Fprintln(os.Stderr, errorStyle.Render("Error: "+err.Error()))
+				os.Exit(1)
+			}
+			path := filepath.Join(dir, def.Name+".yaml")
+			if _, statErr := os.Stat(path); statErr == nil {
+				fmt.Println(dimStyle.Render("Already exists — edit it: " + path))
+				return
+			}
+			if err := os.WriteFile(path, yamlBytes, 0o644); err != nil {
+				fmt.Fprintln(os.Stderr, errorStyle.Render("Error: "+err.Error()))
+				os.Exit(1)
+			}
+			fmt.Println(successStyle.Render("Wrote " + def.Name + " to " + path))
+			fmt.Println(dimStyle.Render("Edit it, then run: task pipeline \"<goal>\" --definition " + def.Name))
+		},
+	}
+	pipelineEditCmd.Flags().Bool("print", false, "Print the YAML instead of writing a file")
+	pipelineCmd.AddCommand(pipelineEditCmd)
+
+	rootCmd.AddCommand(pipelineCmd)
+
 	// List subcommand - list tasks
 	listCmd := &cobra.Command{
 		Use:   "list",
diff --git a/docs/mcp-tools-reference.md b/docs/mcp-tools-reference.md
index 5d99f5e7..ff6af8e0 100644
--- a/docs/mcp-tools-reference.md
+++ b/docs/mcp-tools-reference.md
@@ -128,6 +128,27 @@ Create a new task in the system.
 }
 ```
 
+### taskyou_create_pipeline
+
+Create a multi-step **workflow** for a goal: the goal is split into a DAG of step tasks (default `plan-code-review`: plan → code → two parallel reviewers → collect), each on its own executor/model, sharing one git branch and advancing automatically. Use this instead of a single task when a goal benefits from a plan/code/review split. The first step is queued immediately. Requires a git-worktree project with a remote. See [Workflows](../README.md#workflows).
+
+**Parameters:**
+- `goal` (string, required) - The overall goal, threaded into every step's prompt
+- `project` (string, optional) - Project name (defaults to current project)
+- `definition` (string, optional) - Workflow name (defaults to `plan-code-review`; custom workflows from `~/.config/task/workflows/*.yaml` are also accepted)
+- `permission_mode` (string, optional) - Permission mode for every step (`default`, `accept-edits`, `auto`, `dangerous`)
+
+**Example:**
+```json
+{
+  "name": "taskyou_create_pipeline",
+  "arguments": {
+    "goal": "Add rate limiting to the API",
+    "definition": "plan-code-review"
+  }
+}
+```
+
 ### taskyou_list_tasks
 
 List active tasks in the project.
diff --git a/go.mod b/go.mod
index c5c91a08..38be10e9 100644
--- a/go.mod
+++ b/go.mod
@@ -64,7 +64,7 @@ require (
 	github.com/sahilm/fuzzy v0.1.1 // indirect
 	github.com/spf13/pflag v1.0.9 // indirect
 	github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e // indirect
-	github.com/yuin/goldmark v1.7.13 // indirect
+	github.com/yuin/goldmark v1.7.17 // indirect
 	github.com/yuin/goldmark-emoji v1.0.6 // indirect
 	golang.org/x/crypto v0.52.0 // indirect
 	golang.org/x/exp v0.0.0-20251023183803-a4bb9ffd2546 // indirect
diff --git a/go.sum b/go.sum
index af750753..a9fef723 100644
--- a/go.sum
+++ b/go.sum
@@ -143,8 +143,8 @@ github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu
 github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
 github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e h1:JVG44RsyaB9T2KIHavMF/ppJZNG9ZpyihvCd0w101no=
 github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e/go.mod h1:RbqR21r5mrJuqunuUZ/Dhy/avygyECGrLceyNeo4LiM=
-github.com/yuin/goldmark v1.7.13 h1:GPddIs617DnBLFFVJFgpo1aBfe/4xcvMc3SB5t/D0pA=
-github.com/yuin/goldmark v1.7.13/go.mod h1:ip/1k0VRfGynBgxOz0yCqHrbZXhcjxyuS66Brc7iBKg=
+github.com/yuin/goldmark v1.7.17 h1:p36OVWwRb246iHxA/U4p8OPEpOTESm4n+g+8t0EE5uA=
+github.com/yuin/goldmark v1.7.17/go.mod h1:ip/1k0VRfGynBgxOz0yCqHrbZXhcjxyuS66Brc7iBKg=
 github.com/yuin/goldmark-emoji v1.0.6 h1:QWfF2FYaXwL74tfGOW5izeiZepUDroDJfWubQI9HTHs=
 github.com/yuin/goldmark-emoji v1.0.6/go.mod h1:ukxJDKFpdFb5x0a5HqbdlcKtebh086iJpI31LTKmWuA=
 go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg=
diff --git a/internal/executor/executor.go b/internal/executor/executor.go
index 700fdd91..f079e964 100644
--- a/internal/executor/executor.go
+++ b/internal/executor/executor.go
@@ -3875,7 +3875,7 @@ func (e *Executor) setupWorktree(task *db.Task) (string, error) {
 
 	// Generate slug from title (e.g., "Add contact email" -> "add-contact-email")
 	slug := slugify(task.Title, 40)
-	branchName := fmt.Sprintf("task/%d-%s", task.ID, slug)
+	branchName := newWorktreeBranchName(task, slug)
 	dirName := fmt.Sprintf("%d-%s", task.ID, slug)
 	worktreePath := filepath.Join(worktreesDir, dirName)
 
@@ -5713,6 +5713,22 @@ func isValidWorkDir(workDir string, allowedProjectDir string) bool {
 	return err == nil && info.IsDir()
 }
 
+// newWorktreeBranchName returns the branch name to create for a task's fresh
+// worktree. It normally derives a unique name from the task ID and title slug,
+// but honors a caller-pinned BranchName when one is set. Pinning lets a pipeline
+// route several phase tasks onto one shared branch (each phase after the first
+// checks it out via SourceBranch), so a plan → code → review chain hands work
+// forward on a single branch. A pinned name is only ever set deliberately at
+// creation time: by the point setupWorktree reaches here, an ordinary task with
+// no live worktree has already had BranchName cleared, so this never hijacks the
+// normal task/<id>-<slug> naming.
+func newWorktreeBranchName(task *db.Task, slug string) string {
+	if strings.TrimSpace(task.BranchName) != "" {
+		return task.BranchName
+	}
+	return fmt.Sprintf("task/%d-%s", task.ID, slug)
+}
+
 // slugify converts a string to a URL/branch-friendly slug.
 func slugify(s string, maxLen int) string {
 	// Convert to lowercase
diff --git a/internal/executor/worktree_branch_test.go b/internal/executor/worktree_branch_test.go
new file mode 100644
index 00000000..7f204493
--- /dev/null
+++ b/internal/executor/worktree_branch_test.go
@@ -0,0 +1,29 @@
+package executor
+
+import (
+	"testing"
+
+	"github.com/bborn/workflow/internal/db"
+)
+
+func TestNewWorktreeBranchName(t *testing.T) {
+	// No pinned branch: derive the usual task/<id>-<slug> name.
+	got := newWorktreeBranchName(&db.Task{ID: 42}, "add-dark-mode")
+	if want := "task/42-add-dark-mode"; got != want {
+		t.Errorf("derived branch = %q, want %q", got, want)
+	}
+
+	// A pinned BranchName (as a pipeline sets) is honored verbatim so several
+	// phase tasks can share one branch.
+	pinned := "pipeline/7-add-rate-limiting"
+	got = newWorktreeBranchName(&db.Task{ID: 42, BranchName: pinned}, "add-dark-mode")
+	if got != pinned {
+		t.Errorf("pinned branch = %q, want %q", got, pinned)
+	}
+
+	// Whitespace-only pins are ignored.
+	got = newWorktreeBranchName(&db.Task{ID: 9, BranchName: "   "}, "x")
+	if want := "task/9-x"; got != want {
+		t.Errorf("blank pin branch = %q, want %q", got, want)
+	}
+}
diff --git a/internal/mcp/server.go b/internal/mcp/server.go
index 0de499b5..8b18e151 100644
--- a/internal/mcp/server.go
+++ b/internal/mcp/server.go
@@ -16,6 +16,7 @@ import (
 
 	"github.com/bborn/workflow/internal/db"
 	"github.com/bborn/workflow/internal/github"
+	"github.com/bborn/workflow/internal/pipeline"
 	"github.com/bborn/workflow/internal/tasksummary"
 )
 
@@ -256,6 +257,34 @@ func (s *Server) handleRequest(req *jsonRPCRequest) {
 						"required": []string{"title"},
 					},
 				},
+				{
+					Name:        "taskyou_create_pipeline",
+					Description: "Create a multi-model pipeline for a goal: one goal is split into an ordered chain of phase tasks (default 'plan-code-review': Opus plans → Sonnet codes → two reviewers run in parallel → collect opens the PR), each routed to its own executor/model, all on one shared git branch. Each step runs on its own executor/model; define custom workflows as YAML (ty pipeline new). Steps advance automatically — sequential where they depend on each other, parallel where they don't. Use this instead of a single task when a goal benefits from a plan/code/review split across different models. The first phase is queued immediately. Requires a git-worktree project with a remote to push to.",
+					InputSchema: map[string]interface{}{
+						"type": "object",
+						"properties": map[string]interface{}{
+							"goal": map[string]interface{}{
+								"type":        "string",
+								"description": "The overall goal, threaded into every phase's prompt.",
+							},
+							"project": map[string]interface{}{
+								"type":        "string",
+								"description": "Project name (defaults to the current task's project).",
+							},
+							"definition": map[string]interface{}{
+								"type":        "string",
+								"description": "Pipeline definition name (defaults to 'plan-code-review').",
+								"enum":        pipeline.DefinitionNames(),
+							},
+							"permission_mode": map[string]interface{}{
+								"type":        "string",
+								"description": "Permission mode for every phase (default, accept-edits, auto, dangerous). Defaults to the project's configured default.",
+								"enum":        []string{"default", "accept-edits", "auto", "dangerous"},
+							},
+						},
+						"required": []string{"goal"},
+					},
+				},
 				{
 					Name:        "taskyou_list_tasks",
 					Description: "List active tasks (queued, processing, blocked, backlog) in the project. Use this to see what work is pending or in progress.",
@@ -548,6 +577,58 @@ func (s *Server) handleToolCall(id interface{}, params *toolCallParams) {
 			},
 		})
 
+	case "taskyou_create_pipeline":
+		goal, _ := params.Arguments["goal"].(string)
+		if strings.TrimSpace(goal) == "" {
+			s.sendError(id, -32602, "goal is required")
+			return
+		}
+		project, _ := params.Arguments["project"].(string)
+		definition, _ := params.Arguments["definition"].(string)
+		permissionMode, _ := params.Arguments["permission_mode"].(string)
+
+		// Default project to the current task's project.
+		if project == "" {
+			currentTask, err := s.db.GetTask(s.taskID)
+			if err == nil && currentTask != nil {
+				project = currentTask.Project
+			}
+		}
+
+		result, err := pipeline.Create(s.db, pipeline.Options{
+			Goal:           goal,
+			Project:        project,
+			Definition:     definition,
+			PermissionMode: db.NormalizePermissionMode(permissionMode),
+			Execute:        true,
+		})
+		if err != nil {
+			s.sendError(id, -32603, fmt.Sprintf("Failed to create pipeline: %v", err))
+			return
+		}
+
+		var sb strings.Builder
+		sb.WriteString(fmt.Sprintf("Created %s workflow on branch %s:\n", result.Definition.Name, result.Branch))
+		for i, t := range result.Tasks {
+			s := result.Definition.Steps[i]
+			model := s.Model
+			if model == "" {
+				model = "default"
+			}
+			dep := ""
+			if len(s.Deps) > 0 {
+				dep = " ← " + strings.Join(s.Deps, "+")
+			}
+			sb.WriteString(fmt.Sprintf("- #%d %s (%s/%s) — %s%s\n", t.ID, s.Name, t.Executor, model, t.Status, dep))
+		}
+		sb.WriteString("The root step is running; steps advance automatically, with the two reviewers running in parallel.")
+
+		s.sendResult(id, toolCallResult{
+			Content: []contentBlock{
+				{Type: "text", Text: sb.String()},
+			},
+		})
+
 	case "taskyou_list_tasks":
 		status, _ := params.Arguments["status"].(string)
 		project, _ := params.Arguments["project"].(string)
diff --git a/internal/mcp/server_test.go b/internal/mcp/server_test.go
index 964a469b..823b646a 100644
--- a/internal/mcp/server_test.go
+++ b/internal/mcp/server_test.go
@@ -101,6 +101,7 @@ func TestToolsList(t *testing.T) {
 		"taskyou_needs_input":         false,
 		"taskyou_show_task":           false,
 		"taskyou_create_task":         false,
+		"taskyou_create_pipeline":     false,
 		"taskyou_list_tasks":          false,
 		"taskyou_get_project_context": false,
 		"taskyou_set_project_context": false,
diff --git a/internal/pipeline/compose.go b/internal/pipeline/compose.go
new file mode 100644
index 00000000..dc9c1cc6
--- /dev/null
+++ b/internal/pipeline/compose.go
@@ -0,0 +1,125 @@
+package pipeline
+
+import (
+	"fmt"
+	"strings"
+)
+
+// effectiveInstruction returns the full body template for a step. Built-in steps
+// carry a hand-written Instruction and use it verbatim. Custom (YAML) steps carry
+// only a Prompt — what the step should do — and the git handoff is composed from
+// the step's position in the DAG, so authors never hand-write the commit/push/PR
+// boilerplate (and can't get the parallel-branch handling subtly wrong).
+func effectiveInstruction(def Definition, stepName string) string {
+	step, ok := def.step(stepName)
+	if !ok {
+		return ""
+	}
+	if strings.TrimSpace(step.Instruction) != "" {
+		return step.Instruction
+	}
+	return composeInstruction(def, step)
+}
+
+// step looks a step up by name.
+func (d Definition) step(name string) (Step, bool) {
+	for _, s := range d.Steps {
+		if s.Name == name {
+			return s, true
+		}
+	}
+	return Step{}, false
+}
+
+// dependents returns the steps that depend on the named step.
+func (d Definition) dependents(name string) []Step {
+	var out []Step
+	for _, s := range d.Steps {
+		for _, dep := range s.Deps {
+			if dep == name {
+				out = append(out, s)
+			}
+		}
+	}
+	return out
+}
+
+// hasParallelPeer reports whether the step runs at the same time as another step,
+// so its output must go to its own branch to avoid clobbering the peer. Two steps
+// are parallel if they share a dependency; multiple root steps (no deps) are all
+// parallel entry points.
+func (d Definition) hasParallelPeer(s Step) bool {
+	if len(s.Deps) == 0 {
+		return len(d.Roots()) > 1
+	}
+	depSet := make(map[string]bool, len(s.Deps))
+	for _, dep := range s.Deps {
+		depSet[dep] = true
+	}
+	for _, other := range d.Steps {
+		if other.Name == s.Name {
+			continue
+		}
+		for _, dep := range other.Deps {
+			if depSet[dep] {
+				return true
+			}
+		}
+	}
+	return false
+}
+
+// composeInstruction builds a custom step's body: an intro, the author's prompt,
+// then a git-handoff section derived from the DAG.
+//
+//   - A step is on the shared branch {{branch}}. If it has a parallel peer it
+//     pushes its output to its OWN branch ({{branch}}-{{stepslug}}) so peers can't
+//     clobber each other; otherwise it commits and pushes {{branch}}.
+//   - It reads each dependency that ran in parallel from that dep's own branch.
+//   - A sink step (nothing depends on it) opens the PR; every other step must not.
+func composeInstruction(def Definition, step Step) string {
+	var b strings.Builder
+
+	fmt.Fprintf(&b, "This is the **%s** step of the automated **%s** workflow, running on the shared branch `{{branch}}`.\n\n", step.Name, def.Name)
+	b.WriteString("## Goal\n{{goal}}\n\n")
+	b.WriteString("## Your task\n")
+	b.WriteString(strings.TrimSpace(step.Prompt))
+	b.WriteString("\n\n## Handoff (do this so the workflow can continue)\n")
+
+	// Inputs: dependencies that ran in parallel published to their own branches.
+	var parallelDeps []Step
+	for _, depName := range step.Deps {
+		if dep, ok := def.step(depName); ok && def.hasParallelPeer(dep) {
+			parallelDeps = append(parallelDeps, dep)
+		}
+	}
+	if len(parallelDeps) > 0 {
+		b.WriteString("- Your inputs were produced in parallel and pushed to their own branches; read each:\n")
+		for _, dep := range parallelDeps {
+			slug := slugify(dep.Name, 40)
+			fmt.Fprintf(&b, "    - **%s** → `git fetch origin && git show origin/{{branch}}-%s:<its output file>` (branch `{{branch}}-%s`)\n", dep.Name, slug, slug)
+		}
+	}
+
+	// Output: own branch when parallel, shared branch otherwise.
+	if def.hasParallelPeer(step) {
+		slug := slugify(step.Name, 40)
+		b.WriteString("- You run in parallel with a sibling step, so push your output to YOUR OWN branch (one commit, one push — no rebase, no clobber):\n")
+		fmt.Fprintf(&b, "    `git add <your output file> && git commit -m \"%s\" && git push origin HEAD:{{branch}}-%s`\n", slug, slug)
+		b.WriteString("  Do NOT push to `{{branch}}` itself.\n")
+	} else {
+		b.WriteString("- Commit your work and push the shared branch:\n")
+		b.WriteString("    `git add -A && git commit -m \"{{stepslug}}: ...\" && git push origin HEAD:{{branch}}`\n")
+		b.WriteString("  (If you are on a detached HEAD, run `git checkout -B {{branch}}` first.)\n")
+	}
+
+	// PR: only the sink step opens it.
+	if len(def.dependents(step.Name)) == 0 {
+		b.WriteString("- You are the final step: open a pull request with `gh pr create` summarizing the change. The workflow then parks in 'blocked' for a human to merge.\n")
+	} else {
+		b.WriteString("- Do NOT open a pull request — a later step does that.\n")
+	}
+	b.WriteString("- Then call `taskyou_complete` with a one-line summary. That advances the workflow.")
+
+	return b.String()
+}
diff --git a/internal/pipeline/custom_test.go b/internal/pipeline/custom_test.go
new file mode 100644
index 00000000..86f6aa61
--- /dev/null
+++ b/internal/pipeline/custom_test.go
@@ -0,0 +1,210 @@
+package pipeline
+
+import (
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"github.com/bborn/workflow/internal/db"
+)
+
+const sampleYAML = `
+name: build-and-qa
+description: build then review + qa in parallel
+steps:
+  - name: Plan
+    model: opus
+    prompt: Plan {{goal}}.
+  - name: Build
+    deps: [Plan]
+    prompt: Build it.
+  - name: Security
+    deps: [Build]
+    prompt: Security review.
+  - name: QA
+    executor: codex
+    deps: [Build]
+    prompt: QA it.
+  - name: Finalize
+    deps: [Security, QA]
+    prompt: Finalize.
+`
+
+func TestParseDefinition(t *testing.T) {
+	def, err := ParseDefinition([]byte(sampleYAML))
+	if err != nil {
+		t.Fatalf("ParseDefinition: %v", err)
+	}
+	if !def.Custom {
+		t.Error("custom def should be marked Custom")
+	}
+	if len(def.Steps) != 5 {
+		t.Fatalf("got %d steps, want 5", len(def.Steps))
+	}
+	// Executor defaults to claude; explicit executor honored.
+	if def.Steps[0].Executor != "claude" {
+		t.Errorf("Plan executor = %q, want claude default", def.Steps[0].Executor)
+	}
+	if s, _ := def.step("QA"); s.Executor != "codex" {
+		t.Errorf("QA executor = %q, want codex", s.Executor)
+	}
+	if err := def.validate(); err != nil {
+		t.Errorf("valid def failed validate: %v", err)
+	}
+}
+
+func TestParseDefinitionRejectsInvalid(t *testing.T) {
+	cases := map[string]string{
+		"no name":     "steps:\n  - name: A\n    prompt: x\n",
+		"no prompt":   "name: d\nsteps:\n  - name: A\n",
+		"unknown dep": "name: d\nsteps:\n  - name: A\n    prompt: x\n  - name: B\n    deps: [Z]\n    prompt: y\n",
+	}
+	for name, y := range cases {
+		t.Run(name, func(t *testing.T) {
+			if _, err := ParseDefinition([]byte(y)); err == nil {
+				t.Errorf("expected error for %s", name)
+			}
+		})
+	}
+}
+
+func TestMarshalRoundTrip(t *testing.T) {
+	def, err := ParseDefinition([]byte(sampleYAML))
+	if err != nil {
+		t.Fatal(err)
+	}
+	out, err := Marshal(def)
+	if err != nil {
+		t.Fatal(err)
+	}
+	def2, err := ParseDefinition(out)
+	if err != nil {
+		t.Fatalf("re-parse marshaled: %v", err)
+	}
+	if len(def2.Steps) != len(def.Steps) || def2.Name != def.Name {
+		t.Error("round-trip changed the definition")
+	}
+}
+
+func TestRegistryMergesCustom(t *testing.T) {
+	dir := t.TempDir()
+	t.Setenv("TY_WORKFLOWS_DIR", dir)
+	if err := os.WriteFile(filepath.Join(dir, "build-and-qa.yaml"), []byte(sampleYAML), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	// Custom appears alongside the built-in.
+	if _, ok := Get("build-and-qa"); !ok {
+		t.Error("custom workflow not resolved by Get")
+	}
+	if _, ok := Get(DefaultDefinition); !ok {
+		t.Error("built-in should still resolve")
+	}
+	names := DefinitionNames()
+	if !contains(names, "build-and-qa") || !contains(names, DefaultDefinition) {
+		t.Errorf("DefinitionNames = %v, want built-in + custom", names)
+	}
+}
+
+func TestComposeDerivesHandoffFromDAG(t *testing.T) {
+	def, err := ParseDefinition([]byte(sampleYAML))
+	if err != nil {
+		t.Fatal(err)
+	}
+	// Security runs parallel to QA → push to its own branch, no PR.
+	sec := effectiveInstruction(def, "Security")
+	if !strings.Contains(sec, "{{branch}}-security") || !strings.Contains(sec, "Do NOT open a pull request") {
+		t.Errorf("Security handoff wrong:\n%s", sec)
+	}
+	// Finalize joins the two parallel steps and is the sink → reads their branches + opens PR.
+	fin := effectiveInstruction(def, "Finalize")
+	if !strings.Contains(fin, "{{branch}}-security") || !strings.Contains(fin, "{{branch}}-qa") {
+		t.Errorf("Finalize should read both review branches:\n%s", fin)
+	}
+	if !strings.Contains(fin, "gh pr create") {
+		t.Errorf("Finalize (sink) should open a PR:\n%s", fin)
+	}
+	// Build is linear (no parallel peer, not sink) → shared-branch push, no PR.
+	build := effectiveInstruction(def, "Build")
+	if !strings.Contains(build, "push origin HEAD:{{branch}}") || strings.Contains(build, "gh pr create") {
+		t.Errorf("Build handoff wrong:\n%s", build)
+	}
+	// The author's prompt is carried through.
+	if !strings.Contains(build, "Build it.") {
+		t.Error("composed body should include the author's prompt")
+	}
+}
+
+func TestBuiltinUsesInstructionVerbatim(t *testing.T) {
+	def, _ := Get(DefaultDefinition)
+	got := effectiveInstruction(def, "Plan")
+	if got != planInstruction {
+		t.Error("built-in step should use its Instruction verbatim")
+	}
+}
+
+func TestCreateHonorsCustomWorkflow(t *testing.T) {
+	dir := t.TempDir()
+	t.Setenv("TY_WORKFLOWS_DIR", dir)
+	if err := os.WriteFile(filepath.Join(dir, "build-and-qa.yaml"), []byte(sampleYAML), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	database := testDB(t)
+	res, err := Create(database, Options{Goal: "do it", Project: "test", Definition: "build-and-qa"})
+	if err != nil {
+		t.Fatalf("Create custom: %v", err)
+	}
+	if len(res.Tasks) != 5 {
+		t.Fatalf("got %d tasks, want 5", len(res.Tasks))
+	}
+	// The QA step's configured executor (codex) flows through.
+	if got := taskByStep(res, "QA").Executor; got != db.ExecutorCodex {
+		t.Errorf("QA executor = %q, want codex", got)
+	}
+}
+
+const multiRootYAML = `
+name: three-spikes
+description: try three approaches at once, then pick and build
+steps:
+  - name: Spike A
+    prompt: Approach A to {{goal}}.
+  - name: Spike B
+    prompt: Approach B to {{goal}}.
+  - name: Spike C
+    prompt: Approach C to {{goal}}.
+  - name: Pick
+    deps: [Spike A, Spike B, Spike C]
+    prompt: Pick the best approach and build it.
+`
+
+func TestMultiRootWorkflow(t *testing.T) {
+	def, err := ParseDefinition([]byte(multiRootYAML))
+	if err != nil {
+		t.Fatalf("ParseDefinition (multi-root should be valid): %v", err)
+	}
+	if len(def.Roots()) != 3 {
+		t.Fatalf("got %d roots, want 3", len(def.Roots()))
+	}
+	// Each parallel root pushes to its own branch (they run at once).
+	spike := effectiveInstruction(def, "Spike A")
+	if !strings.Contains(spike, "{{branch}}-spike-a") || !strings.Contains(spike, "Do NOT open a pull request") {
+		t.Errorf("parallel root handoff wrong:\n%s", spike)
+	}
+	// The join reads all three root branches and (as sink) opens the PR.
+	pick := effectiveInstruction(def, "Pick")
+	for _, want := range []string{"{{branch}}-spike-a", "{{branch}}-spike-b", "{{branch}}-spike-c", "gh pr create"} {
+		if !strings.Contains(pick, want) {
+			t.Errorf("Pick handoff missing %q:\n%s", want, pick)
+		}
+	}
+}
+
+func contains(ss []string, want string) bool {
+	for _, s := range ss {
+		if s == want {
+			return true
+		}
+	}
+	return false
+}
diff --git a/internal/pipeline/definition_file.go b/internal/pipeline/definition_file.go
new file mode 100644
index 00000000..88570a76
--- /dev/null
+++ b/internal/pipeline/definition_file.go
@@ -0,0 +1,167 @@
+package pipeline
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+
+	"gopkg.in/yaml.v3"
+)
+
+// Custom workflows are authored as plain YAML files — one workflow per file —
+// so a user (or the LLM in `ty pipeline new`) can define a whole new flow (add a
+// QA step, a different shape) without touching Go. Files live in a global dir
+// (~/.config/task/workflows) and an optional per-project dir
+// (<project>/.taskyou/workflows); a project file shadows a global one of the same
+// name, which shadows a built-in.
+//
+// Authoring is just prompts: each step says what it does and what it depends on;
+// the git handoff (which branch to push to, when to open the PR) is derived from
+// the step's position in the DAG — see compose.go.
+
+// stepYAML is the on-disk form of a step.
+type stepYAML struct {
+	Name     string   `yaml:"name"`
+	Executor string   `yaml:"executor,omitempty"`
+	Model    string   `yaml:"model,omitempty"`
+	Deps     []string `yaml:"deps,omitempty"`
+	Prompt   string   `yaml:"prompt"`
+	// Verbatim marks a step whose prompt IS the full instruction (no DAG-derived
+	// git handoff is added). It's set when `ty pipeline edit` ejects a built-in
+	// workflow, so the ejected file behaves identically to the built-in.
+	Verbatim bool `yaml:"verbatim,omitempty"`
+}
+
+// definitionYAML is the on-disk form of a workflow.
+type definitionYAML struct {
+	Name        string     `yaml:"name"`
+	Description string     `yaml:"description,omitempty"`
+	Steps       []stepYAML `yaml:"steps"`
+}
+
+// WorkflowsDir returns the global directory custom workflow files live in.
+func WorkflowsDir() string {
+	if dir := os.Getenv("TY_WORKFLOWS_DIR"); dir != "" {
+		return dir
+	}
+	home, _ := os.UserHomeDir()
+	return filepath.Join(home, ".config", "task", "workflows")
+}
+
+// WorkflowDirs returns the directories to search for custom workflows: the global
+// dir plus the project-local .taskyou/workflows (when a project dir is given).
+// Later dirs win on name collisions.
+func WorkflowDirs(projectDir string) []string {
+	dirs := []string{WorkflowsDir()}
+	if projectDir != "" {
+		dirs = append(dirs, filepath.Join(projectDir, ".taskyou", "workflows"))
+	}
+	return dirs
+}
+
+// ParseDefinition parses and validates one workflow YAML document.
+func ParseDefinition(data []byte) (Definition, error) {
+	var doc definitionYAML
+	if err := yaml.Unmarshal(data, &doc); err != nil {
+		return Definition{}, fmt.Errorf("parse workflow yaml: %w", err)
+	}
+	if strings.TrimSpace(doc.Name) == "" {
+		return Definition{}, fmt.Errorf("workflow is missing a name")
+	}
+	def := Definition{
+		Name:        strings.TrimSpace(doc.Name),
+		Description: strings.TrimSpace(doc.Description),
+		Custom:      true,
+	}
+	for _, s := range doc.Steps {
+		name := strings.TrimSpace(s.Name)
+		if name == "" {
+			return Definition{}, fmt.Errorf("workflow %q has a step with no name", def.Name)
+		}
+		if strings.TrimSpace(s.Prompt) == "" {
+			return Definition{}, fmt.Errorf("step %q has no prompt", name)
+		}
+		exec := strings.TrimSpace(s.Executor)
+		if exec == "" {
+			exec = "claude"
+		}
+		step := Step{
+			Name:     name,
+			Executor: exec,
+			Model:    strings.TrimSpace(s.Model),
+			Deps:     s.Deps,
+		}
+		// A verbatim step's prompt is its full instruction; otherwise the prompt is
+		// the work and the git handoff is composed from the DAG.
+		if s.Verbatim {
+			step.Instruction = s.Prompt
+		} else {
+			step.Prompt = s.Prompt
+		}
+		def.Steps = append(def.Steps, step)
+	}
+	if err := def.validate(); err != nil {
+		return Definition{}, err
+	}
+	return def, nil
+}
+
+// Marshal renders a Definition back to YAML (used by `ty pipeline new`).
+func Marshal(def Definition) ([]byte, error) {
+	doc := definitionYAML{Name: def.Name, Description: def.Description}
+	for _, s := range def.Steps {
+		out := stepYAML{
+			Name:     s.Name,
+			Executor: s.Executor,
+			Model:    s.Model,
+			Deps:     s.Deps,
+			Prompt:   s.Prompt,
+		}
+		// A built-in step carries a full Instruction — write it as a verbatim
+		// prompt so the ejected file behaves identically when reloaded.
+		if s.Instruction != "" {
+			out.Prompt = s.Instruction
+			out.Verbatim = true
+		}
+		doc.Steps = append(doc.Steps, out)
+	}
+	return yaml.Marshal(doc)
+}
+
+// loadCustomDefinitions reads every *.yaml/*.yml workflow in the given dirs.
+// Later dirs shadow earlier ones by name. Unreadable or invalid files are
+// skipped and returned in the errs slice so callers can surface them without
+// failing the whole load.
+func loadCustomDefinitions(dirs []string) (map[string]Definition, []error) {
+	out := make(map[string]Definition)
+	var errs []error
+	for _, dir := range dirs {
+		entries, err := os.ReadDir(dir)
+		if err != nil {
+			continue // Missing dir is fine.
+		}
+		for _, e := range entries {
+			if e.IsDir() {
+				continue
+			}
+			ext := strings.ToLower(filepath.Ext(e.Name()))
+			if ext != ".yaml" && ext != ".yml" {
+				continue
+			}
+			path := filepath.Join(dir, e.Name())
+			data, err := os.ReadFile(path)
+			if err != nil {
+				errs = append(errs, fmt.Errorf("%s: %w", path, err))
+				continue
+			}
+			def, err := ParseDefinition(data)
+			if err != nil {
+				errs = append(errs, fmt.Errorf("%s: %w", path, err))
+				continue
+			}
+			out[def.Name] = def
+		}
+	}
+	return out, errs
+}
diff --git a/internal/pipeline/generate.go b/internal/pipeline/generate.go
new file mode 100644
index 00000000..2e8b0019
--- /dev/null
+++ b/internal/pipeline/generate.go
@@ -0,0 +1,170 @@
+package pipeline
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"os"
+	"strings"
+	"time"
+)
+
+// GenerateDefinition turns a free-text description of a workflow into a validated
+// Definition (and its YAML), using Claude. It's the engine behind
+// `ty pipeline new "<describe it>"`: the user describes the flow in English and
+// gets a ready-to-edit workflow file, instead of authoring YAML by hand.
+func GenerateDefinition(ctx context.Context, apiKey, description string) (Definition, []byte, error) {
+	if strings.TrimSpace(apiKey) == "" {
+		apiKey = os.Getenv("ANTHROPIC_API_KEY")
+	}
+	if strings.TrimSpace(apiKey) == "" {
+		return Definition{}, nil, fmt.Errorf("no Anthropic API key (set it in settings or ANTHROPIC_API_KEY)")
+	}
+	if strings.TrimSpace(description) == "" {
+		return Definition{}, nil, fmt.Errorf("describe the workflow you want")
+	}
+
+	raw, err := callClaude(ctx, apiKey, workflowGenPrompt(description))
+	if err != nil {
+		return Definition{}, nil, err
+	}
+	yamlDoc := extractYAML(raw)
+	def, err := ParseDefinition([]byte(yamlDoc))
+	if err != nil {
+		// One self-repair attempt: hand the invalid YAML and the exact error back
+		// to the model and ask it to fix it (models routinely fumble the single-root
+		// rule on parallel-first flows).
+		repaired, rerr := callClaude(ctx, apiKey, workflowRepairPrompt(yamlDoc, err))
+		if rerr == nil {
+			yamlDoc = extractYAML(repaired)
+			def, err = ParseDefinition([]byte(yamlDoc))
+		}
+		if err != nil {
+			return Definition{}, nil, fmt.Errorf("model produced an invalid workflow: %w\n---\n%s", err, yamlDoc)
+		}
+	}
+	// Re-marshal so the saved file is clean and canonical.
+	out, err := Marshal(def)
+	if err != nil {
+		return Definition{}, nil, err
+	}
+	return def, out, nil
+}
+
+func workflowGenPrompt(description string) string {
+	return `You design multi-step coding workflows as a DAG. Convert the user's description into a workflow YAML.
+
+Output ONLY the YAML (no prose, no markdown fences). Schema:
+
+name: <kebab-case-id>
+description: <one line>
+steps:
+  - name: <Step Name>
+    executor: <claude|codex|gemini|pi|opencode|openclaw>   # optional, default claude
+    model: <opus|sonnet|haiku>                             # optional, only meaningful for claude
+    deps: [<earlier step names>]                           # omit for the first step
+    prompt: |
+      What this step should DO (imperative). Reference the goal as {{goal}}.
+
+Rules:
+- Exactly ONE step has no deps (the single root/entry step). If the flow starts with parallel work (e.g. "try 3 approaches at once"), add ONE root step first (e.g. a "Kickoff" or "Plan" step that frames {{goal}}) that all the parallel steps depend on.
+- It is a DAG: deps must reference earlier steps; no cycles.
+- Two steps with the same deps run in PARALLEL; a step depending on several steps JOINs them.
+- Do NOT include any git/commit/push/PR/taskyou instructions in prompts — the system adds the branch handoff automatically. Prompts describe only the work.
+- Keep prompts concise and concrete. Prefer strong models (opus) for planning/review, faster ones (sonnet/haiku) for mechanical steps.
+
+User's description:
+` + description
+}
+
+func workflowRepairPrompt(badYAML string, err error) string {
+	return fmt.Sprintf(`This workflow YAML is invalid: %v
+
+Fix it and output ONLY the corrected YAML (no prose, no fences). Keep the same schema and intent. Remember: exactly ONE step may have no deps — if several steps were parallel entry points, add a single root step they all depend on.
+
+--- invalid yaml ---
+%s`, err, badYAML)
+}
+
+// extractYAML pulls the YAML body out of a model response, tolerating markdown
+// code fences the model may add despite instructions.
+func extractYAML(s string) string {
+	s = strings.TrimSpace(s)
+	if i := strings.Index(s, "```"); i >= 0 {
+		s = s[i+3:]
+		s = strings.TrimPrefix(s, "yaml")
+		s = strings.TrimPrefix(s, "yml")
+		if j := strings.Index(s, "```"); j >= 0 {
+			s = s[:j]
+		}
+	}
+	return strings.TrimSpace(s)
+}
+
+// --- minimal Anthropic client (mirrors internal/ai) ---
+
+type genMessage struct {
+	Role    string `json:"role"`
+	Content string `json:"content"`
+}
+
+type genRequest struct {
+	Model     string       `json:"model"`
+	MaxTokens int          `json:"max_tokens"`
+	Messages  []genMessage `json:"messages"`
+}
+
+type genContent struct {
+	Type string `json:"type"`
+	Text string `json:"text"`
+}
+
+type genResponse struct {
+	Content []genContent `json:"content"`
+	Error   *struct {
+		Message string `json:"message"`
+	} `json:"error"`
+}
+
+func callClaude(ctx context.Context, apiKey, prompt string) (string, error) {
+	body, err := json.Marshal(genRequest{
+		Model:     "claude-sonnet-4-5",
+		MaxTokens: 2000,
+		Messages:  []genMessage{{Role: "user", Content: prompt}},
+	})
+	if err != nil {
+		return "", err
+	}
+	req, err := http.NewRequestWithContext(ctx, "POST", "https://api.anthropic.com/v1/messages", bytes.NewReader(body))
+	if err != nil {
+		return "", err
+	}
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("x-api-key", apiKey)
+	req.Header.Set("anthropic-version", "2023-06-01")
+
+	client := &http.Client{Timeout: 60 * time.Second}
+	resp, err := client.Do(req)
+	if err != nil {
+		return "", err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		b, _ := io.ReadAll(resp.Body)
+		return "", fmt.Errorf("anthropic API %d: %s", resp.StatusCode, string(b))
+	}
+	var out genResponse
+	if err := json.NewDecoder(resp.Body).Decode(&out); err != nil {
+		return "", err
+	}
+	if out.Error != nil {
+		return "", fmt.Errorf("anthropic API: %s", out.Error.Message)
+	}
+	if len(out.Content) == 0 {
+		return "", fmt.Errorf("empty response from model")
+	}
+	return out.Content[0].Text, nil
+}
diff --git a/internal/pipeline/group.go b/internal/pipeline/group.go
new file mode 100644
index 00000000..94601590
--- /dev/null
+++ b/internal/pipeline/group.go
@@ -0,0 +1,212 @@
+package pipeline
+
+import (
+	"fmt"
+	"sort"
+	"strings"
+
+	"github.com/bborn/workflow/internal/db"
+)
+
+// A workflow's step tasks would otherwise each show as their own board card,
+// cluttering the board with N cards for one goal. Group collapses a workflow's
+// member tasks so the board can render a single card — the workflow's current
+// frontier — badged with progress.
+
+// Group is a workflow's member tasks, keyed by their shared branch.
+type Group struct {
+	Branch  string
+	Members []*db.Task // in step (creation) order
+}
+
+// groupKey is a task's workflow key: the shared branch it runs on. The root step
+// carries it as BranchName; every other step as SourceBranch.
+func groupKey(t *db.Task) string {
+	if s := strings.TrimSpace(t.SourceBranch); s != "" {
+		return s
+	}
+	return strings.TrimSpace(t.BranchName)
+}
+
+// IsWorkflowTask reports whether a task belongs to a workflow (tagged "pipeline"
+// with a shared branch).
+func IsWorkflowTask(t *db.Task) bool {
+	return t != nil && hasPipelineTag(t.Tags) && groupKey(t) != ""
+}
+
+func hasPipelineTag(tags string) bool {
+	for _, tag := range strings.Split(tags, ",") {
+		if strings.EqualFold(strings.TrimSpace(tag), "pipeline") {
+			return true
+		}
+	}
+	return false
+}
+
+// GroupWorkflows partitions tasks into workflow groups (2+ members sharing a
+// branch) and the remaining ungrouped tasks, preserving input order for the rest.
+func GroupWorkflows(tasks []*db.Task) (groups []*Group, rest []*db.Task) {
+	idx := make(map[string]*Group)
+	var order []string
+	for _, t := range tasks {
+		if !IsWorkflowTask(t) {
+			rest = append(rest, t)
+			continue
+		}
+		key := groupKey(t)
+		g := idx[key]
+		if g == nil {
+			g = &Group{Branch: key}
+			idx[key] = g
+			order = append(order, key)
+		}
+		g.Members = append(g.Members, t)
+	}
+	for _, key := range order {
+		g := idx[key]
+		// A lone member isn't a workflow to collapse — render it normally.
+		if len(g.Members) < 2 {
+			rest = append(rest, g.Members...)
+			continue
+		}
+		sort.Slice(g.Members, func(i, j int) bool { return g.Members[i].ID < g.Members[j].ID })
+		groups = append(groups, g)
+	}
+	return groups, rest
+}
+
+// statusPriority ranks a member's status by how "live" it is, so Lead() surfaces
+// the workflow's current frontier.
+func statusPriority(status string) int {
+	switch status {
+	case db.StatusProcessing:
+		return 5
+	case db.StatusQueued:
+		return 4
+	case db.StatusBlocked:
+		return 3
+	case db.StatusBacklog:
+		return 2
+	case db.StatusDone:
+		return 1
+	default:
+		return 0
+	}
+}
+
+// Lead returns the member that best represents the workflow's current state — the
+// most-live step (processing > queued > blocked > backlog > done), lowest id on a
+// tie. The board renders this task's card for the whole workflow.
+func (g *Group) Lead() *db.Task {
+	if len(g.Members) == 0 {
+		return nil
+	}
+	lead := g.Members[0]
+	for _, t := range g.Members[1:] {
+		if statusPriority(t.Status) > statusPriority(lead.Status) {
+			lead = t
+		}
+	}
+	return lead
+}
+
+// Total returns the number of steps.
+func (g *Group) Total() int { return len(g.Members) }
+
+// DoneCount returns how many steps have completed.
+func (g *Group) DoneCount() int {
+	n := 0
+	for _, t := range g.Members {
+		if t.Status == db.StatusDone {
+			n++
+		}
+	}
+	return n
+}
+
+// Goal returns the workflow goal, recovered by stripping the "[Step] " prefix off
+// a member title.
+func (g *Group) Goal() string {
+	if len(g.Members) == 0 {
+		return ""
+	}
+	return stripStepPrefix(g.Members[0].Title)
+}
+
+// ActiveSteps returns the step names currently processing or queued (there can be
+// two during the parallel review fan-out).
+func (g *Group) ActiveSteps() []string {
+	var names []string
+	for _, t := range g.Members {
+		if t.Status == db.StatusProcessing || t.Status == db.StatusQueued {
+			if n := stepName(t.Title); n != "" {
+				names = append(names, n)
+			}
+		}
+	}
+	return names
+}
+
+// StepLabel is a short, board-friendly description of the workflow's current
+// position: the active step name, "<prefix> ∥" when several steps run in parallel
+// (e.g. "Review ∥"), or "done" when every step has completed.
+func (g *Group) StepLabel() string {
+	active := g.ActiveSteps()
+	switch len(active) {
+	case 0:
+		if g.DoneCount() >= g.Total() {
+			return "done"
+		}
+		if lead := g.Lead(); lead != nil {
+			if n := stepName(lead.Title); n != "" {
+				return n
+			}
+		}
+		return "waiting"
+	case 1:
+		return active[0]
+	default:
+		// Several steps run at once: collapse to their shared first word when
+		// they have one (e.g. "Review A"/"Review B" → "Review ∥").
+		prefix := firstWord(active[0])
+		for _, a := range active[1:] {
+			if firstWord(a) != prefix {
+				prefix = ""
+				break
+			}
+		}
+		if prefix != "" {
+			return prefix + " ∥"
+		}
+		return fmt.Sprintf("%d ∥", len(active))
+	}
+}
+
+func firstWord(s string) string {
+	if f := strings.Fields(s); len(f) > 0 {
+		return f[0]
+	}
+	return ""
+}
+
+// stepName extracts the step label from a title like "[Plan] add rate limiting".
+func stepName(title string) string {
+	title = strings.TrimSpace(title)
+	if strings.HasPrefix(title, "[") {
+		if end := strings.Index(title, "]"); end > 1 {
+			return title[1:end]
+		}
+	}
+	return ""
+}
+
+// stripStepPrefix removes the leading "[Step] " from a title.
+func stripStepPrefix(title string) string {
+	title = strings.TrimSpace(title)
+	if strings.HasPrefix(title, "[") {
+		if end := strings.Index(title, "]"); end > 1 {
+			return strings.TrimSpace(title[end+1:])
+		}
+	}
+	return title
+}
diff --git a/internal/pipeline/group_test.go b/internal/pipeline/group_test.go
new file mode 100644
index 00000000..c30070aa
--- /dev/null
+++ b/internal/pipeline/group_test.go
@@ -0,0 +1,100 @@
+package pipeline
+
+import (
+	"testing"
+
+	"github.com/bborn/workflow/internal/db"
+)
+
+func TestGroupWorkflowsCollapsesMembers(t *testing.T) {
+	database := testDB(t)
+	res, err := Create(database, Options{Goal: "Build a widget", Project: "test", Execute: true})
+	if err != nil {
+		t.Fatalf("Create: %v", err)
+	}
+	// Reload from DB so SourceBranch/BranchName reflect what's persisted.
+	all := make([]*db.Task, 0, len(res.Tasks)+1)
+	for _, tk := range res.Tasks {
+		reloaded, _ := database.GetTask(tk.ID)
+		all = append(all, reloaded)
+	}
+	// A plain, unrelated task should not be grouped.
+	plain := &db.Task{Title: "unrelated", Status: db.StatusBacklog, Type: db.TypeCode, Project: "test"}
+	must(t, database.CreateTask(plain))
+	plain, _ = database.GetTask(plain.ID)
+	all = append(all, plain)
+
+	groups, rest := GroupWorkflows(all)
+	if len(groups) != 1 {
+		t.Fatalf("got %d groups, want 1", len(groups))
+	}
+	if len(rest) != 1 || rest[0].Title != "unrelated" {
+		t.Errorf("rest = %v, want [unrelated]", rest)
+	}
+	g := groups[0]
+	if g.Total() != 5 {
+		t.Errorf("group total = %d, want 5", g.Total())
+	}
+	if g.Goal() != "Build a widget" {
+		t.Errorf("group goal = %q, want 'Build a widget'", g.Goal())
+	}
+	// Right after Create, the root (Plan) is queued → it's the lead.
+	if lead := g.Lead(); lead == nil || stepName(lead.Title) != "Plan" {
+		t.Errorf("lead = %v, want Plan", lead)
+	}
+	if g.DoneCount() != 0 {
+		t.Errorf("done = %d, want 0", g.DoneCount())
+	}
+}
+
+func TestGroupLeadPrefersMostLiveStep(t *testing.T) {
+	// Plan done, Code processing, reviews blocked → lead is Code (most live).
+	members := []*db.Task{
+		{ID: 1, Title: "[Plan] g", Status: db.StatusDone, BranchName: "pipeline/1-g"},
+		{ID: 2, Title: "[Code] g", Status: db.StatusProcessing, SourceBranch: "pipeline/1-g"},
+		{ID: 3, Title: "[Review A] g", Status: db.StatusBlocked, SourceBranch: "pipeline/1-g"},
+		{ID: 4, Title: "[Review B] g", Status: db.StatusBlocked, SourceBranch: "pipeline/1-g"},
+	}
+	for i := range members {
+		members[i].Tags = "pipeline"
+	}
+	groups, _ := GroupWorkflows(members)
+	if len(groups) != 1 {
+		t.Fatalf("got %d groups, want 1", len(groups))
+	}
+	g := groups[0]
+	if lead := g.Lead(); stepName(lead.Title) != "Code" {
+		t.Errorf("lead = %q, want Code", lead.Title)
+	}
+	if g.DoneCount() != 1 {
+		t.Errorf("done = %d, want 1", g.DoneCount())
+	}
+}
+
+func TestGroupParallelReviewActiveSteps(t *testing.T) {
+	members := []*db.Task{
+		{ID: 1, Title: "[Plan] g", Status: db.StatusDone, Tags: "pipeline", BranchName: "pipeline/1-g"},
+		{ID: 2, Title: "[Code] g", Status: db.StatusDone, Tags: "pipeline", SourceBranch: "pipeline/1-g"},
+		{ID: 3, Title: "[Review A] g", Status: db.StatusProcessing, Tags: "pipeline", SourceBranch: "pipeline/1-g"},
+		{ID: 4, Title: "[Review B] g", Status: db.StatusProcessing, Tags: "pipeline", SourceBranch: "pipeline/1-g"},
+		{ID: 5, Title: "[Collect] g", Status: db.StatusBlocked, Tags: "pipeline", SourceBranch: "pipeline/1-g"},
+	}
+	groups, _ := GroupWorkflows(members)
+	g := groups[0]
+	active := g.ActiveSteps()
+	if len(active) != 2 {
+		t.Fatalf("active steps = %v, want 2 (both reviewers)", active)
+	}
+}
+
+func TestIsWorkflowTaskRequiresTagAndBranch(t *testing.T) {
+	if IsWorkflowTask(&db.Task{Tags: "pipeline"}) {
+		t.Error("no branch → not a workflow task")
+	}
+	if IsWorkflowTask(&db.Task{BranchName: "pipeline/1-x"}) {
+		t.Error("no pipeline tag → not a workflow task")
+	}
+	if !IsWorkflowTask(&db.Task{Tags: "a,pipeline,b", SourceBranch: "pipeline/1-x"}) {
+		t.Error("tagged + branch → should be a workflow task")
+	}
+}
diff --git a/internal/pipeline/instructions.go b/internal/pipeline/instructions.go
new file mode 100644
index 00000000..92d6c54f
--- /dev/null
+++ b/internal/pipeline/instructions.go
@@ -0,0 +1,78 @@
+package pipeline
+
+// Step instruction templates. Each is a full task body: the daemon wraps it in
+// the usual TaskYou harness (worktree constraints, taskyou_complete guidance), so
+// these only add the step-specific job. {{goal}}, {{branch}}, {{step}} and
+// {{stepslug}} are substituted at build time.
+//
+// Steps hand work forward through one shared git branch ({{branch}}): each step
+// commits and pushes, and the next step checks the branch out (via SourceBranch).
+// Only the terminal Collect step opens a pull request — earlier steps must NOT,
+// because a PR parks a task in 'blocked' and would stall the DAG (a step only
+// advances the workflow when it reaches 'done').
+
+const planInstruction = `This is the **Plan** step of an automated plan → code → review → collect workflow. Later steps run on the shared branch ` + "`{{branch}}`" + ` — you are the root, and you plan only.
+
+## Goal
+{{goal}}
+
+## Your job (planning only — do not write application code)
+1. Explore the codebase enough to design a concrete, staged implementation plan for the goal above.
+2. Write the plan to ` + "`PLAN.md`" + ` at the repo root: the approach, the exact files to change, the ordered steps, edge cases, and how to test it.
+3. Commit and push so the next step can pick it up:
+   ` + "`git add PLAN.md && git commit -m \"plan: {{goal}}\" && git push -u origin {{branch}}`" + `
+   (If you are on a detached HEAD, run ` + "`git checkout -B {{branch}}`" + ` first.)
+
+Do NOT implement the feature. Do NOT open a pull request.
+
+When ` + "`PLAN.md`" + ` is committed and pushed, call ` + "`taskyou_complete`" + ` with a one-line summary. That automatically starts the Code step.`
+
+const codeInstruction = `This is the **Code** step of an automated plan → code → review → collect workflow. The Plan step committed ` + "`PLAN.md`" + ` on this branch (` + "`{{branch}}`" + `); two independent reviewers run after you.
+
+## Goal
+{{goal}}
+
+## Your job
+1. Read ` + "`PLAN.md`" + ` and implement the plan fully. If the plan is wrong or incomplete, use your judgment and note the deviation in your commit message.
+2. Keep the change focused on the goal. Add or update tests where it makes sense.
+3. Commit and push:
+   ` + "`git add -A && git commit -m \"...\" && git push`" + `
+   (If you are on a detached HEAD, run ` + "`git checkout -B {{branch}}`" + ` first.)
+
+Do NOT open a pull request — the Collect step does that.
+
+When your implementation is committed and pushed, call ` + "`taskyou_complete`" + ` with a summary of what you built. That automatically starts the review steps.`
+
+const reviewInstruction = `This is the **{{step}}** step of an automated plan → code → review → collect workflow. You are ONE of two reviewers running in parallel with independent context — that independence is the point, so review the change on your own merits and don't assume the other reviewer will catch what you miss.
+
+## Goal
+{{goal}}
+
+## Your job (review only — do not change application code)
+1. Review the diff of this branch against the default branch:
+   ` + "`git fetch origin && git diff origin/HEAD...HEAD`" + ` (or ` + "`git diff $(git merge-base HEAD origin/HEAD)...HEAD`" + `).
+   Check for correctness bugs, missed requirements from ` + "`PLAN.md`" + `, security issues, and clear quality problems.
+2. Write your findings to ` + "`review-{{stepslug}}.md`" + ` at the repo root — a short prioritized list (blocking issues first, then nits), each with file:line and a concrete suggestion. If it looks good, say so explicitly.
+3. Commit that file and push it to YOUR OWN review branch. You each use a separate branch, so there is no rebase and no way to clobber the other reviewer — just one commit and one push:
+   ` + "`git add review-{{stepslug}}.md && git commit -m \"review: {{stepslug}}\" && git push origin HEAD:{{branch}}-{{stepslug}}`" + `
+   (If you are on a detached HEAD, that push command still works as-is.)
+
+Do NOT modify application code. Do NOT push to ` + "`{{branch}}`" + ` itself. Do NOT open a pull request.
+
+When your review branch is pushed, call ` + "`taskyou_complete`" + ` with a one-line verdict. The Collect step runs once BOTH reviewers finish.`
+
+const collectInstruction = `This is the **Collect** step of an automated plan → code → review → collect workflow, and the terminal step. Two independent reviewers have each pushed their review to a separate branch.
+
+## Goal
+{{goal}}
+
+## Your job
+1. Get onto the code branch and fetch the reviews:
+   ` + "`git fetch origin && git checkout -B {{branch}} origin/{{branch}}`" + `
+   The reviewers pushed their reviews to these branches:
+{{reviews}}
+   Read each one, e.g. ` + "`git show origin/{{branch}}-review-a:review-review-a.md`" + `.
+2. Address the findings: apply the fixes you agree are correct and safe, keeping changes focused on the code branch. If the reviewers disagree, use your judgment and note the call. Skip anything out of scope and record why.
+3. Commit your fixes and push ` + "`{{branch}}`" + `.
+4. Open a pull request with ` + "`gh pr create`" + `. In the body, summarize the change, then list what each review flagged and how you resolved it (fixed / deferred / disagreed).
+5. Call ` + "`taskyou_complete`" + `. Because a PR now exists, the workflow parks in 'blocked' for a human to review and merge — that is expected and completes the workflow.`
diff --git a/internal/pipeline/pipeline.go b/internal/pipeline/pipeline.go
new file mode 100644
index 00000000..5525b3fb
--- /dev/null
+++ b/internal/pipeline/pipeline.go
@@ -0,0 +1,437 @@
+// Package pipeline builds multi-step "workflow" tasks: a single goal is broken
+// into a small DAG of step tasks that run on one shared git branch and advance
+// automatically. Steps can be sequential (Code waits for Plan) or parallel (two
+// reviewers run at once), and a step can join several predecessors (Collect waits
+// for both reviews) — exactly the plan → code → parallel-review → collect shape
+// the herdr plugin popularized.
+//
+// It's assembled from primitives TaskYou already has, so there's no bespoke
+// execution engine — the normal daemon runs each step task:
+//
+//   - Per-task Executor + Model overrides give each step its own agent.
+//   - Task dependencies with auto_queue encode the DAG: a step is 'blocked' until
+//     all its dependencies complete, then db.ProcessCompletedBlocker queues it.
+//     Fan-out (two steps depending on one) and join (one step depending on two)
+//     both fall out of the dependency graph for free.
+//   - The root step pins a shared BranchName; every other step checks that branch
+//     out via SourceBranch, so steps hand work forward through git.
+//
+// The workflow advances with no human in the loop; it only pauses when a step
+// genuinely needs one — the terminal step opens a PR (landing in 'blocked' for a
+// human merge) or a step calls taskyou_needs_input.
+package pipeline
+
+import (
+	"fmt"
+	"os/exec"
+	"sort"
+	"strings"
+	"unicode"
+
+	"github.com/bborn/workflow/internal/db"
+)
+
+// Step is one node of a workflow DAG. Executor and Model pick the agent that runs
+// it; Instruction is the task body template ({{goal}} and {{branch}} are
+// substituted at build time); Deps names the steps that must finish first.
+type Step struct {
+	Name        string   // Unique label, e.g. "Plan", "Code", "Review A", "Collect".
+	Executor    string   // Executor slug (db.ExecutorClaude, db.ExecutorCodex, ...).
+	Model       string   // Per-task model override ("" = the executor's default).
+	Instruction string   // Full body template (built-in / verbatim steps). Takes precedence over Prompt.
+	Prompt      string   // Custom-workflow body: what the step does; the git handoff is composed from the DAG (see compose.go).
+	Deps        []string // Names of steps that must complete before this one runs.
+}
+
+// Definition is a named workflow DAG.
+type Definition struct {
+	Name        string
+	Description string
+	Steps       []Step
+	Custom      bool // true for workflows loaded from YAML files (vs the built-ins)
+}
+
+// DefaultDefinition is the definition used when none is named.
+const DefaultDefinition = "plan-code-review"
+
+// planCodeReview is the herdr flow as a DAG: Plan → Code → two independent
+// reviewers in parallel → Collect. The parallel reviewers are the point — two
+// agents with independent contexts catch different classes of issue and avoid
+// single-context self-review bias.
+//
+// Defaults are all Claude (so it runs anywhere with Claude and never silently
+// swaps executors), but each step's executor/model is configurable per project
+// (see config.go): point one reviewer at codex — `pipeline config --set
+// "Review B=codex"` — for the herdr-style cross-executor review, or add a QA step.
+var planCodeReview = Definition{
+	Name:        "plan-code-review",
+	Description: "Plan → code → two parallel reviewers → collect, on one shared branch. Each step's model/executor is configurable per project.",
+	Steps: []Step{
+		{Name: "Plan", Executor: db.ExecutorClaude, Model: db.ModelOpus, Instruction: planInstruction},
+		{Name: "Code", Executor: db.ExecutorClaude, Model: db.ModelSonnet, Instruction: codeInstruction, Deps: []string{"Plan"}},
+		{Name: "Review A", Executor: db.ExecutorClaude, Model: db.ModelOpus, Instruction: reviewInstruction, Deps: []string{"Code"}},
+		{Name: "Review B", Executor: db.ExecutorClaude, Model: db.ModelSonnet, Instruction: reviewInstruction, Deps: []string{"Code"}},
+		{Name: "Collect", Executor: db.ExecutorClaude, Model: db.ModelSonnet, Instruction: collectInstruction, Deps: []string{"Review A", "Review B"}},
+	},
+}
+
+var builtins = map[string]Definition{
+	planCodeReview.Name: planCodeReview,
+}
+
+// registry merges the built-in definitions with any custom YAML workflows found
+// in extraDirs (plus the global WorkflowsDir). Custom definitions shadow built-ins
+// of the same name.
+func registry(extraDirs ...string) map[string]Definition {
+	out := make(map[string]Definition, len(builtins)+2)
+	for name, def := range builtins {
+		out[name] = def
+	}
+	dirs := append([]string{WorkflowsDir()}, extraDirs...)
+	custom, _ := loadCustomDefinitions(dirs)
+	for name, def := range custom {
+		out[name] = def
+	}
+	return out
+}
+
+// Definitions returns all workflow definitions — built-in plus custom YAML
+// workflows discovered in extraDirs and the global workflows dir.
+func Definitions(extraDirs ...string) []Definition {
+	reg := registry(extraDirs...)
+	names := make([]string, 0, len(reg))
+	for name := range reg {
+		names = append(names, name)
+	}
+	sort.Strings(names)
+	// Keep the default first for a stable, friendly listing.
+	out := make([]Definition, 0, len(names))
+	if def, ok := reg[DefaultDefinition]; ok {
+		out = append(out, def)
+	}
+	for _, name := range names {
+		if name == DefaultDefinition {
+			continue
+		}
+		out = append(out, reg[name])
+	}
+	return out
+}
+
+// DefinitionNames returns the names of all definitions (built-in + custom).
+func DefinitionNames(extraDirs ...string) []string {
+	defs := Definitions(extraDirs...)
+	out := make([]string, len(defs))
+	for i, d := range defs {
+		out[i] = d.Name
+	}
+	return out
+}
+
+// Get returns the named definition (built-in or custom). An empty name resolves
+// to DefaultDefinition.
+func Get(name string, extraDirs ...string) (Definition, bool) {
+	name = strings.TrimSpace(name)
+	if name == "" {
+		name = DefaultDefinition
+	}
+	def, ok := registry(extraDirs...)[name]
+	return def, ok
+}
+
+// Roots returns the steps with no dependencies (the entry points).
+func (d Definition) Roots() []Step {
+	var roots []Step
+	for _, s := range d.Steps {
+		if len(s.Deps) == 0 {
+			roots = append(roots, s)
+		}
+	}
+	return roots
+}
+
+// validate checks the DAG is well-formed: unique step names, deps that reference
+// real steps, exactly one root (the branch owner), and no cycles.
+func (d Definition) validate() error {
+	if len(d.Steps) == 0 {
+		return fmt.Errorf("workflow %q has no steps", d.Name)
+	}
+	byName := make(map[string]Step, len(d.Steps))
+	for _, s := range d.Steps {
+		if _, dup := byName[s.Name]; dup {
+			return fmt.Errorf("workflow %q has duplicate step %q", d.Name, s.Name)
+		}
+		byName[s.Name] = s
+	}
+	for _, s := range d.Steps {
+		for _, dep := range s.Deps {
+			if _, ok := byName[dep]; !ok {
+				return fmt.Errorf("step %q depends on unknown step %q", s.Name, dep)
+			}
+		}
+	}
+	if len(d.Roots()) == 0 {
+		return fmt.Errorf("workflow %q has no root step (every step depends on another)", d.Name)
+	}
+	if err := d.checkAcyclic(byName); err != nil {
+		return err
+	}
+	return nil
+}
+
+func (d Definition) checkAcyclic(byName map[string]Step) error {
+	const (
+		white = 0
+		gray  = 1
+		black = 2
+	)
+	color := make(map[string]int, len(byName))
+	var visit func(name string) error
+	visit = func(name string) error {
+		switch color[name] {
+		case gray:
+			return fmt.Errorf("workflow %q has a dependency cycle at %q", d.Name, name)
+		case black:
+			return nil
+		}
+		color[name] = gray
+		for _, dep := range byName[name].Deps {
+			if err := visit(dep); err != nil {
+				return err
+			}
+		}
+		color[name] = black
+		return nil
+	}
+	for name := range byName {
+		if err := visit(name); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// Options configures a workflow build.
+type Options struct {
+	Goal           string // The overall goal, threaded into every step's prompt.
+	Project        string // Project the step tasks belong to (must already exist).
+	Definition     string // Definition name; "" resolves to DefaultDefinition.
+	PermissionMode string // Permission mode for every step ("" inherits project default).
+	Execute        bool   // If true, queue the root step so the workflow starts now.
+}
+
+// Result describes a built workflow.
+type Result struct {
+	Definition Definition
+	Branch     string     // The shared branch every step runs on.
+	Tasks      []*db.Task // Step tasks, in definition order.
+}
+
+// Create builds a workflow: one task per step, wired into a dependency DAG on a
+// shared branch. The root step is created (and, with Execute, queued) as a normal
+// task; every other step is created 'blocked' with dependencies on its
+// predecessors, so db.ProcessCompletedBlocker queues it once they all complete.
+func Create(database *db.DB, opts Options) (*Result, error) {
+	goal := strings.TrimSpace(opts.Goal)
+	if goal == "" {
+		return nil, fmt.Errorf("workflow goal is required")
+	}
+	if strings.TrimSpace(opts.Project) == "" {
+		return nil, fmt.Errorf("workflow project is required")
+	}
+	projectDir := projectDirFor(database, opts.Project)
+	dirs := WorkflowDirs(projectDir)
+	def, ok := Get(opts.Definition, dirs...)
+	if !ok {
+		return nil, fmt.Errorf("unknown workflow definition %q (available: %s)", opts.Definition, strings.Join(DefinitionNames(dirs...), ", "))
+	}
+	if err := def.validate(); err != nil {
+		return nil, err
+	}
+
+	steps := def.Steps
+	roots := def.Roots()
+	rootNames := make(map[string]bool, len(roots))
+	for _, r := range roots {
+		rootNames[r.Name] = true
+	}
+	multiRoot := len(roots) > 1
+	slug := slugify(goal, 40)
+
+	// Create every step task first (in 'backlog', so the daemon leaves them alone),
+	// recording name → task so we can wire dependencies and pin the branch.
+	byName := make(map[string]*db.Task, len(steps))
+	tasks := make([]*db.Task, 0, len(steps))
+	for _, s := range steps {
+		task := &db.Task{
+			Title:          stepTitle(s.Name, goal),
+			Body:           goal, // Placeholder; rewritten once the branch is known.
+			Status:         db.StatusBacklog,
+			Type:           db.TypeCode,
+			Project:        opts.Project,
+			Executor:       s.Executor,
+			Model:          s.Model,
+			PermissionMode: opts.PermissionMode,
+			Tags:           "pipeline",
+		}
+		if err := database.CreateTask(task); err != nil {
+			return nil, fmt.Errorf("create %s step: %w", s.Name, err)
+		}
+		byName[s.Name] = task
+		tasks = append(tasks, task)
+	}
+
+	// Seed the shared branch from the first task's id so it's unique.
+	branch := fmt.Sprintf("pipeline/%d-%s", tasks[0].ID, slug)
+
+	// Branch ownership. With a single root, that root creates the branch (pins
+	// BranchName) and everyone else checks it out via SourceBranch — the proven
+	// path. With MULTIPLE roots (true parallel entry points), no single step can
+	// own the branch, so we pre-create it on the remote and every step (roots
+	// included) checks it out; the parallel roots each push to their own branch.
+	if multiRoot {
+		if err := ensureSharedBranch(projectDir, branch); err != nil {
+			for _, t := range tasks {
+				_ = database.DeleteTask(t.ID)
+			}
+			return nil, fmt.Errorf("multi-root workflows need a git-worktree project with a remote: %w", err)
+		}
+	}
+
+	for _, s := range steps {
+		task := byName[s.Name]
+		task.Body = render(effectiveInstruction(def, s.Name), goal, branch, s.Name, reviewsList(s, branch))
+		if rootNames[s.Name] && !multiRoot {
+			task.BranchName = branch // Single root pins/creates the branch.
+		} else {
+			task.SourceBranch = branch // Checked out from the shared branch.
+		}
+		if err := database.UpdateTask(task); err != nil {
+			return nil, fmt.Errorf("configure %s step: %w", s.Name, err)
+		}
+	}
+
+	// Wire the DAG: each dependency blocks the dependent, auto-queuing it on completion.
+	for _, s := range steps {
+		for _, dep := range s.Deps {
+			if err := database.AddDependency(byName[dep].ID, byName[s.Name].ID, true); err != nil {
+				return nil, fmt.Errorf("wire %s → %s: %w", dep, s.Name, err)
+			}
+		}
+	}
+
+	// Set statuses last, once the graph is fully wired: non-root steps wait
+	// 'blocked' on their deps; every root starts (queued) or stages (backlog).
+	for _, s := range steps {
+		if rootNames[s.Name] {
+			continue
+		}
+		if err := database.UpdateTaskStatus(byName[s.Name].ID, db.StatusBlocked); err != nil {
+			return nil, fmt.Errorf("block %s step: %w", s.Name, err)
+		}
+		byName[s.Name].Status = db.StatusBlocked
+	}
+	if opts.Execute {
+		for _, r := range roots {
+			if err := database.UpdateTaskStatus(byName[r.Name].ID, db.StatusQueued); err != nil {
+				return nil, fmt.Errorf("queue %s step: %w", r.Name, err)
+			}
+			byName[r.Name].Status = db.StatusQueued
+		}
+	}
+
+	return &Result{Definition: def, Branch: branch, Tasks: tasks}, nil
+}
+
+// ensureSharedBranch creates the workflow branch on the project's origin remote
+// (from the project's current HEAD) so that several parallel root steps can all
+// check it out at once. A branch that already exists is fine.
+func ensureSharedBranch(projectDir, branch string) error {
+	if strings.TrimSpace(projectDir) == "" {
+		return fmt.Errorf("project directory not found")
+	}
+	cmd := exec.Command("git", "-C", projectDir, "push", "origin", "HEAD:refs/heads/"+branch)
+	if out, err := cmd.CombinedOutput(); err != nil {
+		if strings.Contains(string(out), "already exists") {
+			return nil
+		}
+		return fmt.Errorf("git push %s: %v: %s", branch, err, strings.TrimSpace(string(out)))
+	}
+	return nil
+}
+
+// projectDirFor returns a project's on-disk path (used to find its
+// .taskyou/workflows dir), or "" if it can't be resolved.
+func projectDirFor(database *db.DB, project string) string {
+	if database == nil || project == "" {
+		return ""
+	}
+	if p, err := database.GetProjectByName(project); err == nil && p != nil {
+		return p.Path
+	}
+	return ""
+}
+
+// stepTitle builds a task title like "[Plan] <goal>", trimming a long goal so the
+// board card stays readable.
+func stepTitle(step, goal string) string {
+	const maxGoal = 60
+	g := strings.TrimSpace(goal)
+	if len(g) > maxGoal {
+		g = strings.TrimSpace(g[:maxGoal]) + "…"
+	}
+	return fmt.Sprintf("[%s] %s", step, g)
+}
+
+// render substitutes the {{goal}}, {{branch}}, {{step}}, {{stepslug}} and
+// {{reviews}} placeholders in a step template.
+func render(tmpl, goal, branch, step, reviews string) string {
+	r := strings.NewReplacer(
+		"{{goal}}", goal,
+		"{{branch}}", branch,
+		"{{step}}", step,
+		"{{stepslug}}", slugify(step, 40),
+		"{{reviews}}", reviews,
+	)
+	return strings.TrimSpace(r.Replace(tmpl))
+}
+
+// reviewsList describes, for a step, where each of its dependency steps pushed its
+// review: one bullet per dep naming the review branch and file. It's substituted
+// into the Collect step's {{reviews}} placeholder so Collect reads exactly the
+// reviewers that fed it, regardless of how the workflow is configured.
+func reviewsList(step Step, branch string) string {
+	var b strings.Builder
+	for _, dep := range step.Deps {
+		s := slugify(dep, 40)
+		fmt.Fprintf(&b, "     - `%s-%s` → file `review-%s.md`\n", branch, s, s)
+	}
+	return strings.TrimRight(b.String(), "\n")
+}
+
+// slugify converts a string into a lowercase, dash-separated slug, truncated to
+// maxLen. It mirrors the executor's worktree slug so workflow branch names read
+// like ordinary task branches.
+func slugify(s string, maxLen int) string {
+	s = strings.ToLower(strings.TrimSpace(s))
+	var b strings.Builder
+	lastDash := false
+	for _, r := range s {
+		switch {
+		case unicode.IsLetter(r) || unicode.IsDigit(r):
+			b.WriteRune(r)
+			lastDash = false
+		case !lastDash && b.Len() > 0:
+			b.WriteRune('-')
+			lastDash = true
+		}
+	}
+	slug := strings.Trim(b.String(), "-")
+	if maxLen > 0 && len(slug) > maxLen {
+		slug = strings.Trim(slug[:maxLen], "-")
+	}
+	if slug == "" {
+		slug = "pipeline"
+	}
+	return slug
+}
diff --git a/internal/pipeline/pipeline_test.go b/internal/pipeline/pipeline_test.go
new file mode 100644
index 00000000..e4388d44
--- /dev/null
+++ b/internal/pipeline/pipeline_test.go
@@ -0,0 +1,271 @@
+package pipeline
+
+import (
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"github.com/bborn/workflow/internal/db"
+)
+
+func testDB(t *testing.T) *db.DB {
+	t.Helper()
+	tmpDir := t.TempDir()
+	database, err := db.Open(filepath.Join(tmpDir, "test.db"))
+	if err != nil {
+		t.Fatalf("open db: %v", err)
+	}
+	t.Cleanup(func() { database.Close() })
+	if err := database.CreateProject(&db.Project{Name: "test", Path: tmpDir}); err != nil {
+		t.Fatalf("create project: %v", err)
+	}
+	return database
+}
+
+// taskByStep finds a created step task by its step name.
+func taskByStep(res *Result, step string) *db.Task {
+	for i, s := range res.Definition.Steps {
+		if s.Name == step {
+			return res.Tasks[i]
+		}
+	}
+	return nil
+}
+
+func TestDefaultDefinitionIsValidDAG(t *testing.T) {
+	def, ok := Get("")
+	if !ok {
+		t.Fatal("default definition missing")
+	}
+	if def.Name != DefaultDefinition {
+		t.Errorf("default = %q, want %q", def.Name, DefaultDefinition)
+	}
+	if err := def.validate(); err != nil {
+		t.Fatalf("default definition invalid: %v", err)
+	}
+	// It should be the plan → code → 2 parallel reviews → collect shape.
+	if len(def.Steps) != 5 {
+		t.Errorf("got %d steps, want 5", len(def.Steps))
+	}
+	if roots := def.Roots(); len(roots) != 1 || roots[0].Name != "Plan" {
+		t.Errorf("roots = %v, want [Plan]", roots)
+	}
+}
+
+func TestCreateBuildsDAG(t *testing.T) {
+	database := testDB(t)
+	res, err := Create(database, Options{Goal: "Add rate limiting to the API", Project: "test", Execute: true})
+	if err != nil {
+		t.Fatalf("Create: %v", err)
+	}
+	if len(res.Tasks) != 5 {
+		t.Fatalf("got %d tasks, want 5", len(res.Tasks))
+	}
+
+	wantBranch := "pipeline/" + itoa(taskByStep(res, "Plan").ID) + "-add-rate-limiting-to-the-api"
+	if res.Branch != wantBranch {
+		t.Errorf("branch = %q, want %q", res.Branch, wantBranch)
+	}
+
+	plan := taskByStep(res, "Plan")
+	code := taskByStep(res, "Code")
+	rvA := taskByStep(res, "Review A")
+	rvB := taskByStep(res, "Review B")
+	collect := taskByStep(res, "Collect")
+
+	// Root owns the branch; everyone else checks it out via SourceBranch.
+	if plan.BranchName != wantBranch || plan.SourceBranch != "" {
+		t.Errorf("Plan branch=%q source=%q, want branch pinned", plan.BranchName, plan.SourceBranch)
+	}
+	for _, tk := range []*db.Task{code, rvA, rvB, collect} {
+		if tk.SourceBranch != wantBranch {
+			t.Errorf("%s SourceBranch = %q, want %q", tk.Title, tk.SourceBranch, wantBranch)
+		}
+	}
+
+	// Root starts; everything else waits blocked.
+	if plan.Status != db.StatusQueued {
+		t.Errorf("Plan status = %q, want queued", plan.Status)
+	}
+	for _, tk := range []*db.Task{code, rvA, rvB, collect} {
+		reloaded, _ := database.GetTask(tk.ID)
+		if reloaded.Status != db.StatusBlocked {
+			t.Errorf("%s status = %q, want blocked", tk.Title, reloaded.Status)
+		}
+	}
+
+	// Dependency edges: Code←Plan, ReviewA←Code, ReviewB←Code, Collect←ReviewA+ReviewB.
+	assertDep(t, database, plan.ID, code.ID)
+	assertDep(t, database, code.ID, rvA.ID)
+	assertDep(t, database, code.ID, rvB.ID)
+	assertDep(t, database, rvA.ID, collect.ID)
+	assertDep(t, database, rvB.ID, collect.ID)
+
+	// Every step task is tagged for grouping and carries the goal + branch in its prompt.
+	for _, tk := range res.Tasks {
+		if tk.Tags != "pipeline" {
+			t.Errorf("%s tags = %q, want pipeline", tk.Title, tk.Tags)
+		}
+		if !strings.Contains(tk.Body, "Add rate limiting to the API") || !strings.Contains(tk.Body, wantBranch) {
+			t.Errorf("%s body missing goal/branch", tk.Title)
+		}
+	}
+	// Parallel reviewers write distinct review files and push to their OWN branch
+	// (not the shared branch) so a weak agent can't clobber the other reviewer.
+	if !strings.Contains(rvA.Body, "review-review-a.md") || !strings.Contains(rvA.Body, wantBranch+"-review-a") {
+		t.Errorf("Review A body missing its review file / own branch: %s", rvA.Body)
+	}
+	if !strings.Contains(rvB.Body, wantBranch+"-review-b") {
+		t.Errorf("Review B body missing its own review branch")
+	}
+	// Collect is told exactly which review branches to read.
+	if !strings.Contains(collect.Body, wantBranch+"-review-a") || !strings.Contains(collect.Body, wantBranch+"-review-b") {
+		t.Errorf("Collect body missing review branch references: %s", collect.Body)
+	}
+}
+
+func TestCreateAutoAdvancesDAGWithParallelJoin(t *testing.T) {
+	database := testDB(t)
+	res, err := Create(database, Options{Goal: "Ship it", Project: "test", Execute: true})
+	if err != nil {
+		t.Fatalf("Create: %v", err)
+	}
+	plan := taskByStep(res, "Plan")
+	code := taskByStep(res, "Code")
+	rvA := taskByStep(res, "Review A")
+	rvB := taskByStep(res, "Review B")
+	collect := taskByStep(res, "Collect")
+
+	// Plan done → Code queues.
+	must(t, database.UpdateTaskStatus(plan.ID, db.StatusDone))
+	if s := statusOf(t, database, code.ID); s != db.StatusQueued {
+		t.Errorf("Code = %q after Plan done, want queued", s)
+	}
+
+	// Code done → BOTH reviewers queue at once (fan-out).
+	must(t, database.UpdateTaskStatus(code.ID, db.StatusDone))
+	if s := statusOf(t, database, rvA.ID); s != db.StatusQueued {
+		t.Errorf("Review A = %q after Code done, want queued", s)
+	}
+	if s := statusOf(t, database, rvB.ID); s != db.StatusQueued {
+		t.Errorf("Review B = %q after Code done, want queued", s)
+	}
+	// Collect still waits — both reviews outstanding.
+	if s := statusOf(t, database, collect.ID); s != db.StatusBlocked {
+		t.Errorf("Collect = %q with reviews pending, want blocked", s)
+	}
+
+	// One reviewer done → Collect still blocked (join needs both).
+	must(t, database.UpdateTaskStatus(rvA.ID, db.StatusDone))
+	if s := statusOf(t, database, collect.ID); s != db.StatusBlocked {
+		t.Errorf("Collect = %q after one review, want still blocked", s)
+	}
+
+	// Second reviewer done → Collect queues (join satisfied).
+	must(t, database.UpdateTaskStatus(rvB.ID, db.StatusDone))
+	if s := statusOf(t, database, collect.ID); s != db.StatusQueued {
+		t.Errorf("Collect = %q after both reviews, want queued", s)
+	}
+}
+
+func TestCreateValidation(t *testing.T) {
+	database := testDB(t)
+	cases := []struct {
+		name string
+		opts Options
+	}{
+		{"empty goal", Options{Goal: "  ", Project: "test"}},
+		{"empty project", Options{Goal: "x", Project: ""}},
+		{"unknown definition", Options{Goal: "x", Project: "test", Definition: "nope"}},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			if _, err := Create(database, tc.opts); err == nil {
+				t.Errorf("expected error for %s", tc.name)
+			}
+		})
+	}
+}
+
+func TestValidateRejectsMalformedDAGs(t *testing.T) {
+	cases := map[string]Definition{
+		"dup name":    {Name: "d", Steps: []Step{{Name: "A"}, {Name: "A"}}},
+		"unknown dep": {Name: "d", Steps: []Step{{Name: "A"}, {Name: "B", Deps: []string{"Z"}}}},
+		"no root":     {Name: "d", Steps: []Step{{Name: "A", Deps: []string{"B"}}, {Name: "B", Deps: []string{"A"}}}},
+		"empty":       {Name: "d"},
+	}
+	for name, def := range cases {
+		t.Run(name, func(t *testing.T) {
+			if err := def.validate(); err == nil {
+				t.Errorf("expected validate() error for %s", name)
+			}
+		})
+	}
+}
+
+func TestSlugify(t *testing.T) {
+	cases := map[string]string{
+		"Add rate limiting to the API": "add-rate-limiting-to-the-api",
+		"  Trim & symbols!! ":          "trim-symbols",
+		"":                             "pipeline",
+		"Review B":                     "review-b",
+	}
+	for in, want := range cases {
+		if got := slugify(in, 40); got != want {
+			t.Errorf("slugify(%q) = %q, want %q", in, got, want)
+		}
+	}
+}
+
+func assertDep(t *testing.T, database *db.DB, blocker, blocked int64) {
+	t.Helper()
+	dep, err := database.GetDependency(blocker, blocked)
+	if err != nil {
+		t.Fatalf("GetDependency(%d,%d): %v", blocker, blocked, err)
+	}
+	if dep == nil {
+		t.Fatalf("missing dependency %d → %d", blocker, blocked)
+	}
+	if !dep.AutoQueue {
+		t.Errorf("dependency %d → %d not auto-queue", blocker, blocked)
+	}
+}
+
+func statusOf(t *testing.T, database *db.DB, id int64) string {
+	t.Helper()
+	task, err := database.GetTask(id)
+	if err != nil {
+		t.Fatalf("GetTask(%d): %v", id, err)
+	}
+	return task.Status
+}
+
+func must(t *testing.T, err error) {
+	t.Helper()
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
+// itoa avoids pulling strconv into the test just for one conversion.
+func itoa(n int64) string {
+	if n == 0 {
+		return "0"
+	}
+	neg := n < 0
+	if neg {
+		n = -n
+	}
+	var buf [20]byte
+	i := len(buf)
+	for n > 0 {
+		i--
+		buf[i] = byte('0' + n%10)
+		n /= 10
+	}
+	if neg {
+		i--
+		buf[i] = '-'
+	}
+	return string(buf[i:])
+}
diff --git a/internal/ui/app.go b/internal/ui/app.go
index 21007646..359f6f09 100644
--- a/internal/ui/app.go
+++ b/internal/ui/app.go
@@ -25,6 +25,7 @@ import (
 	"github.com/bborn/workflow/internal/db"
 	"github.com/bborn/workflow/internal/executor"
 	"github.com/bborn/workflow/internal/github"
+	"github.com/bborn/workflow/internal/pipeline"
 	"github.com/bborn/workflow/internal/tasksummary"
 )
 
@@ -426,6 +427,7 @@ type AppModel struct {
 	newTaskForm        *FormModel
 	pendingTask        *db.Task
 	pendingAttachments []string
+	pendingPipeline    string // non-empty when the pending submission is a pipeline definition
 	queueConfirm       *huh.Form
 	queueValue         string
 
@@ -556,7 +558,31 @@ func (m *AppModel) updateTaskInList(task *db.Task) {
 			break
 		}
 	}
-	m.kanban.SetTasks(m.tasks)
+	m.kanban.SetTasks(m.collapseForBoard(m.tasks))
+}
+
+// collapseForBoard turns a task list into what the board should show: a workflow's
+// step tasks are folded into a single lead card (see pipeline.GroupWorkflows) so N
+// steps for one goal don't clutter the board as N cards. It also hands the kanban
+// the lead→group map so those cards can render workflow progress. Non-workflow
+// tasks pass through unchanged.
+func (m *AppModel) collapseForBoard(tasks []*db.Task) []*db.Task {
+	groups, rest := pipeline.GroupWorkflows(tasks)
+	leadMap := make(map[int64]*pipeline.Group, len(groups))
+	out := make([]*db.Task, 0, len(rest)+len(groups))
+	out = append(out, rest...)
+	for _, g := range groups {
+		lead := g.Lead()
+		if lead == nil {
+			continue
+		}
+		leadMap[lead.ID] = g
+		out = append(out, lead)
+	}
+	if m.kanban != nil {
+		m.kanban.SetWorkflowGroups(leadMap)
+	}
+	return out
 }
 
 // NewAppModel creates a new application model.
@@ -649,7 +675,7 @@ func NewAppModel(database *db.DB, exec *executor.Executor, workingDir string, ve
 func (m *AppModel) SetTasks(tasks []*db.Task) {
 	m.tasks = tasks
 	m.loading = false
-	m.kanban.SetTasks(tasks)
+	m.kanban.SetTasks(m.collapseForBoard(tasks))
 }
 
 // SetDebugStatePath sets the path for dumping debug state.
@@ -1145,6 +1171,19 @@ func (m *AppModel) Update(msg tea.Msg) (tea.Model, tea.Cmd) {
 			m.err = msg.err
 		}
 
+	case pipelineCreatedMsg:
+		if msg.err == nil && msg.result != nil {
+			m.currentView = ViewDashboard
+			m.newTaskForm = nil
+			m.showWelcome = false
+			n := len(msg.result.Tasks)
+			m.notification = fmt.Sprintf("%s Created %s workflow (%d steps) on %s", IconDone(), msg.result.Definition.Name, n, msg.result.Branch)
+			m.notifyUntil = time.Now().Add(6 * time.Second)
+			cmds = append(cmds, m.loadTasks())
+		} else {
+			m.err = msg.err
+		}
+
 	case taskUpdatedMsg:
 		if msg.err == nil {
 			// Update the selected task if we're in detail view
@@ -2313,8 +2352,8 @@ func (m *AppModel) refreshDetailPinnedNav() {
 // Uses the same matching logic as the command palette (Ctrl+P) for consistency.
 func (m *AppModel) applyFilter() {
 	if m.filterText == "" {
-		// No filter, show all tasks
-		m.kanban.SetTasks(m.tasks)
+		// No filter, show all tasks (workflows collapsed to one card each)
+		m.kanban.SetTasks(m.collapseForBoard(m.tasks))
 		return
 	}
 
@@ -2344,7 +2383,7 @@ func (m *AppModel) applyFilter() {
 	for i, st := range scored {
 		filtered[i] = st.task
 	}
-	m.kanban.SetTasks(filtered)
+	m.kanban.SetTasks(m.collapseForBoard(filtered))
 }
 
 // parseFilterProjects extracts completed [project] tags, any trailing partial project,
@@ -2722,6 +2761,7 @@ func (m *AppModel) updateNewTaskForm(msg tea.Msg) (tea.Model, tea.Cmd) {
 			// Store pending task and create confirmation form
 			m.pendingTask = form.GetDBTask()
 			m.pendingAttachments = form.GetAttachments()
+			m.pendingPipeline = form.Pipeline()
 			// Default to last queue choice for this project. Permission mode now
 			// lives on the task form, so this is just execute-now vs backlog;
 			// fold any legacy "auto"/"dangerous" choices into "yes".
@@ -2729,14 +2769,21 @@ func (m *AppModel) updateNewTaskForm(msg tea.Msg) (tea.Model, tea.Cmd) {
 			if last, err := m.db.GetSetting("last_queue_choice:" + m.pendingTask.Project); err == nil && last != "" && last != "no" {
 				m.queueValue = "yes"
 			}
+			queueTitle := "Queue for execution?"
+			runOpt, stageOpt := "Yes — execute now", "No — save to backlog"
+			if m.pendingPipeline != "" {
+				queueTitle = "Start this workflow now?"
+				runOpt = "Yes — start now"
+				stageOpt = "No — save for later"
+			}
 			m.queueConfirm = huh.NewForm(
 				huh.NewGroup(
 					huh.NewSelect[string]().
 						Key("queue").
-						Title("Queue for execution?").
+						Title(queueTitle).
 						Options(
-							huh.NewOption("Yes — execute now", "yes"),
-							huh.NewOption("No — save to backlog", "no"),
+							huh.NewOption(runOpt, "yes"),
+							huh.NewOption(stageOpt, "no"),
 						).
 						Value(&m.queueValue),
 				),
@@ -2749,6 +2796,7 @@ func (m *AppModel) updateNewTaskForm(msg tea.Msg) (tea.Model, tea.Cmd) {
 		if form.cancelled {
 			m.currentView = ViewDashboard
 			m.newTaskForm = nil
+			m.pendingPipeline = ""
 			return m, nil
 		}
 	}
@@ -2782,6 +2830,22 @@ func (m *AppModel) updateNewTaskConfirm(msg tea.Msg) (tea.Model, tea.Cmd) {
 			// Remember the choice for this project
 			m.db.SetSetting("last_queue_choice:"+m.pendingTask.Project, m.queueValue)
 
+			// A pipeline selection turns the pending task into the goal for a
+			// multi-phase chain instead of a single task. The queue choice maps to
+			// whether the first phase runs now (Execute) or the chain is staged.
+			if m.pendingPipeline != "" {
+				task := m.pendingTask
+				definition := m.pendingPipeline
+				execute := m.queueValue == "yes"
+				m.pendingTask = nil
+				m.pendingAttachments = nil
+				m.pendingPipeline = ""
+				m.newTaskForm = nil
+				m.queueConfirm = nil
+				m.currentView = ViewDashboard
+				return m, m.createPipeline(task, definition, execute)
+			}
+
 			// Permission mode is already set on the task from the form; this
 			// choice only decides whether to run now or save to the backlog.
 			switch m.queueValue {
@@ -2794,6 +2858,7 @@ func (m *AppModel) updateNewTaskConfirm(msg tea.Msg) (tea.Model, tea.Cmd) {
 			attachments := m.pendingAttachments
 			m.pendingTask = nil
 			m.pendingAttachments = nil
+			m.pendingPipeline = ""
 			m.newTaskForm = nil
 			m.queueConfirm = nil
 			m.currentView = ViewDashboard
@@ -3988,6 +4053,11 @@ type taskCreatedMsg struct {
 	err  error
 }
 
+type pipelineCreatedMsg struct {
+	result *pipeline.Result
+	err    error
+}
+
 type taskUpdatedMsg struct {
 	task *db.Task
 	err  error
@@ -4227,6 +4297,34 @@ func (m *AppModel) createTaskWithAttachments(t *db.Task, attachmentPaths []strin
 	}
 }
 
+// createPipeline builds a multi-phase pipeline from the form's task, using its
+// title/body as the goal and its project/permission mode for every phase. The
+// task itself is not persisted — it is only the goal carrier.
+func (m *AppModel) createPipeline(t *db.Task, definition string, execute bool) tea.Cmd {
+	database := m.db
+	return func() tea.Msg {
+		goal := strings.TrimSpace(t.Title)
+		if body := strings.TrimSpace(t.Body); body != "" {
+			if goal == "" {
+				goal = body
+			} else {
+				goal = goal + "\n\n" + body
+			}
+		}
+		result, err := pipeline.Create(database, pipeline.Options{
+			Goal:           goal,
+			Project:        t.Project,
+			Definition:     definition,
+			PermissionMode: t.PermissionMode,
+			Execute:        execute,
+		})
+		if err == nil {
+			database.CompleteOnboarding()
+		}
+		return pipelineCreatedMsg{result: result, err: err}
+	}
+}
+
 func (m *AppModel) queueTask(id int64) tea.Cmd {
 	database := m.db
 	exec := m.executor
diff --git a/internal/ui/form.go b/internal/ui/form.go
index 5defb5fe..f7cff06e 100644
--- a/internal/ui/form.go
+++ b/internal/ui/form.go
@@ -17,8 +17,16 @@ import (
 
 	"github.com/bborn/workflow/internal/autocomplete"
 	"github.com/bborn/workflow/internal/db"
+	"github.com/bborn/workflow/internal/pipeline"
 )
 
+// pipelineOptions returns the selectable pipeline choices for the form. The
+// first entry is "" — an ordinary single task — followed by each built-in
+// pipeline definition (e.g. plan-code-review).
+func pipelineOptions() []string {
+	return append([]string{""}, pipeline.DefinitionNames()...)
+}
+
 // FormField represents the currently focused field.
 type FormField int
 
@@ -32,6 +40,7 @@ const (
 	FieldEffort
 	FieldModel
 	FieldPermission
+	FieldPipeline
 	FieldCount
 )
 
@@ -80,6 +89,9 @@ type FormModel struct {
 	permissionIdx      int
 	permissionModes    []string // Selectable permission modes
 	permissionTouched  bool     // user picked a permission explicitly; stop following project default
+	pipeline           string   // Pipeline definition ("" = a single ordinary task)
+	pipelineIdx        int
+	pipelines          []string // Selectable pipeline options; "" (first) means "single task"
 	queue              bool
 	attachments        []string // Parsed file paths
 	attachmentCursor   int      // Index of the currently selected attachment chip
@@ -287,7 +299,8 @@ func NewEditFormModel(database *db.DB, task *db.Task, width, height int, availab
 		permissionMode:      task.EffectivePermissionMode(),
 		permissionModes:     permissionModeOptions(),
 		permissionIdx:       permissionIndexFor(permissionModeOptions(), task.EffectivePermissionMode()),
-		permissionTouched:   true, // editing: keep the task's existing mode unless changed
+		pipelines:           pipelineOptions(), // Field hidden while editing; kept non-nil for safety.
+		permissionTouched:   true,              // editing: keep the task's existing mode unless changed
 		isEdit:              true,
 		prURL:               task.PRURL,
 		prNumber:            task.PRNumber,
@@ -411,7 +424,8 @@ func NewFormModel(database *db.DB, width, height int, workingDir string, availab
 		effortLevels:        effortLevelOptions(), // Defaults to "" (Claude's global default)
 		models:              modelOptions(),       // Defaults to "" (Claude's global default)
 		permissionModes:     permissionModeOptions(),
-		showAdvanced:        showAdvanced, // Load from user preference
+		pipelines:           pipelineOptions(), // Defaults to "" (a single ordinary task)
+		showAdvanced:        showAdvanced,      // Load from user preference
 	}
 
 	// Load task types from database
@@ -866,6 +880,11 @@ func (m *FormModel) Update(msg tea.Msg) (tea.Model, tea.Cmd) {
 				m.permissionTouched = true
 				return m, nil
 			}
+			if m.focused == FieldPipeline && len(m.pipelines) > 0 {
+				m.pipelineIdx = (m.pipelineIdx - 1 + len(m.pipelines)) % len(m.pipelines)
+				m.pipeline = m.pipelines[m.pipelineIdx]
+				return m, nil
+			}
 
 		case "right":
 			if m.handleAttachmentNavigation(1) {
@@ -910,6 +929,11 @@ func (m *FormModel) Update(msg tea.Msg) (tea.Model, tea.Cmd) {
 				m.permissionTouched = true
 				return m, nil
 			}
+			if m.focused == FieldPipeline && len(m.pipelines) > 0 {
+				m.pipelineIdx = (m.pipelineIdx + 1) % len(m.pipelines)
+				m.pipeline = m.pipelines[m.pipelineIdx]
+				return m, nil
+			}
 
 		// Alt+Arrow keys for word/paragraph movement in text fields
 		case "alt+left":
@@ -1443,6 +1467,11 @@ func (m *FormModel) isFieldVisible(field FormField) bool {
 		// only shown in advanced mode and only when the Claude executor is selected.
 		return m.showAdvanced && m.executor == db.ExecutorClaude
 	}
+	if field == FieldPipeline {
+		// A pipeline turns creation into a multi-phase chain; it only makes sense
+		// for a brand-new task, never when editing an existing one.
+		return m.showAdvanced && !m.isEdit
+	}
 	if m.showAdvanced {
 		return true
 	}
@@ -2027,6 +2056,29 @@ func (m *FormModel) View() string {
 			b.WriteString("\n")
 		}
 
+		// Pipeline selector: turn creation into a multi-phase plan → code → review
+		// chain routed across models/executors. Only shown for new tasks.
+		if m.isFieldVisible(FieldPipeline) {
+			cursor = " "
+			if m.focused == FieldPipeline {
+				cursor = cursorStyle.Render("▸")
+			}
+			pipelineLabels := make([]string, len(m.pipelines))
+			for i, l := range m.pipelines {
+				if l == "" {
+					pipelineLabels[i] = "single task"
+				} else {
+					pipelineLabels[i] = l
+				}
+			}
+			b.WriteString(cursor + " " + labelStyle.Render("Workflow") + m.renderSelector(pipelineLabels, m.pipelineIdx, m.focused == FieldPipeline, selectedStyle, optionStyle, dimStyle))
+			b.WriteString("\n")
+			if m.pipeline != "" {
+				hint := "  " + dimStyle.Render("runs as a workflow: plan → code → parallel review → collect, on one branch")
+				b.WriteString(hint + "\n")
+			}
+		}
+
 		// Trailing gap to separate the selector list from the help line.
 		b.WriteString("\n")
 	} else {
@@ -2541,6 +2593,14 @@ func (m *FormModel) GetAttachments() []string {
 	return m.attachments
 }
 
+// Pipeline returns the selected pipeline definition name, or "" when the form is
+// creating an ordinary single task. When non-empty, the caller should build a
+// pipeline (see internal/pipeline) using the form's task as the goal carrier
+// rather than creating a single task.
+func (m *FormModel) Pipeline() string {
+	return m.pipeline
+}
+
 // GetAttachment returns the first attachment for backwards compatibility.
 func (m *FormModel) GetAttachment() string {
 	if len(m.attachments) > 0 {
diff --git a/internal/ui/form_pipeline_test.go b/internal/ui/form_pipeline_test.go
new file mode 100644
index 00000000..2ac3ee48
--- /dev/null
+++ b/internal/ui/form_pipeline_test.go
@@ -0,0 +1,76 @@
+package ui
+
+import (
+	"strings"
+	"testing"
+
+	tea "github.com/charmbracelet/bubbletea"
+)
+
+// TestFormPipelineFieldDefaultsToSingleTask verifies a fresh form creates an
+// ordinary task (no pipeline) until the user selects one.
+func TestFormPipelineFieldDefaultsToSingleTask(t *testing.T) {
+	m := NewFormModel(nil, 100, 50, "", nil)
+	if m.Pipeline() != "" {
+		t.Errorf("default Pipeline() = %q, want empty", m.Pipeline())
+	}
+	if len(m.pipelines) < 2 || m.pipelines[0] != "" {
+		t.Errorf("pipelines = %v, want first entry empty followed by definitions", m.pipelines)
+	}
+}
+
+// TestFormPipelineFieldCycles verifies the pipeline selector advances to a real
+// definition and the getter reflects it.
+func TestFormPipelineFieldCycles(t *testing.T) {
+	m := NewFormModel(nil, 100, 50, "", nil)
+	m.showAdvanced = true
+	m.focused = FieldPipeline
+
+	updated, _ := m.Update(tea.KeyMsg{Type: tea.KeyRight})
+	form, ok := updated.(*FormModel)
+	if !ok {
+		t.Fatalf("Update returned %T, want *FormModel", updated)
+	}
+	if form.Pipeline() == "" {
+		t.Fatal("Pipeline() still empty after cycling right")
+	}
+	if form.Pipeline() != "plan-code-review" {
+		t.Errorf("Pipeline() = %q, want plan-code-review", form.Pipeline())
+	}
+
+	// Cycling left returns to "single task".
+	updated, _ = form.Update(tea.KeyMsg{Type: tea.KeyLeft})
+	form = updated.(*FormModel)
+	if form.Pipeline() != "" {
+		t.Errorf("Pipeline() = %q after cycling back, want empty", form.Pipeline())
+	}
+}
+
+// TestFormPipelineFieldHiddenWhenEditing verifies the pipeline option is only
+// offered for new tasks, never when editing an existing one.
+func TestFormPipelineFieldVisibility(t *testing.T) {
+	m := NewFormModel(nil, 100, 50, "", nil)
+	m.showAdvanced = true
+	if !m.isFieldVisible(FieldPipeline) {
+		t.Error("FieldPipeline should be visible for a new task in advanced mode")
+	}
+	m.showAdvanced = false
+	if m.isFieldVisible(FieldPipeline) {
+		t.Error("FieldPipeline should be hidden when advanced fields are collapsed")
+	}
+	m.showAdvanced = true
+	m.isEdit = true
+	if m.isFieldVisible(FieldPipeline) {
+		t.Error("FieldPipeline should be hidden when editing an existing task")
+	}
+}
+
+// TestFormPipelineRendersSelector verifies the selector appears in the rendered
+// form when advanced fields are shown.
+func TestFormPipelineRendersSelector(t *testing.T) {
+	m := NewFormModel(nil, 120, 50, "", nil)
+	m.showAdvanced = true
+	if !strings.Contains(m.View(), "Workflow") {
+		t.Error("advanced form View should render the Workflow selector")
+	}
+}
diff --git a/internal/ui/kanban.go b/internal/ui/kanban.go
index 899653c8..24512181 100644
--- a/internal/ui/kanban.go
+++ b/internal/ui/kanban.go
@@ -8,6 +8,7 @@ import (
 
 	"github.com/bborn/workflow/internal/db"
 	"github.com/bborn/workflow/internal/github"
+	"github.com/bborn/workflow/internal/pipeline"
 )
 
 // emptyColumnMessage returns a contextual message for empty columns.
@@ -48,13 +49,14 @@ type KanbanBoard struct {
 	collapsedColumns  map[int]bool // Columns that are collapsed (show only header)
 	width             int
 	height            int
-	allTasks          []*db.Task               // All tasks
-	prInfo            map[int64]*github.PRInfo // PR info by task ID
-	runningProcesses  map[int64]bool           // Tasks with running shell processes
-	tasksNeedingInput map[int64]bool           // Tasks waiting for user input (active input notification)
-	blockedByDeps     map[int64]int            // Tasks blocked by dependencies (task ID -> open blocker count)
-	hiddenDoneCount   int                      // Number of done tasks not shown (older ones)
-	originColumn      int                      // Column where detail view navigation started (-1 = not set)
+	allTasks          []*db.Task                // All tasks
+	prInfo            map[int64]*github.PRInfo  // PR info by task ID
+	runningProcesses  map[int64]bool            // Tasks with running shell processes
+	tasksNeedingInput map[int64]bool            // Tasks waiting for user input (active input notification)
+	blockedByDeps     map[int64]int             // Tasks blocked by dependencies (task ID -> open blocker count)
+	workflowGroups    map[int64]*pipeline.Group // Lead task ID -> workflow group (collapsed workflow cards)
+	hiddenDoneCount   int                       // Number of done tasks not shown (older ones)
+	originColumn      int                       // Column where detail view navigation started (-1 = not set)
 
 	// Render cache. View() is called on every Bubble Tea Update (key, tick,
 	// mouse move, task event), but the board's pixels only change when one of
@@ -211,6 +213,20 @@ func (k *KanbanBoard) NeedsInput(taskID int64) bool {
 
 // SetBlockedByDeps updates the map of tasks blocked by dependencies.
 // The map contains task ID -> number of open blockers.
+// SetWorkflowGroups records the collapsed workflow groups, keyed by the lead task
+// whose card represents the whole workflow on the board.
+func (k *KanbanBoard) SetWorkflowGroups(groups map[int64]*pipeline.Group) {
+	k.workflowGroups = groups
+}
+
+// workflowGroup returns the workflow group a task leads, or nil.
+func (k *KanbanBoard) workflowGroup(taskID int64) *pipeline.Group {
+	if k.workflowGroups == nil {
+		return nil
+	}
+	return k.workflowGroups[taskID]
+}
+
 func (k *KanbanBoard) SetBlockedByDeps(blockedByDeps map[int64]int) {
 	k.blockedByDeps = blockedByDeps
 }
@@ -797,6 +813,18 @@ func (k *KanbanBoard) hashTaskCard(h *sigHasher, t *db.Task) {
 	if t.Status == db.StatusProcessing {
 		h.int(k.spinnerFrame)
 	}
+	// Workflow lead cards render the goal + step progress instead of the raw
+	// title/sub-line, so the workflow's shape must feed the signature or the card
+	// caches serve a stale badge as the workflow advances.
+	if g := k.workflowGroup(t.ID); g != nil {
+		h.boolean(true)
+		h.str(g.Goal())
+		h.str(g.StepLabel())
+		h.int(g.DoneCount())
+		h.int(g.Total())
+	} else {
+		h.boolean(false)
+	}
 }
 
 // collapsedColumnWidth is the fixed width for collapsed column strips.
@@ -1306,8 +1334,14 @@ func (k *KanbanBoard) renderTaskCard(task *db.Task, width int, isSelected bool)
 		}
 	}
 
-	// Title (truncate if needed)
+	// Title (truncate if needed). Workflow lead cards show the goal with a "⇄"
+	// marker in place of the "[Step] goal" step title, so one card stands for the
+	// whole workflow.
+	wf := k.workflowGroup(task.ID)
 	title := task.Title
+	if wf != nil {
+		title = "⇄ " + wf.Goal()
+	}
 	maxTitleLen := width - 4
 	if maxTitleLen < 10 {
 		maxTitleLen = 10
@@ -1360,7 +1394,18 @@ func (k *KanbanBoard) renderTaskCard(task *db.Task, width int, isSelected bool)
 		cardStyle = cardStyle.Foreground(ColorWarning)
 	}
 
-	content := idLine + "\n" + titleLine + "\n" + k.cardSubLine(task, width, isSelected)
+	subLine := k.cardSubLine(task, width, isSelected)
+	if wf != nil {
+		// Replace the per-task activity line with workflow progress: the current
+		// step (or "Review ∥" during the parallel fan-out) and steps completed.
+		badge := fmt.Sprintf("%s · %d/%d", wf.StepLabel(), wf.DoneCount(), wf.Total())
+		if isSelected {
+			subLine = badge
+		} else {
+			subLine = Dim.Render(badge)
+		}
+	}
+	content := idLine + "\n" + titleLine + "\n" + subLine
 	rendered := cardStyle.Render(content)
 
 	if k.cardCache == nil || len(k.cardCache) >= cardCacheMax {
diff --git a/skills/taskyou/SKILL.md b/skills/taskyou/SKILL.md
index 92de410e..12926c2c 100644
--- a/skills/taskyou/SKILL.md
+++ b/skills/taskyou/SKILL.md
@@ -62,6 +62,8 @@ Use `ty` (short) or `taskyou` (full) — both work identically.
 | List by status | `ty list --status <status> --json` |
 | View task details | `ty show <id> --json --logs` |
 | Create task | `ty create "title" --body "description"` |
+| Create a workflow | `ty pipeline "goal" --project <name>` (plan → code → parallel review → collect) |
+| List / author workflows | `ty pipeline --list` · `ty pipeline new "<describe it>"` · `ty pipeline edit` |
 | Execute task | `ty execute <id>` |
 | Retry with feedback | `ty retry <id> --feedback "..."` |
 | Change status | `ty status <id> <status>` |
@@ -247,6 +249,8 @@ the worktree, not the executor inside it.
 **Task introspection:**
 - `taskyou_show_task` — Get details of any task in the current project.
 - `taskyou_create_task` — Create a follow-up task.
+- `taskyou_create_pipeline` — Spin up a multi-step workflow (plan → code →
+  parallel review → collect) for a goal, when it warrants more than one task.
 - `taskyou_list_tasks` — See other active tasks in this project.
 
 ## Best Practices