feat(run): devstack-native task-graph runner (spec 31)

docs/guide/command-reference.md

@@ -24,6 +24,7 @@ See [lifecycle.md](lifecycle.md).
 | `up [project...]` | Bring the workspace up (network, shared infra, generate, compose up, hooks) — idempotent saga. | `--build`, `--rebuild`, `--skip-clone`, `--health-timeout`, `--no-hooks`, `--no-preflight`, `--no-provision`, `--profile`/`-p` |
 | `down [project...]` | Stop this workspace's project stacks and release their refs (data preserved). | — |
 | `shell [service] [-- cmd...]` | Open a shell (or run a command) in a service container. | `--project` |
+| `run <task...>` | Run a project's `tasks:` graph (deps-ordered, parallel; host or in-container). | `--project`, `--parallel`, `--dry-run`, `--json` |
 | `status` | Service health + last saga outcome + shared-service ref graph. | — |
 | `logs [service...]` | Stream logs across project + shared stacks (color-keyed). | `--follow`/`-f`, `--tail` (200), `--since`, `--timestamps`, `--no-color` |
 | `dashboard` | Live TUI cockpit: services, health, log tail. | `--no-stats` |

docs/guide/whats-next.md

@@ -57,22 +57,29 @@ Bubble Tea theme, same non-TTY fallback), but it hasn't been built. For now,
 projects are authored by editing `devstack.yaml` directly
 ([projects.md](projects.md)) or scaffolded via `init`.
 
-### "Command-runner" / task projects & monorepo orchestration (Turborepo-style)
-
-**Not supported — this is the genuine, biggest conceptual gap.** devstack
-orchestrates **containers and shared infrastructure**, not a task graph across
-packages. There is:
-
-- no `run:` / `task:` service kind (services are containers, not scripts),
-- no `devstack run <task>` verb, and
-- no dependency-aware, monorepo-aware script runner (nothing like
-  Turborepo/Nx pipelines).
-
-Closing this would be the single largest addition: a new **non-container
-"task"/"script" service kind** (or a `devstack run` verb) plus a monorepo-aware
-task graph. We don't want to overclaim — today, if you need
-`build → test → deploy` task graphs across packages, use your existing task
-runner alongside devstack; devstack handles the infra those tasks talk to.
+### "Command-runner" / task projects & monorepo orchestration — ✅ shipped
+
+**Now built-in.** A `tasks:` block in `devstack.yaml` declares non-container
+commands with `deps:` edges; `devstack run <task>` plans the dependency graph and
+runs it — independent tasks in parallel, output streamed and prefixed per task.
+`run: host` runs on your host toolchain; `run: exec` runs inside a service
+container via `compose exec`. Monorepo/Turborepo pipelines are covered two ways:
+the `turborepo` template runs `turbo run` inside its container, or you map each
+package's scripts into `tasks:` so `devstack run` owns the graph.
+
+```yaml
+# devstack.yaml
+tasks:
+  build: { run: host, command: ["pnpm", "build"] }
+  test:  { run: host, command: ["pnpm", "test"], deps: [build] }
+  lint:  { run: host, command: ["pnpm", "lint"] }
+```
+
+```bash
+devstack run test          # runs build → test
+devstack run test lint     # build+lint in parallel, then test
+devstack run test --dry-run
+```
 
 ### Framework dev servers with watch mode (Next.js, NestJS) — ✅ shipped
 

internal/cli/root.go

@@ -85,6 +85,7 @@ func NewRootCmd(opts Options) *cobra.Command {
 		newStatusCmd(g),
 		newUseCmd(g),
 		newContextCmd(g),
+		newRunCmd(g),
 		newExposeCmd(g),
 		newPortsCmd(g),
 		newShellInitCmd(g),

internal/cli/run.go

@@ -0,0 +1,210 @@
+package cli
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"io"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"runtime"
+	"sort"
+	"strings"
+	"sync"
+
+	"github.com/spf13/cobra"
+	"golang.org/x/sync/errgroup"
+
+	"github.com/open-source-cloud/devstack/internal/config"
+	"github.com/open-source-cloud/devstack/internal/generate"
+	"github.com/open-source-cloud/devstack/internal/task"
+)
+
+// newRunCmd wires `run <task...>` (spec 31): execute a project's task graph. Tasks
+// are non-container commands with `deps` edges; they run in dependency order,
+// independent tasks in parallel (bounded by --parallel). `run: host` runs on the
+// host; `run: exec` runs inside a service container via compose exec. Streams each
+// task's output live, prefixed by task name. `devstack run` takes no flock — it
+// mutates no shared/ledger state.
+func newRunCmd(g *GlobalOpts) *cobra.Command {
+	var project string
+	var parallel int
+	var dryRun bool
+	cmd := &cobra.Command{
+		Use:   "run <task> [task2 ...]",
+		Short: "Run a project's task graph (deps-ordered, parallel; host or in-container)",
+		Args:  cobra.MinimumNArgs(1),
+		RunE: func(cmd *cobra.Command, args []string) error {
+			mgr, closeFn, err := buildManager(cmd)
+			if err != nil {
+				return err
+			}
+			defer closeFn()
+
+			proj := project
+			if proj == "" {
+				proj = resolveActiveProject(mgr.Model, mgr.DB)
+			}
+			if proj == "" {
+				return fmt.Errorf("no project selected (pass --project)")
+			}
+			p, ok := mgr.Model.Projects[proj]
+			if !ok {
+				return fmt.Errorf("unknown project %q", proj)
+			}
+			if len(p.Tasks) == 0 {
+				return fmt.Errorf("project %q declares no tasks: (add a tasks: block to devstack.yaml)", proj)
+			}
+			layers, err := task.Plan(p.Tasks, args)
+			if err != nil {
+				return err
+			}
+			if dryRun {
+				return renderRunPlan(cmd, g, proj, layers)
+			}
+			if parallel <= 0 {
+				parallel = min(8, 2*runtime.NumCPU())
+			}
+			projDir := mgr.Model.ProjectDir(proj)
+			composeFile := filepath.Join(projDir, generate.GenDir, generate.ComposeFile)
+			r := &taskExec{out: cmd.OutOrStdout(), projDir: projDir, composeFile: composeFile}
+			return runLayers(cmd.Context(), r, p.Tasks, layers, parallel)
+		},
+	}
+	cmd.Flags().StringVar(&project, "project", "", "target project (default: the active/first project)")
+	cmd.Flags().IntVar(&parallel, "parallel", 0, "max concurrent tasks (default min(8, 2*CPUs))")
+	cmd.Flags().BoolVar(&dryRun, "dry-run", false, "print the resolved task DAG and exit")
+	return cmd
+}
+
+func renderRunPlan(cmd *cobra.Command, g *GlobalOpts, project string, layers [][]string) error {
+	if g.JSON {
+		return writeJSON(cmd, map[string]any{"project": project, "layers": layers})
+	}
+	w := cmd.OutOrStdout()
+	fmt.Fprintf(w, "run plan for %q (%d layer(s)):\n", project, len(layers))
+	for i, l := range layers {
+		fmt.Fprintf(w, "  %d: %s\n", i+1, strings.Join(l, ", "))
+	}
+	return nil
+}
+
+// runLayers executes each layer in order; tasks within a layer run concurrently
+// up to `parallel`. A failing task fails the run (its layer's siblings finish).
+func runLayers(ctx context.Context, r *taskExec, tasks map[string]config.Task, layers [][]string, parallel int) error {
+	for _, layer := range layers {
+		eg, ectx := errgroup.WithContext(ctx)
+		eg.SetLimit(parallel)
+		for _, name := range layer {
+			name := name
+			t := tasks[name]
+			eg.Go(func() error {
+				if err := r.run(ectx, name, t); err != nil {
+					return fmt.Errorf("task %q: %w", name, err)
+				}
+				return nil
+			})
+		}
+		if err := eg.Wait(); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// taskExec runs a single task with live, name-prefixed output. Concurrent tasks
+// share one output writer, so emit serializes whole (prefix+line) writes under a
+// mutex to keep lines intact and race-free.
+type taskExec struct {
+	mu          sync.Mutex
+	out         io.Writer
+	projDir     string
+	composeFile string
+}
+
+// emit writes prefix+data as one atomic unit under the lock.
+func (r *taskExec) emit(prefix string, data []byte) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	_, _ = io.WriteString(r.out, prefix)
+	_, _ = r.out.Write(data)
+}
+
+func (r *taskExec) run(ctx context.Context, name string, t config.Task) error {
+	prefix := name + " | "
+	pw := &prefixWriter{emit: r.emit, prefix: prefix}
+	env := append(os.Environ(), envKV(t.Env)...)
+
+	var c *exec.Cmd
+	if t.Run == "exec" {
+		if t.Service == "" {
+			return fmt.Errorf("run: exec requires a service")
+		}
+		args := []string{"compose", "-f", r.composeFile, "exec", "-T"}
+		if t.Workdir != "" {
+			args = append(args, "-w", t.Workdir)
+		}
+		for _, kv := range envKV(t.Env) {
+			args = append(args, "-e", kv)
+		}
+		args = append(args, t.Service)
+		args = append(args, t.Command...)
+		c = exec.CommandContext(ctx, "docker", args...)
+		c.Dir = r.projDir
+		c.Env = env
+	} else {
+		c = exec.CommandContext(ctx, t.Command[0], t.Command[1:]...)
+		c.Dir = taskWorkdir(r.projDir, t.Workdir)
+		c.Env = env
+	}
+	c.Stdout = pw
+	c.Stderr = pw
+	r.emit(prefix, []byte("→ "+strings.Join(t.Command, " ")+"\n"))
+	return c.Run()
+}
+
+func taskWorkdir(projDir, workdir string) string {
+	if workdir == "" {
+		return projDir
+	}
+	if filepath.IsAbs(workdir) {
+		return workdir
+	}
+	return filepath.Join(projDir, workdir)
+}
+
+func envKV(m map[string]string) []string {
+	if len(m) == 0 {
+		return nil
+	}
+	out := make([]string, 0, len(m))
+	for k, v := range m {
+		out = append(out, k+"="+v)
+	}
+	sort.Strings(out)
+	return out
+}
+
+// prefixWriter buffers bytes and flushes each complete line through emit, so a
+// line and its prefix are written atomically even under concurrent tasks.
+type prefixWriter struct {
+	emit   func(prefix string, data []byte)
+	prefix string
+	buf    []byte
+}
+
+func (p *prefixWriter) Write(b []byte) (int, error) {
+	p.buf = append(p.buf, b...)
+	for {
+		i := bytes.IndexByte(p.buf, '\n')
+		if i < 0 {
+			break
+		}
+		line := make([]byte, i+1)
+		copy(line, p.buf[:i+1])
+		p.emit(p.prefix, line)
+		p.buf = p.buf[i+1:]
+	}
+	return len(b), nil
+}

internal/cli/run_test.go

@@ -0,0 +1,53 @@
+package cli
+
+import (
+	"bytes"
+	"context"
+	"strings"
+	"testing"
+
+	"github.com/open-source-cloud/devstack/internal/config"
+	"github.com/open-source-cloud/devstack/internal/task"
+)
+
+func TestRunRegistered(t *testing.T) {
+	if !findCmd(t, "run") {
+		t.Fatal("run must be a real RunE command")
+	}
+}
+
+func TestRunLayersOrder(t *testing.T) {
+	tasks := map[string]config.Task{
+		"build": {Run: "host", Command: []string{"sh", "-lc", "echo BUILD"}},
+		"lint":  {Run: "host", Command: []string{"sh", "-lc", "echo LINT"}},
+		"test":  {Run: "host", Command: []string{"sh", "-lc", "echo TEST"}, Deps: []string{"build"}},
+		"ci":    {Run: "host", Command: []string{"sh", "-lc", "echo CI"}, Deps: []string{"test", "lint"}},
+	}
+	layers, err := task.Plan(tasks, []string{"ci"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	var buf bytes.Buffer
+	r := &taskExec{out: &buf, projDir: t.TempDir()}
+	if err := runLayers(context.Background(), r, tasks, layers, 4); err != nil {
+		t.Fatalf("run: %v", err)
+	}
+	out := buf.String()
+	// Dependency order: BUILD before TEST, TEST before CI, LINT before CI.
+	for _, pair := range [][2]string{{"BUILD", "TEST"}, {"TEST", "CI"}, {"LINT", "CI"}} {
+		if strings.Index(out, pair[0]) >= strings.Index(out, pair[1]) {
+			t.Errorf("%s should run before %s\n%s", pair[0], pair[1], out)
+		}
+	}
+}
+
+func TestRunLayersPropagatesFailure(t *testing.T) {
+	tasks := map[string]config.Task{
+		"boom": {Run: "host", Command: []string{"sh", "-lc", "exit 3"}},
+	}
+	layers, _ := task.Plan(tasks, []string{"boom"})
+	r := &taskExec{out: &bytes.Buffer{}, projDir: t.TempDir()}
+	if err := runLayers(context.Background(), r, tasks, layers, 1); err == nil {
+		t.Fatal("a failing task must fail the run")
+	}
+}

internal/config/model.go

@@ -143,6 +143,23 @@ type Project struct {
 	Services   map[string]Service `yaml:"services" validate:"required,dive"`
 	Hooks      Hooks              `yaml:"hooks"`                     // spec 11 — project-scope lifecycle hooks
 	Resources  []ResourceDecl     `yaml:"resources" validate:"dive"` // spec 27 — declarative data-plane resources
+	Tasks      map[string]Task    `yaml:"tasks" validate:"dive"`     // spec 31 — non-container task graph (`devstack run`)
+}
+
+// Task is one node in a project's task graph (spec 31): a short-lived command run
+// on demand by `devstack run`, NOT a container. `run: host` executes on the host
+// (inheriting your toolchain); `run: exec` runs inside a service container via
+// `compose exec`. `deps` are other task names that must complete first; the graph
+// is executed in dependency order (cycles are rejected at run time). `watch` marks
+// long-running dev-server tasks that `--watch` keeps alive.
+type Task struct {
+	Command []string          `yaml:"command" validate:"required,min=1"`
+	Run     string            `yaml:"run" validate:"omitempty,oneof=host exec"` // default host
+	Service string            `yaml:"service"`                                  // target for run:exec
+	Deps    []string          `yaml:"deps"`
+	Workdir string            `yaml:"workdir"`
+	Env     map[string]string `yaml:"env"`
+	Watch   bool              `yaml:"watch"`
 }
 
 // ResourceDecl is one declarative data-plane resource a project needs INSIDE a