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