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 has_attached_session(&self, task_id: &str) -> Option<bool> {
68        self.join_finished_tasks();
69        self.tasks
70            .lock()
71            .expect("sub-task manager poisoned")
72            .get(task_id)
73            .map(|record| record.session.is_some())
74    }
75
76    pub fn is_running(&self, task_id: &str) -> bool {
77        self.join_finished_tasks();
78        self.tasks
79            .lock()
80            .expect("sub-task manager poisoned")
81            .get(task_id)
82            .is_some_and(ManagedSubTask::is_running)
83    }
84
85    pub fn wait(&self, task_id: &str, timeout: Option<Duration>) -> bool {
86        let deadline = timeout.map(|duration| Instant::now() + duration);
87        loop {
88            let handle = {
89                let mut tasks = self.tasks.lock().expect("sub-task manager poisoned");
90                let Some(record) = tasks.get_mut(task_id) else {
91                    return false;
92                };
93                match record.handle.as_ref() {
94                    Some(handle) if timeout.is_none() || handle.is_finished() => {
95                        record.handle.take()
96                    }
97                    Some(_) => None,
98                    None if record.is_running() => None,
99                    None => return true,
100                }
101            };
102
103            if let Some(handle) = handle {
104                let _ = handle.join();
105                return true;
106            }
107
108            if deadline.is_some_and(|deadline| Instant::now() >= deadline) {
109                return false;
110            }
111            thread::sleep(Duration::from_millis(10));
112        }
113    }
114
115    pub fn wait_for_record(
116        &self,
117        task_id: &str,
118        timeout: Option<Duration>,
119    ) -> Option<ManagedSubTaskSnapshot> {
120        if !self.wait(task_id, timeout) {
121            return None;
122        }
123        self.get(task_id)
124    }
125
126    fn join_finished_tasks(&self) {
127        let mut tasks = self.tasks.lock().expect("sub-task manager poisoned");
128        for record in tasks.values_mut() {
129            if record.handle.as_ref().is_some_and(JoinHandle::is_finished) {
130                if let Some(handle) = record.handle.take() {
131                    let _ = handle.join();
132                }
133            }
134        }
135    }
136}
vv_agent/runtime/sub_task_manager/status.rs

vv_agent/runtime/sub_task_manager/
status.rs
