| 1 | package agent |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "context" |
| 6 | "encoding/json" |
| 7 | "errors" |
| 8 | "fmt" |
| 9 | "slices" |
| 10 | "strings" |
| 11 | "sync" |
| 12 | |
| 13 | "reasonix/internal/event" |
| 14 | "reasonix/internal/tool" |
| 15 | ) |
| 16 | |
| 17 | // ParallelTasksTool dispatches multiple read-only sub-agent tasks concurrently |
| 18 | // and collects all results. Each sub-task runs as a foreground sub-agent in its |
| 19 | // own goroutine, emitting nested events so the frontend renders independent |
| 20 | // cards for each sub-task. |
| 21 | type ParallelTasksTool struct { |
| 22 | taskTool *TaskTool |
| 23 | } |
| 24 | |
| 25 | // NewParallelTasksTool creates a parallel dispatch tool that reuses the given |
| 26 | // TaskTool's sub-agent infrastructure. |
| 27 | func NewParallelTasksTool(taskTool *TaskTool, reg *tool.Registry) *ParallelTasksTool { |
| 28 | _ = reg // retained for source compatibility with existing constructors |
| 29 | return &ParallelTasksTool{taskTool: taskTool} |
| 30 | } |
| 31 | |
| 32 | func (p *ParallelTasksTool) Name() string { return tool.HostParallelTasks } |
| 33 | |
| 34 | func (p *ParallelTasksTool) Description() string { |
| 35 | return "Dispatch multiple read-only sub-agent tasks concurrently. Blocks until all complete, then returns a bounded preview and a stable Subagent reference for every completed persisted child; use read_subagent_result to page through any full answer without combined-result truncation." |
| 36 | } |
| 37 | |
| 38 | func (p *ParallelTasksTool) Schema() json.RawMessage { |
| 39 | return json.RawMessage(`{ |
| 40 | "type":"object", |
| 41 | "properties":{ |
| 42 | "tasks":{ |
| 43 | "type":"array", |
| 44 | "description":"Array of sub-task descriptions to run in parallel.", |
| 45 | "items":{ |
| 46 | "type":"object", |
| 47 | "properties":{ |
| 48 | "prompt":{"type":"string","description":"The task prompt for the sub-agent."}, |
| 49 | "description":{"type":"string","description":"Optional short label shown in the job list."}, |
| 50 | "tools":{"type":"array","items":{"type":"string"},"description":"Optional tool whitelist for the sub-agent."}, |
| 51 | "max_steps":{"type":"integer","description":"Optional max tool-call rounds. Defaults to half the parent agent's step budget (minimum 5), same as task.","minimum":1}, |
| 52 | "model":{"type":"string","description":"Optional model override."}, |
| 53 | "effort":{"type":"string","description":"Optional reasoning effort override."} |
| 54 | }, |
| 55 | "required":["prompt"] |
| 56 | } |
| 57 | } |
| 58 | }, |
| 59 | "required":["tasks"] |
| 60 | }`) |
| 61 | } |
| 62 | |
| 63 | func (p *ParallelTasksTool) ReadOnly() bool { return true } |
| 64 | |
| 65 | func (p *ParallelTasksTool) PlanModeSafe() bool { return true } |
| 66 | |
| 67 | type parallelTaskItem struct { |
| 68 | Prompt string `json:"prompt"` |
| 69 | Description string `json:"description"` |
| 70 | Tools []string `json:"tools"` |
| 71 | MaxSteps int `json:"max_steps"` |
| 72 | Model string `json:"model"` |
| 73 | Effort string `json:"effort"` |
| 74 | } |
| 75 | |
| 76 | type parallelTaskStatus string |
| 77 | |
| 78 | // parallelTasksMaxTasks bounds the request before any task-sized slices, |
| 79 | // channels, or goroutines are allocated. The scheduler limits how many |
| 80 | // children run simultaneously, but without an input cap a single model call |
| 81 | // could still reserve unbounded memory and queue unbounded API work (#6933). |
| 82 | const parallelTasksMaxTasks = 64 |
| 83 | |
| 84 | const ( |
| 85 | parallelTaskPending parallelTaskStatus = "pending" |
| 86 | parallelTaskCompleted parallelTaskStatus = "completed" |
| 87 | parallelTaskFailed parallelTaskStatus = "failed" |
| 88 | parallelTaskCancelled parallelTaskStatus = "cancelled" |
| 89 | parallelTaskSkipped parallelTaskStatus = "skipped" |
| 90 | ) |
| 91 | |
| 92 | func (p *ParallelTasksTool) Execute(ctx context.Context, args json.RawMessage) (result string, err error) { |
| 93 | // Group lifecycle: the group card's terminal is an explicit event from |
| 94 | // the tool itself (running once children start, exactly one terminal at |
| 95 | // the end) so frontends never infer group completion from the children |
| 96 | // they happen to have observed — children dispatch asynchronously, and a |
| 97 | // fast first child can finish before later children even appear. Every |
| 98 | // exit path (including validation failures) emits a terminal. |
| 99 | parentID, sink, _, ok := CallContext(ctx) |
| 100 | if !ok || sink == nil { |
| 101 | parentID = "parallel_tasks" |
| 102 | sink = event.Discard |
| 103 | } |
| 104 | merger := newSubagentProgressMerger(realProgressClock{}, sink, parentID) |
| 105 | defer merger.Close() |
| 106 | var statuses []parallelTaskStatus |
| 107 | defer func() { |
| 108 | merger.directStatus(parentID, parallelGroupTerminalPhase(ctx, err, statuses)) |
| 109 | }() |
| 110 | ctx = withSubagentProgressMerger(ctx, merger) |
| 111 | |
| 112 | var params struct { |
| 113 | Tasks []parallelTaskItem `json:"tasks"` |
| 114 | } |
| 115 | dec := json.NewDecoder(bytes.NewReader(args)) |
| 116 | dec.DisallowUnknownFields() |
| 117 | if err := dec.Decode(¶ms); err != nil { |
| 118 | return "", fmt.Errorf("invalid args: %w", err) |
| 119 | } |
| 120 | if len(params.Tasks) == 0 { |
| 121 | return "", fmt.Errorf("at least one task is required") |
| 122 | } |
| 123 | if len(params.Tasks) == 1 { |
| 124 | return "", fmt.Errorf("parallel_tasks with a single task is equivalent to task; use task instead") |
| 125 | } |
| 126 | if len(params.Tasks) > parallelTasksMaxTasks { |
| 127 | return "", fmt.Errorf("parallel_tasks accepts at most %d tasks; got %d", parallelTasksMaxTasks, len(params.Tasks)) |
| 128 | } |
| 129 | if err := validateParallelTaskItems(params.Tasks); err != nil { |
| 130 | return "", err |
| 131 | } |
| 132 | if p.taskTool == nil { |
| 133 | return "", fmt.Errorf("parallel_tasks is not configured") |
| 134 | } |
| 135 | |
| 136 | // The group starts running once children begin dispatching. |
| 137 | merger.directStatus(parentID, subagentPhaseRunning) |
| 138 | |
| 139 | type subResult struct { |
| 140 | index int |
| 141 | output string |
| 142 | ref string |
| 143 | err error |
| 144 | } |
| 145 | |
| 146 | n := len(params.Tasks) |
| 147 | |
| 148 | running := make([]bool, n) |
| 149 | done := make([]bool, n) |
| 150 | outputs := make([]string, n) |
| 151 | refs := make([]string, n) |
| 152 | taskErrs := make([]error, n) |
| 153 | statuses = make([]parallelTaskStatus, n) |
| 154 | for i := range params.Tasks { |
| 155 | statuses[i] = parallelTaskPending |
| 156 | } |
| 157 | |
| 158 | doneCh := make(chan subResult, n) |
| 159 | var wg sync.WaitGroup |
| 160 | |
| 161 | makeLabel := func(t parallelTaskItem, idx int) string { |
| 162 | if t.Description != "" { |
| 163 | return t.Description |
| 164 | } |
| 165 | return fmt.Sprintf("task-%d", idx+1) |
| 166 | } |
| 167 | startTask := func(idx int) { |
| 168 | t := params.Tasks[idx] |
| 169 | running[idx] = true |
| 170 | label := makeLabel(t, idx) |
| 171 | subID := fmt.Sprintf("%s/sub-%d", parentID, idx+1) |
| 172 | dispatchArgs, _ := json.Marshal(map[string]string{"prompt": t.Prompt, "description": label}) |
| 173 | sink.Emit(event.Event{ |
| 174 | Kind: event.ToolDispatch, |
| 175 | Tool: event.Tool{ |
| 176 | ID: subID, ParentID: parentID, Name: "task", |
| 177 | Args: string(dispatchArgs), ReadOnly: true, |
| 178 | }, |
| 179 | }) |
| 180 | |
| 181 | wg.Go(func() { |
| 182 | modelRef, effortRef := p.taskTool.effectiveProfile(t.Model, t.Effort) |
| 183 | itemCtx := withCallContext(ctx, subID, subSinkFor(subID, sink), nil, PlanModeFromContext(ctx)) |
| 184 | // Route through TaskTool's unified runner so persisted parent sessions |
| 185 | // retain one independently readable transcript per child. Headless runs |
| 186 | // remain ephemeral and still receive fair bounded previews. |
| 187 | output, runErr := p.taskTool.RunProfileSpec(itemCtx, ProfileExecSpec{ |
| 188 | Task: TaskSpec{Objective: t.Prompt, Description: label}, |
| 189 | Worker: WorkerSpec{Kind: "task", Name: "task", SystemPrompt: DefaultReadOnlyTaskSystemPrompt, Model: modelRef, Effort: effortRef}, |
| 190 | Grant: CapabilityGrant{ReadOnly: true, AllowNoTools: true, CallTools: t.Tools}, |
| 191 | Sched: SchedulerPolicy{MaxSteps: t.MaxSteps, Nested: SubagentDepth(ctx) > 0}, |
| 192 | }) |
| 193 | |
| 194 | if ctx.Err() != nil && runErr == nil { |
| 195 | runErr = ctx.Err() |
| 196 | } |
| 197 | if runErr != nil { |
| 198 | errText := runErr.Error() |
| 199 | if errors.Is(runErr, context.Canceled) || errors.Is(runErr, context.DeadlineExceeded) { |
| 200 | errText = "cancelled: " + errText |
| 201 | } |
| 202 | sink.Emit(event.Event{ |
| 203 | Kind: event.ToolResult, |
| 204 | Tool: event.Tool{ID: subID, ParentID: parentID, Name: "task", Err: errText}, |
| 205 | }) |
| 206 | doneCh <- subResult{index: idx, err: runErr} |
| 207 | return |
| 208 | } |
| 209 | sink.Emit(event.Event{ |
| 210 | Kind: event.ToolResult, |
| 211 | Tool: event.Tool{ID: subID, ParentID: parentID, Name: "task", Output: output}, |
| 212 | }) |
| 213 | answer, ref := splitSubagentRunResult(output) |
| 214 | doneCh <- subResult{index: idx, output: answer, ref: ref} |
| 215 | }) |
| 216 | } |
| 217 | |
| 218 | markCancelled := func(err error) { |
| 219 | for i := range params.Tasks { |
| 220 | if done[i] { |
| 221 | continue |
| 222 | } |
| 223 | done[i] = true |
| 224 | if running[i] { |
| 225 | statuses[i] = parallelTaskCancelled |
| 226 | taskErrs[i] = err |
| 227 | continue |
| 228 | } |
| 229 | statuses[i] = parallelTaskSkipped |
| 230 | taskErrs[i] = err |
| 231 | } |
| 232 | } |
| 233 | |
| 234 | completed := 0 |
| 235 | for i := range params.Tasks { |
| 236 | startTask(i) |
| 237 | } |
| 238 | processResult := func(r subResult) { |
| 239 | if done[r.index] { |
| 240 | return |
| 241 | } |
| 242 | completed++ |
| 243 | done[r.index] = true |
| 244 | outputs[r.index] = r.output |
| 245 | refs[r.index] = r.ref |
| 246 | taskErrs[r.index] = r.err |
| 247 | switch { |
| 248 | case r.err == nil: |
| 249 | statuses[r.index] = parallelTaskCompleted |
| 250 | case errors.Is(r.err, context.Canceled), errors.Is(r.err, context.DeadlineExceeded): |
| 251 | statuses[r.index] = parallelTaskCancelled |
| 252 | default: |
| 253 | statuses[r.index] = parallelTaskFailed |
| 254 | } |
| 255 | } |
| 256 | for completed < n { |
| 257 | select { |
| 258 | case r := <-doneCh: |
| 259 | processResult(r) |
| 260 | case <-ctx.Done(): |
| 261 | err := ctx.Err() |
| 262 | drain: |
| 263 | for { |
| 264 | select { |
| 265 | case r := <-doneCh: |
| 266 | processResult(r) |
| 267 | default: |
| 268 | break drain |
| 269 | } |
| 270 | } |
| 271 | markCancelled(err) |
| 272 | wg.Wait() |
| 273 | return formatParallelTasksAggregate(outputs, refs, taskErrs, statuses, true), err |
| 274 | } |
| 275 | } |
| 276 | wg.Wait() |
| 277 | if parallelTasksWereCancelled(statuses) { |
| 278 | err := ctx.Err() |
| 279 | if err == nil { |
| 280 | err = context.Canceled |
| 281 | } |
| 282 | return formatParallelTasksAggregate(outputs, refs, taskErrs, statuses, true), err |
| 283 | } |
| 284 | return formatParallelTasksAggregate(outputs, refs, taskErrs, statuses, false), nil |
| 285 | } |
| 286 | |
| 287 | // parallelGroupTerminalPhase classifies a parallel_tasks group's single |
| 288 | // terminal status: cancellation/deadline wins, then any failed child, then |
| 289 | // any error (including validation failures), then completed. |
| 290 | func parallelGroupTerminalPhase(ctx context.Context, err error, statuses []parallelTaskStatus) subagentProgressPhase { |
| 291 | if ctx.Err() != nil { |
| 292 | return subagentPhaseCancelled |
| 293 | } |
| 294 | if slices.Contains(statuses, parallelTaskFailed) { |
| 295 | return subagentPhaseFailed |
| 296 | } |
| 297 | if err != nil { |
| 298 | return subagentPhaseFailed |
| 299 | } |
| 300 | return subagentPhaseCompleted |
| 301 | } |
| 302 | |
| 303 | func parallelTasksWereCancelled(statuses []parallelTaskStatus) bool { |
| 304 | for _, st := range statuses { |
| 305 | if st == parallelTaskCancelled || st == parallelTaskSkipped { |
| 306 | return true |
| 307 | } |
| 308 | } |
| 309 | return false |
| 310 | } |
| 311 | |
| 312 | func formatParallelTasksAggregate(outputs, refs []string, errs []error, statuses []parallelTaskStatus, cancelled bool) string { |
| 313 | n := len(statuses) |
| 314 | var prefix string |
| 315 | if cancelled { |
| 316 | completed := 0 |
| 317 | for _, st := range statuses { |
| 318 | if st == parallelTaskCompleted { |
| 319 | completed++ |
| 320 | } |
| 321 | } |
| 322 | prefix = fmt.Sprintf("Cancelled parallel tasks after completing %d of %d tasks:\n", completed, n) |
| 323 | } else { |
| 324 | prefix = fmt.Sprintf("Completed %d parallel tasks:\n", n) |
| 325 | } |
| 326 | items := make([]subagentAggregateItem, 0, n) |
| 327 | for i, st := range statuses { |
| 328 | item := subagentAggregateItem{header: fmt.Sprintf("── task-%d ──\n", i+1)} |
| 329 | switch st { |
| 330 | case parallelTaskCompleted: |
| 331 | item.status = "status: completed\n" |
| 332 | item.answer = strings.TrimSpace(outputs[i]) |
| 333 | if i < len(refs) { |
| 334 | item.ref = refs[i] |
| 335 | } |
| 336 | case parallelTaskCancelled: |
| 337 | item.status = "status: cancelled\n" |
| 338 | if errs[i] != nil { |
| 339 | item.detail = fmt.Sprintf("[CANCELLED] %s\n", boundedInline(errs[i].Error(), 256)) |
| 340 | } else { |
| 341 | item.detail = "[CANCELLED]\n" |
| 342 | } |
| 343 | case parallelTaskSkipped: |
| 344 | item.status = "status: skipped\n" |
| 345 | if errs[i] != nil { |
| 346 | item.detail = fmt.Sprintf("[SKIPPED] cancelled before start: %s\n", boundedInline(errs[i].Error(), 256)) |
| 347 | } else { |
| 348 | item.detail = "[SKIPPED] cancelled before start\n" |
| 349 | } |
| 350 | case parallelTaskFailed: |
| 351 | item.status = "status: failed\n" |
| 352 | if errs[i] != nil { |
| 353 | item.detail = fmt.Sprintf("[FAILED] %s\n", boundedInline(errs[i].Error(), 256)) |
| 354 | } else { |
| 355 | item.detail = "[FAILED]\n" |
| 356 | } |
| 357 | default: |
| 358 | item.status = "status: pending\n" |
| 359 | } |
| 360 | items = append(items, item) |
| 361 | } |
| 362 | return formatBoundedSubagentAggregate(prefix, items) |
| 363 | } |
| 364 | |
| 365 | func validateParallelTaskItems(tasks []parallelTaskItem) error { |
| 366 | for i, t := range tasks { |
| 367 | if strings.TrimSpace(t.Prompt) == "" { |
| 368 | return fmt.Errorf("task %d: prompt is required", i+1) |
| 369 | } |
| 370 | } |
| 371 | return nil |
| 372 | } |
| 373 |