The DAG engine

Operators

An operator is a Go struct with tagged fields

Each operator declares its inputs and outputs as struct fields annotated with dag:"input" and dag:"output". The daggen tool reads those tags and generates the field-routing methods (InputFields, OutputFields, SetInputField, ResetFields) so you only write Setup, Reset, and Run.

a custom operator

type DominantCategoryOp struct {
    // dag:"input"  — pointer fields the engine writes to before Run
    Labels *[]string `dag:"input"`

    // dag:"output" — value fields the engine reads after Run
    Result string      `dag:"output"`
}

func (op *DominantCategoryOp) Setup(_ *config.Params) error { return nil }
func (op *DominantCategoryOp) Reset() error                  { return nil }

func (op *DominantCategoryOp) Run(_ context.Context) error {
    op.Result = mostFrequent(*op.Labels)
    return nil
}

Inputs are pointers so the engine can leave them nil when an upstream vertex was skipped (see predicate gates below). Outputs are values; the engine takes their address after Run returns so downstream vertices can read them.

Builder

Wiring a graph

The fluent builder names vertices, picks the operator type by registered name, supplies vertex-scoped params, and connects field names to wire names. Wires are typed: a vertex output's static type must match the consumer's input type, or the graph fails to build.

building a small graph

g, err := graph.NewBuilder("hello").
    Vertex("greet").Op("StringConcatOp").
        Params(map[string]string{"a": "Hello, ", "b": "world"}).
        Output("Result", "out").
    Build()

eng, _ := dagor.NewEngine(g, pool)
_ = eng.Run(context.Background())
out, _ := eng.GetOutput("out")
fmt.Println(*(out.(*string)))

Scheduling

Parallelism comes from the DAG workflow, not your code

The engine performs a topological sort and schedules vertices to a goroutine pool the moment all their inputs are ready. Two independent subtrees execute concurrently for free — there is no go func() plumbing in your driver. The pool's worker count caps total concurrency (typically 2–4× CPU cores for compute, higher for I/O-bound graphs).

🪡

Goroutine pool

Use github.com/panjf2000/ants/v2 to create a worker pool and hand it to dagor.NewEngine. The pool bounds concurrency and shields you from unbounded goroutine fan-out on wide graphs.

⚙️

Context propagation

engine.Run(ctx) threads the same context.Context through every operator's Run. Cancel the context and every in-flight vertex sees ctx.Done(). Use context.WithTimeout for graph-wide deadlines.

📜

Reporters & logs

Pass dagor.WithReporter(reporter.New(slog.Default())) to log per-vertex start/end events and durations. Pair with structured slog to capture timings, AI token counts, and failures in one stream.

Predicate gates

Conditional execution without branching code

A vertex can carry a .Condition() predicate that reads upstream wire values. If the predicate returns false, the vertex is skipped — and the engine propagates that skip to every downstream vertex in the lane automatically. No AI tokens spent, no I/O performed, no error raised.

predicate + skip-propagation

predicate.Register("lane_is_billing", func(inputs map[string]any) bool {
    v, ok := inputs["ticket_category"].(*string)
    return ok && v != nil && *v == "billing"
})

Vertex("gate_billing").Op("LaneGateOp").
    Condition("lane_is_billing").
    ConditionInput("ticket_category").
    Input("Body", "ticket_body").
    Output("BodyOut", "billing_body").

Pair gates with CoalesceNStringOp downstream to merge several mutually-exclusive lanes into a single output wire. The ticket triager example walks through the full fan-out/coalesce pattern.

MapOver

Per-item fan-out with one sub-graph

When you have N items and want to run the same sub-graph on each in parallel, use MapOver. Declare a SubVertex bound to the items wire and a CollectInto sink — the engine spawns one sub-vertex per item and gathers results into a typed slice.

MapOver fan-out

Vertex("shoot_each").
    Input("Items", "result_urls").
    MapOver("url").
    SubVertex("shoot").
        Op("MCPScreenshotURLOp").
        Params(shootParams).
        Input("Input", "url").
        Output("Result", "shot_result").
    CollectInto("shot_result", "screenshot_results").

The collected output is a *[]any. The local-MCP example shows how a typed read-helper turns it back into a concrete slice.

Merge

Joining branches when some inputs may be absent

When several inputs feed one vertex and you expect some of them to be skipped (typical of predicate-gated lanes converging on a coalesce vertex), set .Merge(config.MergeCoalesce) so the vertex waits for all inputs to settle before evaluating — even when most settle as "skipped".

N-to-1 coalesce

Vertex("final").Op("CoalesceNStringOp").
    Params(map[string]int{"n": 4}).
    Merge(config.MergeCoalesce).
    Input("Input0", "billing_json").
    Input("Input1", "bug_json").
    Input("Input2", "feature_json").
    Input("Input3", "other_json").
    Output("Result", "final_brief").

Where to go next

The operator library

Every pre-built op you can drop into a vertex — math, strings, predicates, slice, time, JSON, HTTP, MCP, retrieval, repair.

Developer guide

Setup, first workflow, the fluent builder API, testing strategies, the bundled Claude Code skills.

Worked example

The ticket-triager walkthrough applies every concept on this page: predicate gates, skip-propagation, coalesce.