| 467 | } |
| 468 | |
| 469 | func (rt *AgentTaskRuntime) WaitForTasks(ctx context.Context, taskIDs []string, scopeID string, timeoutSeconds int) map[string]any { |
| 470 | if len(taskIDs) == 0 { |
| 471 | return map[string]any{"tasks": []map[string]any{}, "timeout": false, "pending_task_ids": []string{}} |
| 472 | } |
| 473 | softDeadline := time.Now().Add(time.Duration(timeoutSeconds) * time.Second) |
| 474 | hardDeadline := time.Now().Add(HardMaxWaitSeconds * time.Second) |
| 475 | for { |
| 476 | rt.consumeMatchingNotifications(scopeID, taskIDs) |
| 477 | snapshot := rt.snapshot(taskIDs, scopeID) |
| 478 | inaccessible := rt.findInaccessibleTaskIDs(taskIDs, scopeID) |
| 479 | expired := rt.findExpiredTaskIDs(taskIDs, scopeID) |
| 480 | if len(inaccessible) > 0 { |
| 481 | snapshot["error"] = "Error: Unknown or inaccessible task ids: " + strings.Join(inaccessible, ", ") + "." |
| 482 | snapshot["inaccessible_task_ids"] = inaccessible |
| 483 | snapshot["expired_task_ids"] = expired |
| 484 | return snapshot |
| 485 | } |
| 486 | if len(expired) > 0 { |
| 487 | snapshot["error"] = "Error: Task records expired or were already cleaned up: " + strings.Join(expired, ", ") + "." |
| 488 | snapshot["inaccessible_task_ids"] = []string{} |
| 489 | snapshot["expired_task_ids"] = expired |
| 490 | return snapshot |
| 491 | } |
| 492 | missing := stringSliceFromAny(snapshot["missing_task_ids"]) |
| 493 | if len(missing) > 0 { |
| 494 | snapshot["error"] = "Error: Unknown or inaccessible task ids: " + strings.Join(missing, ", ") + "." |
| 495 | snapshot["inaccessible_task_ids"] = missing |
| 496 | snapshot["expired_task_ids"] = []string{} |
| 497 | return snapshot |
| 498 | } |
| 499 | allStable := len(snapshot["pending_task_ids"].([]string)) == 0 |
| 500 | now := time.Now() |
| 501 | if allStable || !now.Before(softDeadline) || !now.Before(hardDeadline) { |
| 502 | if !now.Before(hardDeadline) { |
| 503 | snapshot["error"] = fmt.Sprintf("Error: TaskWait timed out waiting for workers after %d seconds.", HardMaxWaitSeconds) |
| 504 | } |
| 505 | snapshot["timeout"] = !now.Before(softDeadline) && !allStable |
| 506 | snapshot["expired_task_ids"] = expired |
| 507 | snapshot["inaccessible_task_ids"] = []string{} |
| 508 | return snapshot |
| 509 | } |
| 510 | wait := make(chan struct{}, 1) |
| 511 | rt.mu.Lock() |
| 512 | for _, id := range taskIDs { |
| 513 | rt.waiters[id] = append(rt.waiters[id], wait) |
| 514 | } |
| 515 | rt.mu.Unlock() |
| 516 | maxWait := time.Until(hardDeadline) |
| 517 | if time.Until(softDeadline) < maxWait { |
| 518 | maxWait = time.Until(softDeadline) |
| 519 | } |
| 520 | if maxWait < 0 { |
| 521 | maxWait = 0 |
| 522 | } |
| 523 | timer := time.NewTimer(maxWait) |
| 524 | select { |
| 525 | case <-ctx.Done(): |
| 526 | timer.Stop() |