Internal helper: mark a task_id as terminal in the FIFO queue and evict oldest entries past the cap. Idempotent for tasks that are already in the terminal queue (a SubagentEnd arriving after a cancel won't double-push).
(&self, task_id: &str)
| 82 | /// that are already in the terminal queue (a SubagentEnd arriving |
| 83 | /// after a cancel won't double-push). |
| 84 | async fn mark_terminal_and_evict(&self, task_id: &str) { |
| 85 | let cap = match self.max_terminal_tasks { |
| 86 | Some(n) => n, |
| 87 | None => return, |
| 88 | }; |
| 89 | // Hold all three structures together for the push + eviction so a |
| 90 | // concurrent `record_event` (which takes only `tasks`) cannot |
| 91 | // re-insert a victim into `tasks` in the window between its removal |
| 92 | // from `tasks` and `cancellers`. Canonical order: |
| 93 | // terminal_order -> tasks -> cancellers. Callers (`cancel`, |
| 94 | // `record_event`) always drop their `tasks`/`cancellers` guards |
| 95 | // before invoking this, so holding all three here cannot deadlock. |
| 96 | let mut order = self.terminal_order.write().await; |
| 97 | let mut tasks = self.tasks.write().await; |
| 98 | let mut cancellers = self.cancellers.write().await; |
| 99 | if !order.iter().any(|id| id == task_id) { |
| 100 | order.push_back(task_id.to_string()); |
| 101 | } |
| 102 | while order.len() > cap { |
| 103 | if let Some(victim) = order.pop_front() { |
| 104 | tasks.remove(&victim); |
| 105 | cancellers.remove(&victim); |
| 106 | } |
| 107 | } |
| 108 | } |
| 109 | |
| 110 | /// Register a `CancellationToken` for a running task so callers can |
| 111 | /// trigger cancellation through `cancel(task_id)`. The task executor |
no test coverage detected