vv_agent/runtime/sub_task_manager/
status.rs

1use std::thread::{self, JoinHandle};
2use std::time::{Duration, Instant};
3
4use serde_json::{json, Value};
5
6use super::manager::SubTaskManager;
7use super::record::ManagedSubTask;
8use super::types::ManagedSubTaskSnapshot;
9
10impl SubTaskManager {
11    pub fn status_entries(
12        &self,
13        task_ids: &[String],
14        detail_level: &str,
15        workspace_file_limit: usize,
16    ) -> Vec<Value> {
17        let mut tasks = self.tasks.lock().expect("sub-task manager poisoned");
18        for record in tasks.values_mut() {
19            if record.handle.as_ref().is_some_and(JoinHandle::is_finished) {
20                if let Some(handle) = record.handle.take() {
21                    let _ = handle.join();
22                }
23            }
24        }
25        task_ids
26            .iter()
27            .map(|task_id| {
28                let Some(record) = tasks.get(task_id) else {
29                    return json!({
30                        "task_id": task_id,
31                        "status": "missing",
32                        "error": format!("Sub-task {task_id} not found."),
33                    });
34                };
35                record.to_status_entry(detail_level, workspace_file_limit)
36            })
37            .collect()
38    }
39
40    pub fn get(&self, task_id: &str) -> Option<ManagedSubTaskSnapshot> {
41        self.join_finished_tasks();
42        self.tasks
43            .lock()
44            .expect("sub-task manager poisoned")
45            .get(task_id)
46            .map(ManagedSubTask::snapshot)
47    }
48
49    pub fn task_session_id(&self, task_id: &str) -> Option<String> {
50        self.join_finished_tasks();
51        self.tasks
52            .lock()
53            .expect("sub-task manager poisoned")
54            .get(task_id)
55            .map(|record| record.session_id.clone())
56    }
57
58    pub fn task_status_label(&self, task_id: &str) -> Option<String> {
59        self.join_finished_tasks();
60        self.tasks
61            .lock()
62            .expect("sub-task manager poisoned")
63            .get(task_id)
64            .map(|record| record.status_label().to_string())
65    }
66
67    pub fn is_running(&self, task_id: &str) -> bool {
68        self.join_finished_tasks();
69        self.tasks
70            .lock()
71            .expect("sub-task manager poisoned")
72            .get(task_id)
73            .is_some_and(ManagedSubTask::is_running)
74    }
75
76    pub fn wait(&self, task_id: &str, timeout: Option<Duration>) -> bool {
77        let deadline = timeout.map(|duration| Instant::now() + duration);
78        loop {
79            let handle = {
80                let mut tasks = self.tasks.lock().expect("sub-task manager poisoned");
81                let Some(record) = tasks.get_mut(task_id) else {
82                    return false;
83                };
84                match record.handle.as_ref() {
85                    Some(handle) if timeout.is_none() || handle.is_finished() => {
86                        record.handle.take()
87                    }
88                    Some(_) => None,
89                    None => return true,
90                }
91            };
92
93            if let Some(handle) = handle {
94                let _ = handle.join();
95                return true;
96            }
97
98            if deadline.is_some_and(|deadline| Instant::now() >= deadline) {
99                return false;
100            }
101            thread::sleep(Duration::from_millis(10));
102        }
103    }
104
105    pub fn wait_for_record(
106        &self,
107        task_id: &str,
108        timeout: Option<Duration>,
109    ) -> Option<ManagedSubTaskSnapshot> {
110        if !self.wait(task_id, timeout) {
111            return None;
112        }
113        self.get(task_id)
114    }
115
116    fn join_finished_tasks(&self) {
117        let mut tasks = self.tasks.lock().expect("sub-task manager poisoned");
118        for record in tasks.values_mut() {
119            if record.handle.as_ref().is_some_and(JoinHandle::is_finished) {
120                if let Some(handle) = record.handle.take() {
121                    let _ = handle.join();
122                }
123            }
124        }
125    }
126}
vv_agent/runtime/sub_task_manager/status.rs

vv_agent/runtime/sub_task_manager/
status.rs
