diff --git a/tools/flow/flow.go b/tools/flow/flow.go
new file mode 100644
index 0000000..5f0b412
--- /dev/null
+++ b/tools/flow/flow.go
@@ -0,0 +1,402 @@
+// Copyright 2020 CUE Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Package flow provides a low-level workflow manager based on a CUE Instance.
+//
+// A Task defines an operational unit in a Workflow and corresponds to a struct
+// in a CUE instance. This package does not define what a Task looks like in a
+// CUE Instance. Instead, the user of this package must supply a TaskFunc that
+// creates a Runner for cue.Values that are deemed to be a Task.
+//
+// Tasks may depend on other tasks. Cyclic dependencies are thereby not allowed.
+// A Task A depends on another Task B if A, directly or indirectly, has a
+// reference to any field of Task B, including its root.
+//
+// Example:
+//   var inst cue.Instance
+//
+//   // taskFunc takes a Value v and returns a Runner if v is a Task.
+//   w := flow.New(inst, taskFunc, nil)
+//
+//   err := w.Run(context.Background())
+//   if err != nil {
+//       ...
+//   }
+//
+package flow
+
+// TODO: Add hooks. This would allow UIs, for instance, to report on progress.
+//
+// - New(inst *cue.Instance, options ...Option)
+// - AddTask(v cue.Value, r Runner) *Task
+// - AddDependency(a, b *Task)
+// - AddTaskGraph(root cue.Value, fn taskFunc)
+// - AddSequence(list cue.Value, fn taskFunc)
+// - Err()
+
+// TODO:
+// Should we allow lists as a shorthand for a sequence of tasks?
+// If so, how do we specify termination behavior?
+
+// TODO:
+// Should we allow tasks to be a child of another task? Currently, the search
+// for tasks end once a task root is found.
+//
+// Semantically it is somewhat unclear to do so: for instance, if an $after
+// is used to refer to an explicit task dependency, it is logically
+// indistinguishable whether this should be a subtask or is a dependency.
+// Using higher-order constructs for analysis is generally undesirable.
+//
+// A possible solution would be to define specific "grouping tasks" whose sole
+// purpose is to define sub tasks. The user of this package would then need
+// to explicitly distinguish between tasks that are dependencies and tasks that
+// are subtasks.
+
+// TODO: streaming tasks/ server applications
+//
+// Workflows are currently implemented for batch processing, for instance to
+// implement shell scripting or other kinds of batch processing.
+//
+// This API has been designed, however, to also allow for streaming
+// applications. For instance, a streaming Task could listen for Etcd changes
+// or incoming HTTP requests and send updates each time an input changes.
+// Downstream tasks could then alternate between a Waiting and Running state.
+//
+// Note that such streaming applications would also cause configurations to
+// potentially not become increasingly more specific. Instead, a Task would
+// replace its old result each time it is updated. This would require tracking
+// of which conjunct was previously created by a task.
+
+import (
+	"context"
+
+	"cuelang.org/go/cue"
+	"cuelang.org/go/cue/errors"
+	"cuelang.org/go/internal"
+	"cuelang.org/go/internal/core/adt"
+	"cuelang.org/go/internal/core/convert"
+	"cuelang.org/go/internal/core/eval"
+	"cuelang.org/go/internal/core/runtime"
+)
+
+var (
+	// ErrAbort may be returned by a task to avoid processing downstream tasks.
+	// This can be used by control nodes to influence execution.
+	ErrAbort = errors.New("abort dependant tasks without failure")
+
+	// TODO: ErrUpdate: update and run a dependency, but don't complete a
+	// dependency as more results may come. This is useful in server mode.
+)
+
+// A TaskFunc creates a Runner for v if v defines a task or reports nil
+// otherwise. It reports an error for illformed tasks.
+type TaskFunc func(v cue.Value) (Runner, error)
+
+// A Runner executes a Task.
+type Runner interface {
+	// Run runs a Task. If any of the tasks it depends on returned an error it
+	// is passed to this task. It reports an error upon failure.
+	//
+	// Any results to be returned can be set by calling Fill on the passed task.
+	//
+	// TODO: what is a good contract for receiving and passing errors and abort.
+	//
+	// If for a returned error x errors.Is(x, ErrAbort), all dependant tasks
+	// will not be run, without this being an error.
+	Run(t *Task, err error) error
+}
+
+// A RunnerFunc runs a Task.
+type RunnerFunc func(t *Task) error
+
+func (f RunnerFunc) Run(t *Task, err error) error {
+	return f(t)
+}
+
+// A Config defines options for interpreting an Instance as a Workflow.
+type Config struct {
+	// Root limits the search for tasks to be within the path indicated to root.
+	// For the cue command, this is set to ["command"]. The default value is
+	// for all tasks to be root.
+	Root cue.Path
+
+	// InferTasks allows tasks to be defined outside of the Root. Such tasks
+	// will only be included in the workflow if any of its fields is referenced
+	// by any of the tasks defined within Root.
+	InferTasks bool
+
+	// UpdateFunc is called whenever the information in the controller is
+	// updated. This includes directly after initialization. The task may be
+	// nil if this call is not the result of a task completing.
+	UpdateFunc func(c *Controller, t *Task) error
+}
+
+// A Controller defines a set of Tasks to be executed.
+type Controller struct {
+	cfg    Config
+	isTask TaskFunc
+
+	inst        cue.Value
+	valueSeqNum int64
+
+	env *adt.Environment
+
+	conjuncts   []adt.Conjunct
+	conjunctSeq int64
+
+	taskCh chan *Task
+
+	opCtx      *adt.OpContext
+	context    context.Context
+	cancelFunc context.CancelFunc
+
+	// keys maps task keys to their index. This allows a recreation of the
+	// Instance while retaining the original task indices.
+	//
+	// TODO: do instance updating in place to allow for more efficient
+	// processing.
+	keys  map[string]*Task
+	tasks []*Task
+
+	// Only used during task initialization.
+	nodes map[*adt.Vertex]*Task
+
+	errs errors.Error
+}
+
+// Tasks reports the tasks that are currently registered with the controller.
+//
+// This may currently only be called before Run is called or from within
+// a call to UpdateFunc. Task pointers returned by this call are not guaranteed
+// to be the same between successive calls to this method.
+func (c *Controller) Tasks() []*Task {
+	return c.tasks
+}
+
+func (c *Controller) cancel() {
+	if c.cancelFunc != nil {
+		c.cancelFunc()
+	}
+}
+
+func (c *Controller) addErr(err error, msg string) {
+	c.errs = errors.Append(c.errs, errors.Promote(err, msg))
+}
+
+// New creates a Controller for a given Instance and TaskFunc.
+func New(cfg *Config, inst *cue.Instance, f TaskFunc) *Controller {
+	v := inst.Value()
+	rx, nx := internal.CoreValue(v)
+	ctx := eval.NewContext(rx.(*runtime.Runtime), nx.(*adt.Vertex))
+
+	c := &Controller{
+		isTask: f,
+		inst:   v,
+		opCtx:  ctx,
+
+		taskCh: make(chan *Task),
+		keys:   map[string]*Task{},
+	}
+
+	if cfg != nil {
+		c.cfg = *cfg
+	}
+
+	c.initTasks()
+	return c
+
+}
+
+// Run runs the tasks of a workflow until completion.
+func (c *Controller) Run(ctx context.Context) error {
+	c.context, c.cancelFunc = context.WithCancel(ctx)
+	defer c.cancelFunc()
+
+	c.runLoop()
+	return c.errs
+}
+
+// A State indicates the state of a Task.
+//
+// The following state diagram indicates the possible state transitions:
+//
+//          Ready
+//       ↗︎        ↘︎
+//   Waiting  ←  Running
+//       ↘︎        ↙︎
+//       Terminated
+//
+// A Task may move from Waiting to Terminating if one of
+// the tasks on which it dependends fails.
+//
+// NOTE: transitions from Running to Waiting are currently not supported. In
+// the future this may be possible if a task depends on continuously running
+// tasks that send updates.
+//
+type State int
+
+const (
+	// Waiting indicates a task is blocked on input from another task.
+	//
+	// NOTE: although this is currently not implemented, a task could
+	// theoretically move from the Running to Waiting state.
+	Waiting State = iota
+
+	// Ready means a tasks is ready to run, but currently not running.
+	Ready
+
+	// Running indicates a goroutine is currently active for a task and that
+	// it is not Waiting.
+	Running
+
+	// Terminated means a task has stopped running either because it terminated
+	// while Running or was aborted by task on which it depends. The error
+	// value of a Task indicates the reason for the termination.
+	Terminated
+)
+
+var stateStrings = map[State]string{
+	Waiting:    "Waiting",
+	Ready:      "Ready",
+	Running:    "Running",
+	Terminated: "Terminated",
+}
+
+// String reports a human readable string of status s.
+func (s State) String() string {
+	return stateStrings[s]
+}
+
+// A Task contains the context for a single task execution.
+type Task struct {
+	// Static
+	c    *Controller
+	ctxt *adt.OpContext
+	r    Runner
+
+	index  int
+	path   cue.Path
+	key    string
+	labels []adt.Feature
+
+	// Dynamic
+	update adt.Expr
+	deps   map[*Task]bool
+
+	conjunctSeq int64
+	valueSeq    int64
+	v           cue.Value
+	err         errors.Error
+	state       State
+	depTasks    []*Task
+}
+
+// Context reports the Controller's Context.
+func (t *Task) Context() context.Context {
+	return t.c.context
+}
+
+// Path reports the path of Task within the Instance in which it is defined.
+func (t *Task) Path() cue.Path {
+	return t.path
+}
+
+// Index reports the sequence number of the Task. This will not change over
+// time.
+func (t *Task) Index() int {
+	return t.index
+}
+
+func (t *Task) done() bool {
+	return t.state > Running
+}
+
+func (t *Task) isReady() bool {
+	for _, d := range t.depTasks {
+		if !d.done() {
+			return false
+		}
+	}
+	return true
+}
+
+func (t *Task) vertex() *adt.Vertex {
+	_, x := internal.CoreValue(t.v)
+	return x.(*adt.Vertex)
+}
+
+func (t *Task) addDep(dep *Task) {
+	if dep == nil || dep == t {
+		return
+	}
+	if t.deps == nil {
+		t.deps = map[*Task]bool{}
+	}
+	if !t.deps[dep] {
+		t.deps[dep] = true
+		t.depTasks = append(t.depTasks, dep)
+	}
+}
+
+// Fill fills in values of the Controller's configuration for the current task.
+// The changes take effect after the task completes.
+//
+// This method may currently only be called by the runner.
+func (t *Task) Fill(x interface{}) error {
+	expr := convert.GoValueToExpr(t.ctxt, true, x)
+	if t.update == nil {
+		t.update = expr
+		return nil
+	}
+	t.update = &adt.BinaryExpr{
+		Op: adt.AndOp,
+		X:  t.update,
+		Y:  expr,
+	}
+	return nil
+}
+
+// Value reports the latest value of this task.
+//
+// This method may currently only be called before Run is called or after a
+// Task completed, or from within a call to UpdateFunc.
+func (t *Task) Value() cue.Value {
+	// TODO: synchronize
+	return t.v
+}
+
+// Dependencies reports the Tasks t depends on.
+//
+// This method may currently only be called before Run is called or after a
+// Task completed, or from within a call to UpdateFunc.
+func (t *Task) Dependencies() []*Task {
+	// TODO: add synchronization.
+	return t.depTasks
+}
+
+// Err returns the error of a completed Task.
+//
+// This method may currently only be called before Run is called, after a
+// Task completed, or from within a call to UpdateFunc.
+func (t *Task) Err() error {
+	return t.err
+}
+
+// State is the current state of the Task.
+//
+// This method may currently only be called before Run is called or after a
+// Task completed, or from within a call to UpdateFunc.
+func (t *Task) State() State {
+	return t.state
+}