pipeline

package
v0.3.16 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Jul 14, 2026 License: MIT Imports: 15 Imported by: 0

Documentation

Overview

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.

Index

Constants

View Source
const DefaultDefinition = ""

DefaultDefinition is empty: there is no built-in workflow. Kinds live in the DB (task types); a kind runs as a workflow only when a same-named file adds steps.

View Source
const TerminalStepParkedLog = "Final workflow step finished — parked in 'blocked' for a human to review and merge."

TerminalStepParkedLog is the system-log line marking a finished terminal step as parked in 'blocked' awaiting a human merge. The Stop hook writes it when it catches the step finished; the daemon sweep writes it (once) when the Stop hook fired mid- push and left the generic "waiting for input" state instead. Shared so both paths use the same text and the sweep can dedupe against it.

Variables

This section is empty.

Functions

func DefinitionNames

func DefinitionNames(extraDirs ...string) []string

DefinitionNames returns the names of the workflow files.

func IsTerminalStep

func IsTerminalStep(database *db.DB, task *db.Task) bool

IsTerminalStep reports whether a workflow step is the sink — the step nothing else depends on, which is the one that opens the PR (see composeInstruction). A finished terminal step parks in 'blocked' awaiting a human merge review rather than advancing to 'done'; every other step goes 'done' so its dependents auto-queue. Non-workflow tasks are never terminal steps.

func IsWorkflowTask

func IsWorkflowTask(t *db.Task) bool

IsWorkflowTask reports whether a task belongs to a workflow (tagged "pipeline" with a shared branch).

func LookupKindInstructions

func LookupKindInstructions(name string, extraDirs ...string) string

LookupKindInstructions returns the instructions of a single-task kind defined in a YAML/built-in file (not a DB task type), or "" if there is no such kind. The executor uses it so a file-defined kind works as a task Type just like a DB type.

func Marshal

func Marshal(def Definition) ([]byte, error)

Marshal renders a Definition back to YAML (used by `ty pipeline new`).

func WorkflowDirs

func WorkflowDirs(projectDir string) []string

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 WorkflowsDir

func WorkflowsDir() string

WorkflowsDir returns the global directory custom workflow files live in.

Types

type Definition

type Definition struct {
	Name         string
	Description  string
	Instructions string // Single-task kind's prompt preset (used when Steps is empty). Mirrors db.TaskType.Instructions.
	Steps        []Step
	Custom       bool // true for kinds loaded from YAML files (vs the built-ins)
}

Definition is a "kind": one authored recipe. With Steps it is a workflow (a DAG of steps); without Steps it is a single-task kind whose Instructions become the task's prompt preset — the same thing a DB task type is. A step may reference another kind by name (Step.Kind), so a workflow composes kinds, and picking a kind is one action whether it fans out or not.

func Definitions

func Definitions(extraDirs ...string) []Definition

Definitions returns the workflow definitions discovered as files.

func Flatten

func Flatten(def Definition, resolve ResolveFunc) (Definition, error)

Flatten expands every step that references a multi-step kind into that kind's sub-DAG, returning a workflow whose steps are all leaves. Each leaf keeps its referenced Kind, so the built task's Type applies that kind's instructions. A single-task kind is returned unchanged. Cycles across kinds and nesting past maxKindDepth are rejected.

func GenerateDefinition

func GenerateDefinition(ctx context.Context, apiKey, description string) (Definition, []byte, error)

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 Get

func Get(name string, extraDirs ...string) (Definition, bool)

Get returns the workflow definition for a name (a same-named file), if one exists. Single-task kinds are DB task types and are not returned here — use KindResolver when you need both.

func ParseDefinition

func ParseDefinition(data []byte) (Definition, error)

ParseDefinition parses and validates one workflow YAML document.

func (Definition) IsSingle

func (d Definition) IsSingle() bool

IsSingle reports whether this kind is a single task (no DAG) — its Instructions are the whole recipe. The steps-less, instructions-only case is exactly a task type.

func (Definition) Roots

func (d Definition) Roots() []Step

Roots returns the steps with no dependencies (the entry points).

type Group

type Group struct {
	Branch  string
	Members []*db.Task // in step (creation) order
}

Group is a workflow's member tasks, keyed by their shared branch.

func GroupWorkflows

func GroupWorkflows(tasks []*db.Task) (groups []*Group, rest []*db.Task)

GroupWorkflows partitions tasks into workflow groups (2+ members sharing a branch) and the remaining ungrouped tasks, preserving input order for the rest.

func (*Group) ActiveSteps

func (g *Group) ActiveSteps() []string

ActiveSteps returns the step names currently processing or queued (there can be two during the parallel review fan-out).

func (*Group) DoneCount

func (g *Group) DoneCount() int

DoneCount returns how many steps have completed.

func (*Group) Goal

func (g *Group) Goal() string

Goal returns the workflow goal, recovered by stripping the "Step " prefix off a member title.

func (*Group) Lead

func (g *Group) Lead() *db.Task

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 (*Group) StepLabel

func (g *Group) StepLabel() string

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 (*Group) Total

func (g *Group) Total() int

Total returns the number of steps.

type Options

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.
}

Options configures a workflow build.

type ResolveFunc

type ResolveFunc func(name string) (Definition, bool)

ResolveFunc returns the kind definition for a name and whether it exists. It is how Flatten looks up a step's referenced kind (a YAML kind, a built-in, or — via the convention bridge — a DB task type surfaced as a single-task kind).

func KindResolver

func KindResolver(database *db.DB, extraDirs ...string) ResolveFunc

KindResolver resolves a kind name to its definition, by convention: if a same-named workflow file exists it runs as a workflow; otherwise the kind is a DB task type (a single task using its instructions). The DB is the single kind store; a workflow is just a file that adds steps to a kind of the same name.

type Result

type Result struct {
	Definition Definition
	Branch     string     // The shared branch every step runs on.
	Tasks      []*db.Task // Step tasks, in definition order.
}

Result describes a built workflow.

func Create

func Create(database *db.DB, opts Options) (*Result, error)

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.

type Step

type Step struct {
	Name        string            // Unique label, e.g. "Plan", "Code", "Review A", "Collect".
	Kind        string            // Optional: another kind this step runs. Sets the task's Type (so that kind's instructions apply); if that kind has steps, it's a sub-workflow inlined at build (see flatten.go).
	Executor    string            // Executor slug (db.ExecutorClaude, db.ExecutorCodex, ...).
	Model       string            // Per-task model override ("" = the executor's default).
	ConfigDir   string            // Per-task CLAUDE_CONFIG_DIR override ("" = use the project's/default config dir). Routes this step's Claude through a different config (e.g. ollama) without changing the project.
	Env         map[string]string // Per-task env overrides injected as a process-env prefix on the claude command (e.g. ANTHROPIC_BASE_URL/AUTH_TOKEN to route through ollama). nil = no overrides. Kept distinct from ConfigDir: env injection keeps the default config dir (plugins, MCP, trusted worktrees) intact and process env wins over stored creds — the mechanism that actually reaches ollama.
	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.
}

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.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL