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
- Variables
- func DefinitionNames(extraDirs ...string) []string
- func GroupKey(t *db.Task) string
- func IsGateStep(t *db.Task) bool
- func IsTerminalStep(database *db.DB, task *db.Task) bool
- func IsWorkflowTask(t *db.Task) bool
- func LookupKindInstructions(name string, extraDirs ...string) string
- func Marshal(def Definition) ([]byte, error)
- func RunStepVerify(dir, command string) (output string, ok bool)
- func WorkflowDirs(projectDir string) []string
- func WorkflowsDir() string
- type Definition
- func Definitions(extraDirs ...string) []Definition
- func Flatten(def Definition, resolve ResolveFunc) (Definition, error)
- func GenerateDefinition(ctx context.Context, apiKey, description string) (Definition, []byte, error)
- func Get(name string, extraDirs ...string) (Definition, bool)
- func ParseDefinition(data []byte) (Definition, error)
- type Group
- type Options
- type ResolveFunc
- type Result
- type Step
Constants ¶
const DefaultDefinition = ""
DefaultDefinition is the definition used when none is named. It is empty: there are no built-in workflows — every workflow is installed from a plugin (or dropped into the workflows dir). `ty pipeline` with no -d therefore has nothing to run until the user installs a workflow, and Create says so.
const GateStepParkedLog = "Gate step finished — parked in 'blocked' for human review. Approve it with `ty close <id>` to release the next phase."
GateStepParkedLog is the system-log line marking a finished gate step as parked in 'blocked' awaiting human review. Shared so the completion paths (taskyou_complete, the Stop hook, the daemon sweep) use the same text and can dedupe against it.
const StepVerifyTimeout = 15 * time.Minute
StepVerifyTimeout bounds a step's evidence-gate command. Generous because the common case is a full build + test suite; a run that exceeds it fails closed.
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 ¶
var PluginWorkflowDirs func() []string
PluginWorkflowDirs, when set, returns extra directories to search for workflow definitions — the workflows/ subdir of every installed plugin. It is a seam so the pipeline package needn't import internal/hooks (which would cycle); main wires it at startup. nil = no plugin workflows.
Functions ¶
func DefinitionNames ¶
DefinitionNames returns the names of the workflow files.
func GroupKey ¶ added in v0.3.18
GroupKey exports groupKey so callers outside this package (e.g. the MCP artifact handlers) can derive a workflow task's shared-branch key without re-implementing the SourceBranch-else-BranchName rule.
func IsGateStep ¶ added in v0.3.18
IsGateStep reports whether a workflow step is a human-in-the-loop gate — a step that, once it produces its output, parks in 'blocked' for human review instead of advancing the DAG. A human releases it (and its dependents) with `ty close`. The gate is declared in the workflow YAML (stepYAML.Gate) and persisted as a "gate" tag token on the step task (see pipeline.Create). Sibling to IsWorkflowTask / IsTerminalStep. Non-workflow tasks are never gate steps.
func IsTerminalStep ¶
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 ¶
IsWorkflowTask reports whether a task belongs to a workflow (tagged "pipeline" with a shared branch).
func LookupKindInstructions ¶
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 RunStepVerify ¶ added in v0.3.18
RunStepVerify runs a workflow step's `verify:` command in dir via `sh -c` and reports whether it passed (exit 0). On failure it returns a tail of the combined stdout+stderr. The gate fails CLOSED: a command that can't start, or times out, counts as a failure — a broken environment must never rubber-stamp a step as complete. Shared by the MCP taskyou_complete handler and the daemon's git-based completion sweep so both enforce the gate identically.
func WorkflowDirs ¶
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 ¶
Group is a workflow's member tasks, keyed by their shared branch.
func GroupWorkflows ¶
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 ¶
ActiveSteps returns the step names currently processing or queued (there can be two during the parallel review fan-out).
func (*Group) Goal ¶
Goal returns the workflow goal, recovered by stripping the "Step " prefix off a member title.
func (*Group) Lead ¶
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.
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 ¶
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.
Gate bool // Human-in-the-loop: when this step finishes it parks 'blocked' for human review instead of advancing the DAG; a human releases it (and its dependents) with `ty close`. Persisted on the task as a "gate" tag token (see the Tags field in Create).
Verify string // Opt-in evidence gate: a shell command run in the step's worktree when the agent calls taskyou_complete. Non-zero exit REJECTS the completion (the step stays running and the command output is handed back so the agent fixes it) instead of trusting the agent's say-so. "" = no gate (today's behavior). Persisted in the task-keyed pipeline_step_verify table (a command string can't ride a tag token like Gate does).
}
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.