vv_agent/runtime/sub_task_manager/
record.rs

1use std::collections::BTreeMap;
2use std::sync::Arc;
3use std::thread::JoinHandle;
4
5use serde_json::{json, Value};
6
7use crate::runtime::sub_agent_sessions::SubAgentSession;
8use crate::types::SubTaskOutcome;
9use crate::workspace::WorkspaceBackend;
10
11use super::helpers::{is_internal_workspace_file, preview_text, status_label};
12use super::types::ManagedSubTaskSnapshot;
13
14pub struct ManagedSubTask {
15    pub(super) task_id: String,
16    pub(super) session_id: String,
17    pub(super) agent_name: String,
18    pub(super) task_title: String,
19    pub(super) workspace_backend: Option<Arc<dyn WorkspaceBackend>>,
20    pub(super) session: Option<Arc<dyn SubAgentSession>>,
21    pub(super) outcome: Option<SubTaskOutcome>,
22    pub(super) resolved: BTreeMap<String, String>,
23    pub(super) current_cycle_index: Option<u32>,
24    pub(super) recent_activity: Option<String>,
25    pub(super) latest_cycle: Option<Value>,
26    pub(super) latest_tool_call: Option<Value>,
27    pub(super) handle: Option<JoinHandle<()>>,
28    pub(super) updated_at: String,
29    pub(super) manager_listener_attached: bool,
30}
31
32impl ManagedSubTask {
33    pub(super) fn is_running(&self) -> bool {
34        self.handle
35            .as_ref()
36            .is_some_and(|handle| !handle.is_finished())
37    }
38
39    pub(super) fn to_status_entry(&self, detail_level: &str, workspace_file_limit: usize) -> Value {
40        let status = self.status_label();
41        let mut entry = json!({
42            "task_id": self.task_id,
43            "session_id": self.session_id,
44            "agent_name": self.agent_name,
45            "status": status,
46            "task_description": self.task_title,
47        });
48        if let Some(outcome) = &self.outcome {
49            if let Some(final_answer) = &outcome.final_answer {
50                entry["final_answer"] = Value::String(final_answer.clone());
51            }
52            if let Some(wait_reason) = &outcome.wait_reason {
53                entry["wait_reason"] = Value::String(wait_reason.clone());
54            }
55            if let Some(error) = &outcome.error {
56                entry["error"] = Value::String(error.clone());
57            }
58            if outcome.cycles > 0 {
59                entry["cycles"] = Value::Number(outcome.cycles.into());
60            }
61            if !outcome.todo_list.is_empty() {
62                entry["todo_list"] = Value::Array(outcome.todo_list.clone());
63            }
64            if !outcome.resolved.is_empty() {
65                entry["resolved"] = json!(outcome.resolved);
66            }
67        }
68        if detail_level == "snapshot" {
69            let recent_activity = self
70                .recent_activity
71                .clone()
72                .or_else(|| {
73                    self.outcome.as_ref().and_then(|outcome| {
74                        outcome
75                            .final_answer
76                            .clone()
77                            .or_else(|| outcome.wait_reason.clone())
78                            .or_else(|| outcome.error.clone())
79                    })
80                })
81                .unwrap_or_else(|| self.task_title.clone());
82            let workspace_snapshot = self.workspace_snapshot(workspace_file_limit);
83            entry["snapshot"] = json!({
84                "current_cycle_index": self.current_cycle_index,
85                "task_title": self.task_title,
86                "recent_activity": recent_activity,
87                "updated_at": self.updated_at,
88                "workspace_files": workspace_snapshot.files,
89                "workspace_file_count": workspace_snapshot.file_count,
90                "workspace_files_truncated": workspace_snapshot.truncated,
91            });
92            if let Some(latest_cycle) = &self.latest_cycle {
93                entry["snapshot"]["latest_cycle"] = latest_cycle.clone();
94            }
95            if let Some(latest_tool_call) = &self.latest_tool_call {
96                entry["snapshot"]["latest_tool_call"] = latest_tool_call.clone();
97            }
98        }
99        entry
100    }
101
102    pub(super) fn snapshot(&self) -> ManagedSubTaskSnapshot {
103        ManagedSubTaskSnapshot {
104            task_id: self.task_id.clone(),
105            session_id: self.session_id.clone(),
106            agent_name: self.agent_name.clone(),
107            task_title: self.task_title.clone(),
108            status: self.status_label().to_string(),
109            running: self.is_running(),
110            outcome: self.outcome.clone(),
111            resolved: self.resolved.clone(),
112            current_cycle_index: self.current_cycle_index,
113            recent_activity: self.recent_activity.clone(),
114            latest_cycle: self.latest_cycle.clone(),
115            latest_tool_call: self.latest_tool_call.clone(),
116            updated_at: self.updated_at.clone(),
117        }
118    }
119
120    pub(super) fn status_label(&self) -> &'static str {
121        self.outcome
122            .as_ref()
123            .map(|outcome| status_label(outcome.status))
124            .unwrap_or_else(|| {
125                if self.is_running() {
126                    "running"
127                } else {
128                    "pending"
129                }
130            })
131    }
132
133    pub(super) fn update_from_outcome(&mut self, outcome: &SubTaskOutcome) {
134        if outcome.cycles > 0 {
135            self.current_cycle_index = Some(outcome.cycles);
136        }
137        self.set_latest_cycle_status(status_label(outcome.status));
138        if let Some(detail) = outcome
139            .final_answer
140            .as_ref()
141            .or(outcome.wait_reason.as_ref())
142            .or(outcome.error.as_ref())
143            .and_then(|value| preview_text(Some(&Value::String(value.clone()))))
144        {
145            self.recent_activity = Some(detail);
146        }
147    }
148
149    pub(super) fn set_latest_cycle_status(&mut self, status: &str) {
150        let mut latest_cycle = self
151            .latest_cycle
152            .as_ref()
153            .and_then(Value::as_object)
154            .cloned()
155            .unwrap_or_default();
156        latest_cycle.insert("status".to_string(), Value::String(status.to_string()));
157        if let Some(current_cycle_index) = self.current_cycle_index {
158            latest_cycle
159                .entry("cycle_index".to_string())
160                .or_insert_with(|| Value::from(current_cycle_index));
161        }
162        self.latest_cycle = Some(Value::Object(latest_cycle));
163    }
164
165    pub(super) fn mark_terminal_state(&mut self, status: &str, detail: Option<&str>) {
166        self.set_latest_cycle_status(status);
167        if let Some(detail) = detail {
168            self.recent_activity = Some(detail.to_string());
169        }
170    }
171
172    fn workspace_snapshot(&self, workspace_file_limit: usize) -> WorkspaceSnapshot {
173        let Some(workspace_backend) = &self.workspace_backend else {
174            return WorkspaceSnapshot::default();
175        };
176        let Ok(raw_files) = workspace_backend.list_files(".", "**/*") else {
177            return WorkspaceSnapshot::default();
178        };
179        let visible_files = raw_files
180            .into_iter()
181            .filter(|path| !is_internal_workspace_file(path))
182            .collect::<Vec<_>>();
183        let file_count = visible_files.len();
184        WorkspaceSnapshot {
185            files: visible_files
186                .into_iter()
187                .take(workspace_file_limit)
188                .collect::<Vec<_>>(),
189            file_count,
190            truncated: file_count > workspace_file_limit,
191        }
192    }
193}
194
195#[derive(Default)]
196struct WorkspaceSnapshot {
197    files: Vec<String>,
198    file_count: usize,
199    truncated: bool,
200}
vv_agent/runtime/sub_task_manager/record.rs

vv_agent/runtime/sub_task_manager/
record.rs
