vv_agent/runtime/sub_task_manager/
submission.rs

1use std::collections::BTreeMap;
2use std::panic::{catch_unwind, AssertUnwindSafe};
3use std::sync::Arc;
4use std::thread;
5
6use crate::types::{AgentStatus, SubTaskOutcome};
7use crate::workspace::WorkspaceBackend;
8
9use super::helpers::{now_iso, panic_payload_to_string};
10use super::manager::SubTaskManager;
11use super::record::ManagedSubTask;
12
13impl SubTaskManager {
14    pub fn submit(
15        &self,
16        task_id: impl Into<String>,
17        session_id: impl Into<String>,
18        agent_name: impl Into<String>,
19        task_title: impl Into<String>,
20        runner: impl FnOnce() -> SubTaskOutcome + Send + 'static,
21    ) -> Result<(), String> {
22        self.submit_with_workspace(task_id, session_id, agent_name, task_title, None, runner)
23    }
24
25    pub fn submit_with_workspace(
26        &self,
27        task_id: impl Into<String>,
28        session_id: impl Into<String>,
29        agent_name: impl Into<String>,
30        task_title: impl Into<String>,
31        workspace_backend: Option<Arc<dyn WorkspaceBackend>>,
32        runner: impl FnOnce() -> SubTaskOutcome + Send + 'static,
33    ) -> Result<(), String> {
34        let task_id = task_id.into();
35        let session_id = session_id.into();
36        let agent_name = agent_name.into();
37        let task_title = task_title.into();
38        {
39            let mut tasks = self.tasks.lock().expect("sub-task manager poisoned");
40            if tasks.get(&task_id).is_some_and(ManagedSubTask::is_running) {
41                return Err(format!("Sub-task {task_id} is already running."));
42            }
43            tasks.insert(
44                task_id.clone(),
45                ManagedSubTask {
46                    task_id: task_id.clone(),
47                    session_id: session_id.clone(),
48                    agent_name: agent_name.clone(),
49                    task_title,
50                    workspace_backend,
51                    session: None,
52                    outcome: None,
53                    resolved: BTreeMap::new(),
54                    current_cycle_index: None,
55                    recent_activity: None,
56                    latest_cycle: None,
57                    latest_tool_call: None,
58                    handle: None,
59                    updated_at: now_iso(),
60                    manager_listener_attached: false,
61                },
62            );
63        }
64
65        let tasks = self.tasks.clone();
66        let task_id_for_thread = task_id.clone();
67        let handle = thread::spawn(move || {
68            let outcome = catch_unwind(AssertUnwindSafe(runner));
69            let mut tasks = tasks.lock().expect("sub-task manager poisoned");
70            if let Some(record) = tasks.get_mut(&task_id_for_thread) {
71                let outcome = match outcome {
72                    Ok(outcome) => outcome,
73                    Err(payload) => SubTaskOutcome {
74                        task_id: record.task_id.clone(),
75                        agent_name: record.agent_name.clone(),
76                        status: AgentStatus::Failed,
77                        session_id: Some(record.session_id.clone()),
78                        final_answer: None,
79                        wait_reason: None,
80                        error: Some(panic_payload_to_string(payload.as_ref())),
81                        cycles: 0,
82                        todo_list: Vec::new(),
83                        resolved: record.resolved.clone(),
84                    },
85                };
86                if !outcome.resolved.is_empty() {
87                    record.resolved = outcome.resolved.clone();
88                }
89                record.update_from_outcome(&outcome);
90                record.outcome = Some(outcome);
91                record.updated_at = now_iso();
92                record.handle = None;
93            }
94        });
95
96        let mut tasks = self.tasks.lock().expect("sub-task manager poisoned");
97        if let Some(record) = tasks.get_mut(&task_id) {
98            record.handle = Some(handle);
99            record.updated_at = now_iso();
100        }
101        Ok(())
102    }
103
104    pub fn record_outcome(&self, task_id: &str, outcome: SubTaskOutcome) {
105        let mut tasks = self.tasks.lock().expect("sub-task manager poisoned");
106        let task_id = task_id.to_string();
107        match tasks.get_mut(&task_id) {
108            Some(record) => {
109                record.session_id = outcome
110                    .session_id
111                    .clone()
112                    .unwrap_or_else(|| record.session_id.clone());
113                record.agent_name = outcome.agent_name.clone();
114                if !outcome.resolved.is_empty() {
115                    record.resolved = outcome.resolved.clone();
116                }
117                record.update_from_outcome(&outcome);
118                record.outcome = Some(outcome);
119                record.updated_at = now_iso();
120            }
121            None => {
122                let mut record = ManagedSubTask {
123                    session_id: outcome.session_id.clone().unwrap_or_default(),
124                    agent_name: outcome.agent_name.clone(),
125                    task_title: String::new(),
126                    workspace_backend: None,
127                    session: None,
128                    outcome: None,
129                    resolved: outcome.resolved.clone(),
130                    current_cycle_index: None,
131                    recent_activity: None,
132                    latest_cycle: None,
133                    latest_tool_call: None,
134                    task_id: task_id.clone(),
135                    handle: None,
136                    updated_at: now_iso(),
137                    manager_listener_attached: false,
138                };
139                record.update_from_outcome(&outcome);
140                record.outcome = Some(outcome);
141                tasks.insert(task_id.clone(), record);
142            }
143        }
144    }
145}
vv_agent/runtime/sub_task_manager/submission.rs

vv_agent/runtime/sub_task_manager/
submission.rs
