diff --git a/internal/specialist/plan.go b/internal/specialist/plan.go index b68b9639..75e39f75 100644 --- a/internal/specialist/plan.go +++ b/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. // diff --git a/internal/specialist/plan_concurrent_test.go b/internal/specialist/plan_concurrent_test.go new file mode 100644 index 00000000..8b62c507 --- /dev/null +++ b/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) + } + } +} diff --git a/internal/specialist/plan_exec.go b/internal/specialist/plan_exec.go index c08c596a..d75c7849 100644 --- a/internal/specialist/plan_exec.go +++ b/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() { diff --git a/internal/specialist/plan_schedule.go b/internal/specialist/plan_schedule.go new file mode 100644 index 00000000..bfc411d6 --- /dev/null +++ b/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() +} diff --git a/internal/specialist/plan_test.go b/internal/specialist/plan_test.go index d1e421bd..4c6c45d4 100644 --- a/internal/specialist/plan_test.go +++ b/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) } } } diff --git a/internal/specialist/plan_tool.go b/internal/specialist/plan_tool.go index 0230cdeb..f149f6ce 100644 --- a/internal/specialist/plan_tool.go +++ b/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 diff --git a/internal/specialist/plan_worktree.go b/internal/specialist/plan_worktree.go index c20b87e9..517e43ed 100644 --- a/internal/specialist/plan_worktree.go +++ b/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. diff --git a/internal/tui/model.go b/internal/tui/model.go index 5679f314..d2f25238 100644 --- a/internal/tui/model.go +++ b/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 diff --git a/internal/tui/plan_messages.go b/internal/tui/plan_messages.go index fde97cf2..36ca9c73 100644 --- a/internal/tui/plan_messages.go +++ b/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 diff --git a/internal/tui/plan_progress.go b/internal/tui/plan_progress.go index b3709813..53c49018 100644 --- a/internal/tui/plan_progress.go +++ b/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()} }) }