feat(specialist): run independent plan tasks in parallel (gap report §5.1)

Admission, scheduler and worker cap in ONE commit, because splitting them the
way the plan suggested would have shipped a regression: relaxing max_workers
while the executor stayed sequential means a plan asking for 8 silently runs
with 1, which is exactly what the old rejection existed to prevent.

ONE SCHEDULER, NOT TWO. With a single worker it walks plan.Order() one task at
a time, applying the same checks in the same order and producing the same
report — so the entire existing suite, written against the sequential executor,
IS the equivalence oracle. It passed unchanged except for the one test that
encoded "max_workers must be 1". A second executor would have been easier to
write and impossible to keep in step; a duplicated rule drifts, and a duplicated
executor drifts faster.

THE WALK IS WHAT PRESERVES ORDER. Rather than keeping a ready SET and picking
from it — which would let task 7 start before task 3 because its dependencies
happened to resolve first — the scheduler walks the validated topological order
and, for each task in turn, waits until that task's dependencies have resolved
AND a worker is free. With one worker the wait is "until the previous task
finished", which is today exactly.

TWO NUMBERS, BOTH REPORTED. max_workers is what the plan asked for; the machine
may carry less. A plan that asked for sixteen and ran six has not been given
sixteen, so the report carries Workers and WorkersRequested rather than letting
the request stand as if honoured. Out of range is REJECTED, not trimmed, for the
same reason the old rule rejected: a trimmed number is one nobody can reason
about afterwards.

THE TUI STOPS ATTRIBUTING RATHER THAN ATTRIBUTING WRONGLY. Child progress events
carry no task identity — the loop's callback holds only the parent's tool-call
id — so with two tasks in flight there is nothing to route them by. Pointing
every child's output at whichever task started last would show one task's tool
calls under another's name. A card that says nothing is honest; a card that says
something false is not. Per-task progress needs the child's identity threaded
through the callback, and that is its own change.

Eight mutations, all caught, including the two that would be invisible in
production: dependencies ignored, and one worker overlapping anyway.
Race detector clean at -count=10.

MEASURED IN THE REAL BINARY, same plan, same build — a five-task diamond:

    max_workers=1   WALL 40.0s
    max_workers=4   WALL 27.0s

Note what this spends: MaxSpeedup was defined as the theoretical ceiling
measured FROM a sequential run, and it was the evidence the decision to build
this rested on. Once plans run parallel that instrument no longer measures a
ceiling. The 1.67x it reports for this diamond still matches the observed
40→27; on plans that are mostly chain it will not.

internal/specialist/plan.go

@@ -56,9 +56,11 @@ type Task struct {
 // Budget bounds a plan. Every field is required to be sane at admission and
 // MaxTokens is enforced again at dispatch.
 type Budget struct {
-	// MaxWorkers must be 1. Phase 2 is sequential, and a plan asking for more is
-	// REJECTED rather than coerced — coercion would let a caller believe it got
-	// concurrency, and would make the field meaningless for Phase 3.
+	// MaxWorkers is how many tasks the plan may run at once, 1 to maxPlanWorkers.
+	//
+	// It is what the PLAN asked for, not what it will get: the machine's own
+	// capacity may be lower, and the report says which number actually applied
+	// rather than letting the request stand as if it were honoured.
 	MaxWorkers int
 	MaxTokens  int
 	MaxWall    time.Duration
@@ -504,13 +506,22 @@ func planBudget(args map[string]any, limits Limits) (Budget, error) {
 		}
 		budget.MaxRetries = retries
 	}
-	// MaxWorkers must be exactly 1. Rejecting rather than coercing keeps the
-	// field meaningful: a caller that asked for 8 and silently got 1 would have
-	// been told nothing, and Phase 3 would inherit a field nobody trusts.
-	if budget.MaxWorkers != 1 {
+	// MaxWorkers is a RANGE now, and still rejected rather than coerced outside
+	// it. The rule it replaces — "must be exactly 1" — existed because the
+	// executor was sequential and a caller that asked for 8 and silently got 1
+	// would have been told nothing. That reasoning is unchanged; only the
+	// executor changed, so the bound moved rather than dissolved.
+	//
+	// The ceiling is absolute and small. A plan is one tool call, and every task
+	// it runs is a child process inheriting a 320-turn budget under this
+	// posture; sixteen of those at once is already a great deal of machine and
+	// a great deal of money. A plan asking for more is refused, not trimmed,
+	// because a trimmed number is a number nobody can reason about afterwards.
+	if budget.MaxWorkers < 1 || budget.MaxWorkers > maxPlanWorkers {
 		return Budget{}, fmt.Errorf(
-			"budget.max_workers must be 1: this phase executes plan tasks sequentially, and a plan asking for %d is rejected rather than quietly run with one worker",
-			budget.MaxWorkers)
+			"budget.max_workers must be between 1 and %d; %d was requested. "+
+				"1 runs the plan sequentially, which is the right answer unless its tasks are genuinely independent",
+			maxPlanWorkers, budget.MaxWorkers)
 	}
 	// max_tokens is OPTIONAL and unbounded by default.
 	//

internal/specialist/plan_concurrent_test.go

@@ -0,0 +1,268 @@
+package specialist
+
+import (
+	"context"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+// concurrencyProbe records the maximum number of tasks in flight at once, and
+// the order dispatches began.
+type concurrencyProbe struct {
+	mu       sync.Mutex
+	inFlight int
+	peak     int
+	started  []string
+	release  chan struct{}
+}
+
+func (p *concurrencyProbe) runner() PlanRunner {
+	return func(_ context.Context, req PlanTaskRequest) (TaskResult, error) {
+		p.mu.Lock()
+		p.inFlight++
+		if p.inFlight > p.peak {
+			p.peak = p.inFlight
+		}
+		p.started = append(p.started, req.Task.ID)
+		p.mu.Unlock()
+
+		if p.release != nil {
+			<-p.release
+		}
+		p.mu.Lock()
+		p.inFlight--
+		p.mu.Unlock()
+		return TaskResult{Outcome: TaskSucceeded}, nil
+	}
+}
+
+func fanOutPlan(t *testing.T, workers int, n int) Plan {
+	t.Helper()
+	tasks := make([]any, 0, n)
+	for i := 0; i < n; i++ {
+		tasks = append(tasks, task(string(rune('a'+i)), "x"))
+	}
+	budget := map[string]any{"max_workers": float64(workers)}
+	return mustPlan(t, tasks, budget, readOnlyLimits())
+}
+
+// ONE WORKER IS STILL ONE AT A TIME. The whole equivalence claim rests on this:
+// there is one scheduler, and with a single worker it must never overlap.
+func TestOneWorkerNeverOverlaps(t *testing.T) {
+	probe := &concurrencyProbe{}
+	report := ExecutePlan(context.Background(), fanOutPlan(t, 1, 6), []string{"read_file"}, probe.runner(), nil)
+	if probe.peak != 1 {
+		t.Fatalf("peak concurrency %d with one worker; it must be 1", probe.peak)
+	}
+	if report.Succeeded != 6 {
+		t.Fatalf("report = %+v", report)
+	}
+	if report.Workers != 1 || report.WorkersRequested != 1 {
+		t.Fatalf("workers reported %d/%d", report.Workers, report.WorkersRequested)
+	}
+}
+
+// ...and one worker dispatches in the plan's validated order, which is what
+// makes the sequential path reproducible rather than merely serial.
+func TestOneWorkerDispatchesInPlanOrder(t *testing.T) {
+	probe := &concurrencyProbe{}
+	plan := fanOutPlan(t, 1, 6)
+	ExecutePlan(context.Background(), plan, []string{"read_file"}, probe.runner(), nil)
+	for index, id := range plan.Order() {
+		if probe.started[index] != id {
+			t.Fatalf("dispatch order %v, want the plan's order %v", probe.started, plan.Order())
+		}
+	}
+}
+
+// TASKS ACTUALLY RUN AT ONCE. Asserting only that a plan with 4 workers
+// completes would pass against a scheduler that ignored the setting entirely.
+func TestIndependentTasksRunConcurrently(t *testing.T) {
+	probe := &concurrencyProbe{release: make(chan struct{})}
+	plan := fanOutPlan(t, 4, 4)
+
+	done := make(chan PlanReport, 1)
+	go func() {
+		done <- ExecutePlan(context.Background(), plan, []string{"read_file"}, probe.runner(), nil)
+	}()
+
+	// Wait for all four to be in flight together, which can only happen if they
+	// genuinely overlap.
+	deadline := time.After(5 * time.Second)
+	for {
+		probe.mu.Lock()
+		peak := probe.peak
+		probe.mu.Unlock()
+		if peak >= 4 {
+			break
+		}
+		select {
+		case <-deadline:
+			t.Fatalf("peak concurrency reached only %d of 4", peak)
+		case <-time.After(5 * time.Millisecond):
+		}
+	}
+	close(probe.release)
+	report := <-done
+	if report.Succeeded != 4 {
+		t.Fatalf("report = %+v", report)
+	}
+}
+
+// DEPENDENCIES STILL HOLD. Concurrency may overlap independent work and must
+// never start a task before what it waits on has finished.
+func TestADependentTaskNeverStartsEarly(t *testing.T) {
+	// a -> (b, c) -> d, with room to run everything at once if the scheduler
+	// forgot the edges.
+	plan := mustPlan(t, []any{
+		task("a", "root"), task("b", "left", "a"), task("c", "right", "a"), task("d", "join", "b", "c"),
+	}, map[string]any{"max_workers": float64(4)}, readOnlyLimits())
+
+	var mu sync.Mutex
+	finished := map[string]bool{}
+	var violations []string
+
+	ExecutePlan(context.Background(), plan, []string{"read_file"},
+		func(_ context.Context, req PlanTaskRequest) (TaskResult, error) {
+			mu.Lock()
+			for _, dep := range req.Task.DependsOn {
+				if !finished[dep] {
+					violations = append(violations, req.Task.ID+" started before "+dep)
+				}
+			}
+			mu.Unlock()
+			time.Sleep(5 * time.Millisecond)
+			mu.Lock()
+			finished[req.Task.ID] = true
+			mu.Unlock()
+			return TaskResult{Outcome: TaskSucceeded}, nil
+		}, nil)
+
+	if len(violations) > 0 {
+		t.Fatalf("dependency order broken: %v", violations)
+	}
+}
+
+// A DIAMOND OVERLAPS ITS MIDDLE. b and c are independent, so the scheduler must
+// run them together — otherwise concurrency is wired but inert.
+func TestADiamondsIndependentTasksOverlap(t *testing.T) {
+	plan := mustPlan(t, []any{
+		task("a", "root"), task("b", "left", "a"), task("c", "right", "a"), task("d", "join", "b", "c"),
+	}, map[string]any{"max_workers": float64(4)}, readOnlyLimits())
+
+	var mu sync.Mutex
+	inFlight, peak := 0, 0
+	ExecutePlan(context.Background(), plan, []string{"read_file"},
+		func(_ context.Context, req PlanTaskRequest) (TaskResult, error) {
+			mu.Lock()
+			inFlight++
+			if inFlight > peak {
+				peak = inFlight
+			}
+			mu.Unlock()
+			time.Sleep(20 * time.Millisecond)
+			mu.Lock()
+			inFlight--
+			mu.Unlock()
+			return TaskResult{Outcome: TaskSucceeded}, nil
+		}, nil)
+
+	if peak < 2 {
+		t.Fatalf("peak concurrency %d on a diamond; b and c are independent and must overlap", peak)
+	}
+}
+
+// THE WORKER COUNT IS A CAP. A plan asking for 2 must never run 3, however many
+// tasks are ready.
+func TestTheWorkerCountIsRespected(t *testing.T) {
+	probe := &concurrencyProbe{}
+	report := ExecutePlan(context.Background(), fanOutPlan(t, 2, 8), []string{"read_file"}, probe.runner(), nil)
+	if probe.peak > 2 {
+		t.Fatalf("peak concurrency %d exceeded the 2 workers asked for", probe.peak)
+	}
+	if report.Succeeded != 8 {
+		t.Fatalf("report = %+v", report)
+	}
+}
+
+// THE MACHINE'S CAPACITY BOUNDS THE REQUEST, and both numbers are reported. A
+// plan that asked for sixteen and ran six has not been given sixteen.
+func TestTheEffectiveWorkerCountIsReported(t *testing.T) {
+	if got := effectivePlanWorkers(1); got != 1 {
+		t.Fatalf("one worker must stay one on every host, got %d", got)
+	}
+	if got := effectivePlanWorkers(maxPlanWorkers); got > machinePlanWorkers() {
+		t.Fatalf("effective workers %d exceeds the machine's %d", got, machinePlanWorkers())
+	}
+	if machinePlanWorkers() < minPlanWorkers {
+		t.Fatalf("machine workers %d fell below the floor", machinePlanWorkers())
+	}
+
+	report := ExecutePlan(context.Background(), fanOutPlan(t, maxPlanWorkers, 2), []string{"read_file"},
+		func(context.Context, PlanTaskRequest) (TaskResult, error) {
+			return TaskResult{Outcome: TaskSucceeded}, nil
+		}, nil)
+	if report.WorkersRequested != maxPlanWorkers {
+		t.Fatalf("requested %d, want %d", report.WorkersRequested, maxPlanWorkers)
+	}
+	if report.Workers > report.WorkersRequested || report.Workers < 1 {
+		t.Fatalf("effective workers %d is not a bound on %d", report.Workers, report.WorkersRequested)
+	}
+}
+
+// A PANICKING TASK MUST NOT HANG THE PLAN. The slot has to come back, or the
+// scheduler waits forever for a worker that will never report.
+func TestAPanickingTaskFreesItsWorker(t *testing.T) {
+	plan := fanOutPlan(t, 2, 4)
+	var runs atomic.Int32
+	done := make(chan PlanReport, 1)
+	go func() {
+		done <- ExecutePlan(context.Background(), plan, []string{"read_file"},
+			func(_ context.Context, req PlanTaskRequest) (TaskResult, error) {
+				if runs.Add(1) == 1 {
+					panic("boom")
+				}
+				return TaskResult{Outcome: TaskSucceeded}, nil
+			}, nil)
+	}()
+	select {
+	case report := <-done:
+		if report.Failed != 1 {
+			t.Fatalf("report = %+v; the panicking task must be recorded as failed", report)
+		}
+		if report.Succeeded != 3 {
+			t.Fatalf("report = %+v; the other tasks must still run", report)
+		}
+	case <-time.After(10 * time.Second):
+		t.Fatal("a panicking task hung the plan; its worker slot was never returned")
+	}
+}
+
+// EVERY TASK IS HARVESTED before the report is assembled. A plan that reported
+// while work was still in flight would report on tasks that had not finished.
+func TestNoTaskIsLeftInFlightWhenThePlanReports(t *testing.T) {
+	plan := fanOutPlan(t, 4, 12)
+	var running atomic.Int32
+	report := ExecutePlan(context.Background(), plan, []string{"read_file"},
+		func(context.Context, PlanTaskRequest) (TaskResult, error) {
+			running.Add(1)
+			time.Sleep(2 * time.Millisecond)
+			running.Add(-1)
+			return TaskResult{Outcome: TaskSucceeded}, nil
+		}, nil)
+	if left := running.Load(); left != 0 {
+		t.Fatalf("%d tasks still running when the plan reported", left)
+	}
+	if len(report.Tasks) != 12 || report.Succeeded != 12 {
+		t.Fatalf("report = %+v", report)
+	}
+	// The report is in the PLAN's order, not completion order — a concurrent run
+	// finishes out of order and the record must not.
+	for index, id := range plan.Order() {
+		if report.Tasks[index].ID != id {
+			t.Fatalf("report order %v, want plan order %v", report.Tasks[index].ID, id)
+		}
+	}
+}

internal/specialist/plan_exec.go

@@ -107,6 +107,14 @@ type PlanReport struct {
 	// 0 regardless and would decide nothing.
 	MaxSpeedup float64
 	TokensUsed int
+	// Workers is how many tasks this plan ACTUALLY ran at once, and
+	// WorkersRequested is what it asked for. Both, because the machine's
+	// capacity may be lower than the request and a plan that asked for sixteen
+	// and ran six has not been given sixteen — reporting only the request would
+	// make the number a fiction, which is the same reason max_workers is
+	// rejected outside its range rather than trimmed into it.
+	Workers          int
+	WorkersRequested int
 }
 
 // PlanTaskRequest is everything one task needs to run.
@@ -214,9 +222,88 @@ func ExecutePlanIn(ctx context.Context, plan Plan, workspace PlanWorkspace, pare
 	stallTimeout := stallTimeoutFor(plan.Budget())
 
 	cancelled := false
+
+	// THE WORKER POOL. One worker is the sequential path: every wait below
+	// becomes "until the previous task finished", and the walk applies the same
+	// checks in the same order it always has.
+	workers := effectivePlanWorkers(plan.Budget().MaxWorkers)
+	report.Workers = workers
+	report.WorkersRequested = plan.Budget().MaxWorkers
+	slots := newPlanSlots(workers)
+
+	// harvest applies one completed dispatch: its spend, its outcome, its
+	// record. Called only from THIS goroutine, so results, failed, report and
+	// budgetLeft are never touched concurrently and need no lock — the
+	// concurrency is in the dispatches, not in the bookkeeping.
+	harvest := func(completion taskCompletion) {
+		slots.release()
+		id, result, err := completion.id, completion.result, completion.err
+		result.ID = id
+		if result.Duration == 0 {
+			result.Duration = time.Since(completion.started)
+		}
+		report.SequentialTotal += result.Duration
+		budgetLeft -= result.Tokens
+		report.TokensUsed += result.Tokens
+
+		if err != nil || result.Outcome == TaskFailed {
+			// A task cut short by cancellation is CANCELLED, not failed. The
+			// distinction survives all the way to the terminal status, so a
+			// stopped plan never reports as a broken one.
+			if ctx.Err() != nil || errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
+				cancelled = true
+				result.Outcome = TaskCancelled
+				if result.Err == "" {
+					result.Err = "cancelled: the run was stopped while this task was running"
+				}
+				results[id] = result
+				failed[id] = true
+				report.Cancelled++
+				recordFailed(recorder, result)
+				return
+			}
+			result.Outcome = TaskFailed
+			if result.Err == "" && err != nil {
+				result.Err = err.Error()
+			}
+			results[id] = result
+			failed[id] = true
+			report.Failed++
+			recordFailed(recorder, result)
+			return
+		}
+		result.Outcome = TaskSucceeded
+		results[id] = result
+		report.Succeeded++
+		recordCompleted(recorder, result)
+	}
+
+	// resolved reports whether every dependency of a task has finished, one way
+	// or another. A task waits for its dependencies to RESOLVE, not to succeed:
+	// a failed dependency resolves it too, and the skip check below turns that
+	// into the recorded dependency_failed.
+	resolved := func(task Task) bool {
+		for _, dep := range task.DependsOn {
+			if _, done := results[dep]; !done {
+				return false
+			}
+		}
+		return true
+	}
+
 	for _, id := range plan.Order() {
 		task := tasks[id]
 
+		// WAIT FOR THIS TASK'S TURN: until its dependencies have resolved and a
+		// worker is free. Waiting on the task the WALK reached — rather than
+		// picking whichever ready task appeared first — is what keeps dispatch
+		// in the plan's validated order, so one worker reproduces the sequential
+		// executor exactly and more workers only overlap what was already
+		// adjacent.
+		for slots.busy() && (!resolved(task) || slots.full()) {
+			harvest(<-slots.done)
+		}
+
 		// PAUSE, at the task boundary, BEFORE the cancellation check — so a user
 		// who stops a paused plan is not left waiting for a resume that will
 		// never come. WaitWhilePaused returns on ctx, and the check below then
@@ -286,49 +373,37 @@ func ExecutePlanIn(ctx context.Context, plan Plan, workspace PlanWorkspace, pare
 		}
 
 		recordDispatched(recorder, task)
-		result, err := runTaskWithRetries(ctx, retryPolicy{
+		policy := retryPolicy{
 			task:         task,
 			tools:        granted,
 			cwd:          workspace.Path,
 			stallTimeout: stallTimeout,
 			maxRetries:   plan.Budget().MaxRetries,
 			deadline:     deadline,
-		}, run)
-		result.ID = id
-		report.SequentialTotal += result.Duration
-		budgetLeft -= result.Tokens
-		report.TokensUsed += result.Tokens
-
-		if err != nil || result.Outcome == TaskFailed {
-			// A task cut short by cancellation is CANCELLED, not failed. The
-			// distinction survives all the way to the terminal status, so a
-			// stopped plan never reports as a broken one.
-			if ctx.Err() != nil || errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
-				cancelled = true
-				result.Outcome = TaskCancelled
-				if result.Err == "" {
-					result.Err = "cancelled: the run was stopped while this task was running"
-				}
-				results[id] = result
-				failed[id] = true
-				report.Cancelled++
-				recordFailed(recorder, result)
-				continue
-			}
-			result.Outcome = TaskFailed
-			if result.Err == "" && err != nil {
-				result.Err = err.Error()
-			}
-			results[id] = result
-			failed[id] = true
-			report.Failed++
-			recordFailed(recorder, result)
-			continue
 		}
-		result.Outcome = TaskSucceeded
-		results[id] = result
-		report.Succeeded++
-		recordCompleted(recorder, result)
+		slots.take()
+		started := time.Now()
+		go func(id string) {
+			// Every goroutine gets recover(): a panic in one task must not take
+			// the plan with it, and the slot must be returned either way or the
+			// scheduler waits forever on a worker that will never report.
+			defer func() {
+				if panicked := recover(); panicked != nil {
+					slots.done <- taskCompletion{
+						id: id, started: started,
+						result: TaskResult{Outcome: TaskFailed, Err: fmt.Sprintf("task panicked: %v", panicked)},
+					}
+				}
+			}()
+			result, err := runTaskWithRetries(ctx, policy, run)
+			slots.done <- taskCompletion{id: id, result: result, err: err, started: started}
+		}(id)
+	}
+
+	// DRAIN. Everything still in flight is harvested before the report is
+	// assembled, or a plan would report on tasks that had not finished.
+	for slots.busy() {
+		harvest(<-slots.done)
 	}
 
 	for _, id := range plan.Order() {

internal/specialist/plan_schedule.go

@@ -0,0 +1,133 @@
+package specialist
+
+import (
+	"runtime"
+	"sync"
+	"time"
+)
+
+// How many tasks a plan runs at once.
+//
+// THE SEQUENTIAL PATH IS NOT A SEPARATE PATH. There is one scheduler, and with
+// one worker it walks plan.Order() one task at a time, applying the same checks
+// in the same order and producing the same report — which is why every test
+// written against the sequential executor is the oracle for this. Two executors
+// would have been safer to write and impossible to keep in step: invariant 5
+// says a duplicated rule drifts, and a duplicated EXECUTOR drifts faster.
+//
+// The walk is the thing that preserves order. Rather than maintaining a ready
+// SET and choosing from it — which would let task 7 start before task 3 merely
+// because its dependencies resolved first — the scheduler walks the validated
+// topological order and, for each task in turn, waits until that task's
+// dependencies are resolved and a worker is free. With one worker the wait is
+// "until the previous task finished", which is exactly today.
+
+const (
+	// maxPlanWorkers is the absolute ceiling on what a plan may ASK for. Small
+	// on purpose: each task is a child process inheriting a 320-turn budget
+	// under this posture.
+	maxPlanWorkers = 16
+	// minPlanWorkers keeps the machine cap from collapsing to zero on a
+	// single-core box, where the answer must still be "run the plan".
+	minPlanWorkers = 2
+)
+
+// machinePlanWorkers is what the HOST can carry, independent of what a plan
+// asked for. Leaving two cores is not superstition: the parent agent, the TUI
+// render loop and every child's own I/O all want one.
+func machinePlanWorkers() int {
+	available := runtime.NumCPU() - 2
+	if available < minPlanWorkers {
+		available = minPlanWorkers
+	}
+	if available > maxPlanWorkers {
+		available = maxPlanWorkers
+	}
+	return available
+}
+
+// effectivePlanWorkers is what the plan will ACTUALLY run with: the smaller of
+// what it asked for and what the machine can carry.
+//
+// The two numbers are kept apart and both reported. A plan that asked for
+// sixteen and ran six has not been given sixteen, and saying so is the
+// difference between a bound and a fiction — the same reason max_workers is
+// rejected outside its range rather than trimmed into it.
+func effectivePlanWorkers(requested int) int {
+	if requested < 1 {
+		requested = 1
+	}
+	if requested == 1 {
+		// Never consult the machine for a sequential plan. One worker must mean
+		// one worker on every host, or the sequential path stops being
+		// reproducible.
+		return 1
+	}
+	if machine := machinePlanWorkers(); requested > machine {
+		return machine
+	}
+	return requested
+}
+
+// planSlots is the worker pool: a counting semaphore plus the completions.
+//
+// A channel of results rather than a WaitGroup, because the scheduler needs to
+// ACT on each completion as it lands — a finished task unblocks its dependents
+// and returns its spend to the budget — and a WaitGroup only says "all done".
+type planSlots struct {
+	limit    int
+	inFlight int
+	done     chan taskCompletion
+	mu       sync.Mutex
+}
+
+// taskCompletion is one finished dispatch on its way back to the scheduler.
+type taskCompletion struct {
+	id     string
+	result TaskResult
+	err    error
+	// started is when the dispatch began, so a runner that reported no duration
+	// is measured by the scheduler rather than recorded as instantaneous.
+	started time.Time
+}
+
+func newPlanSlots(limit int) *planSlots {
+	if limit < 1 {
+		limit = 1
+	}
+	return &planSlots{limit: limit, done: make(chan taskCompletion, limit)}
+}
+
+// full reports whether every worker is busy.
+func (slots *planSlots) full() bool {
+	slots.mu.Lock()
+	defer slots.mu.Unlock()
+	return slots.inFlight >= slots.limit
+}
+
+// busy reports whether anything is still running, which is what tells the
+// scheduler it must keep draining before it can finish.
+func (slots *planSlots) busy() bool {
+	slots.mu.Lock()
+	defer slots.mu.Unlock()
+	return slots.inFlight > 0
+}
+
+// take claims a worker slot.
+func (slots *planSlots) take() {
+	slots.mu.Lock()
+	slots.inFlight++
+	slots.mu.Unlock()
+}
+
+// release frees a slot. Called by the scheduler when it harvests a completion,
+// not by the goroutine that produced it — so a slot is free exactly when the
+// scheduler has finished acting on the result, never while it is still applying
+// it to the budget.
+func (slots *planSlots) release() {
+	slots.mu.Lock()
+	if slots.inFlight > 0 {
+		slots.inFlight--
+	}
+	slots.mu.Unlock()
+}

internal/specialist/plan_test.go

@@ -144,16 +144,33 @@ func TestParsePlanRejectsIDsOutsideTheAllowList(t *testing.T) {
 
 // (i) MaxWorkers > 1 is REJECTED, not coerced. Coercion would let a caller
 // believe it got concurrency and would make the field meaningless for Phase 3.
-func TestParsePlanRejectsMoreThanOneWorker(t *testing.T) {
-	for _, workers := range []float64{0, 2, 8} {
+func TestMaxWorkersIsARangeAndIsRejectedOutsideIt(t *testing.T) {
+	// THE RULE MOVED, IT DID NOT DISSOLVE. It used to be "must be exactly 1",
+	// because the executor was sequential and a caller who asked for 8 and
+	// silently got 1 would have been told nothing. The executor changed; that
+	// reasoning did not, so the bound became a range and is still REJECTED
+	// outside it rather than trimmed into it — a trimmed number is one nobody
+	// can reason about afterwards.
+	for _, workers := range []float64{0, -1, maxPlanWorkers + 1, 100} {
 		budget := okBudget()
 		budget["max_workers"] = workers
 		_, err := ParsePlan(planArgs([]any{task("a", "x")}, budget), readOnlyLimits())
 		if err == nil {
 			t.Fatalf("max_workers %v must be rejected", workers)
 		}
-		if !strings.Contains(err.Error(), "sequentially") {
-			t.Fatalf("the error must say why: %v", err)
+		if !strings.Contains(err.Error(), "between 1 and") {
+			t.Fatalf("the error must state the range: %v", err)
+		}
+	}
+	for _, workers := range []float64{1, 2, 8, maxPlanWorkers} {
+		budget := okBudget()
+		budget["max_workers"] = workers
+		plan, err := ParsePlan(planArgs([]any{task("a", "x")}, budget), readOnlyLimits())
+		if err != nil {
+			t.Fatalf("max_workers %v must be accepted: %v", workers, err)
+		}
+		if plan.Budget().MaxWorkers != int(workers) {
+			t.Fatalf("max_workers parsed as %d, want %v", plan.Budget().MaxWorkers, workers)
 		}
 	}
 }

internal/specialist/plan_tool.go

@@ -88,7 +88,7 @@ func (tool *OrchestrateTool) Name() string { return OrchestrateToolName }
 
 func (tool *OrchestrateTool) Description() string {
 	return "Execute a structured plan of read-only sub-agent tasks in dependency order. " +
-		"Tasks run sequentially; declare dependencies with depends_on so the plan records which work was independent. " +
+		"Independent tasks run in parallel up to budget.max_workers; declare dependencies with depends_on and a task never starts before what it waits on has finished. " +
 		// The either/or the schema cannot express, and the shape worth
 		// encouraging: a plan is only worth more than reading the code yourself
 		// when its tasks are genuinely independent.
@@ -149,7 +149,7 @@ func (tool *OrchestrateTool) Parameters() tools.Schema {
 			},
 			"budget": {
 				Type:        "object",
-				Description: "Required. max_workers must be 1 (this phase executes sequentially). max_tokens and max_wall_seconds are optional bounds; omit them to run unbounded — spend is reported either way. max_stall_seconds bounds how long ONE task may emit nothing (default 180); it resets on every event, so a slow-but-working task is never stopped. max_retries (0-3, default 1) is how many extra attempts a STALLED task gets; a task that failed with a real error is never retried.",
+				Description: "Required. max_workers (1-16) is how many tasks may run at once; 1 is sequential and is the right answer unless the tasks are genuinely independent. The machine's own capacity may be lower and the report says which number applied. max_tokens and max_wall_seconds are optional bounds; omit them to run unbounded — spend is reported either way. max_stall_seconds bounds how long ONE task may emit nothing (default 180); it resets on every event, so a slow-but-working task is never stopped. max_retries (0-3, default 1) is how many extra attempts a STALLED task gets; a task that failed with a real error is never retried.",
 			},
 		},
 		// EMPTY, and the either/or lives in the DESCRIPTION instead.
@@ -290,9 +290,13 @@ func (tool *OrchestrateTool) Run(ctx context.Context, args map[string]any) tools
 //
 // The callback is shared by every task rather than keyed per task, because the
 // loop's callback carries only the parent's tool-call id and has no room for a
-// sub-key. That is sound HERE and only here: MaxWorkers is validated to be 1,
+// sub-key. THE CONSEQUENCE, now that plans can be concurrent: with more than one
+// worker a consumer CANNOT attribute an event to a task, and the TUI stops
+// trying rather than attributing every child to whichever task started last.
+// Threading the child's identity through the loop's callback is what would fix
+// it properly. Previously this read: MaxWorkers is validated to be 1,
 // so exactly one task is in flight at any moment and the consumer can attribute
-// events to the task it last saw dispatched. Stage 2d must revisit this the
+// events to the task it last saw dispatched — which is no longer true.
 // moment two tasks can run at once — see the note on the TUI plan recorder.
 func (tool *OrchestrateTool) runnerForCall(options tools.RunOptions) PlanRunner {
 	run := tool.RunTask

internal/specialist/plan_worktree.go

@@ -17,7 +17,7 @@ import (
 // remembering to set a flag.
 //
 // PER PLAN, NOT PER TASK, and the choice matters. Per-task isolation is more
-// isolation and less use: this phase executes sequentially and a later task
+// isolation and less use: a later task
 // routinely needs an earlier one's output, so per-task worktrees would either
 // hide that or need a merge step nobody wrote. A plan is the unit of work a
 // user reviews, so a plan is the unit that gets a tree — one branch, one diff.

internal/tui/model.go

@@ -130,9 +130,21 @@ type model struct {
 	// planRunningCardKey is the card key of the plan task currently in flight.
 	// A plan's child progress events arrive keyed by the ORCHESTRATE tool call
 	// (the loop's callback carries only the parent's tool-call id), so they are
-	// attributed to this. Sound only because MaxWorkers is validated to be 1 —
-	// exactly one task runs at a time. Stage 2d must revisit it.
+	// attributed to this.
+	//
+	// EMPTY WHEN THE PLAN IS CONCURRENT. With one worker exactly one task runs
+	// at a time and the attribution is sound. With more, the events carry no
+	// task identity at all, so attributing them to whichever task was dispatched
+	// last would show one task's tool calls under another's name. No attribution
+	// is worse than none only if you have never seen the wrong one: a card that
+	// says nothing is honest, a card that says something false is not. Per-task
+	// progress needs the child's identity threaded through the loop's callback,
+	// which is its own change.
 	planRunningCardKey string
+	// planConcurrent records that the running plan has more than one worker, so
+	// the attribution above stays off for its whole life rather than being
+	// re-enabled by the next task that starts.
+	planConcurrent bool
 	// planProgress is the shared recorder the orchestrate tool holds. A POINTER
 	// for the same reason PostureGate is one: the TUI model is a value type
 	// copied on every update, so a closure over it would freeze the first run.
@@ -2682,6 +2694,9 @@ func (m model) updateModel(msg tea.Msg) (tea.Model, tea.Cmd) {
 			return m, nil
 		}
 		m.orchestrate.admit(msg, m.now())
+		// More than one worker means child progress cannot be attributed to a
+		// task; see planRunningCardKey.
+		m.planConcurrent = msg.workers > 1
 		m.transcript = appendTranscriptRow(m.transcript, transcriptRow{
 			kind:  rowSystem,
 			runID: msg.runID,
@@ -2701,9 +2716,14 @@ func (m model) updateModel(msg tea.Msg) (tea.Model, tea.Cmd) {
 		m.specialists.start(msg.taskID, msg.summary, msg.cardKey, m.now())
 		// Remember which task is in flight so the plan's progress events — which
 		// arrive keyed by the ORCHESTRATE tool call, not by task — can be
-		// attributed. Sound only because MaxWorkers is validated to be 1, so
-		// exactly one task runs at a time. Stage 2d must revisit this.
-		m.planRunningCardKey = msg.cardKey
+		// attributed. Sound ONLY when one task runs at a time; a concurrent plan
+		// leaves this empty rather than pointing every child's output at
+		// whichever task started last.
+		if m.planConcurrent {
+			m.planRunningCardKey = ""
+		} else {
+			m.planRunningCardKey = msg.cardKey
+		}
 		return m, nil
 	case planTaskDoneMsg:
 		// A BACKGROUND plan outlives the run that launched it, so the

internal/tui/plan_messages.go

@@ -19,9 +19,12 @@ import (
 // the order — so sending it here means the panel is complete before the first
 // task starts rather than assembling itself as tasks finish.
 type planAdmittedMsg struct {
-	runID      int
-	name       string
-	taskCount  int
+	runID     int
+	name      string
+	taskCount int
+	// workers is how many tasks this plan may run at once. More than one means
+	// child progress CANNOT be attributed to a task — see planRunningCardKey.
+	workers    int
 	tasks      []planGraphTask
 	tokenLimit int
 	// background marks a plan that outlives the run that launched it, so the

internal/tui/plan_progress.go

@@ -338,6 +338,7 @@ func (bridge *PlanProgressBridge) PlanAdmitted(plan specialist.Plan) {
 	name := plan.Name()
 	count := plan.TaskCount()
 	limit := plan.Budget().MaxTokens
+	workers := plan.Budget().MaxWorkers
 
 	// Copied into the message in EXECUTION ORDER, with the dependency edges, so
 	// the panel can draw the graph without reaching back into the plan — which
@@ -358,7 +359,8 @@ func (bridge *PlanProgressBridge) PlanAdmitted(plan specialist.Plan) {
 	}
 
 	bridge.send(func(runID int) tea.Msg {
-		return planAdmittedMsg{runID: runID, name: name, taskCount: count, tasks: graph, tokenLimit: limit, background: bridge.isBackground()}
+		return planAdmittedMsg{runID: runID, name: name, taskCount: count, tasks: graph, tokenLimit: limit,
+			workers: workers, background: bridge.isBackground()}
 	})
 }