1use std::collections::BTreeMap;
2use std::panic::{catch_unwind, AssertUnwindSafe};
3use std::sync::atomic::{AtomicU64, Ordering};
4use std::sync::{Arc, Mutex};
5
6use serde_json::{json, Value};
7
8use crate::agent::Agent;
9use crate::result::RunResult;
10use crate::run_config::RunConfig;
11use crate::runner::{NormalizedInput, Runner};
12use crate::runtime::CancellationToken;
13use crate::tools::{Tool, ToolContext, ToolOutput, ToolSpec, ToolSpecKind};
14use crate::types::{AgentStatus, ToolArguments};
15
16static NEXT_BACKGROUND_AGENT_TASK_ID: AtomicU64 = AtomicU64::new(1);
17const BACKGROUND_AGENT_TASK_WORKER_PANIC: &str = "background agent task worker panicked";
18
19#[derive(Clone)]
20pub struct BackgroundAgentTask {
21 agent: Agent,
22 name: String,
23 description: String,
24 parameters_schema: Value,
25 handles: Arc<Mutex<BTreeMap<String, BackgroundAgentTaskHandle>>>,
26}
27
28impl BackgroundAgentTask {
29 pub fn start(
30 &self,
31 runner: &Runner,
32 context: &ToolContext,
33 raw_arguments: Value,
34 run_config: Option<RunConfig>,
35 ) -> Result<BackgroundAgentTaskHandle, String> {
36 let input = self.input_from_arguments(raw_arguments)?;
37 let run_config = inherited_run_config(context, run_config);
38 let task_id = format!(
39 "bg_agent_{:012x}",
40 NEXT_BACKGROUND_AGENT_TASK_ID.fetch_add(1, Ordering::Relaxed)
41 );
42 let state = Arc::new(Mutex::new(BackgroundAgentTaskState {
43 status: AgentStatus::Running,
44 result: None,
45 error: None,
46 }));
47 let state_for_worker = state.clone();
48 let runner = runner.clone();
49 let agent = self.agent.clone();
50 let task_id_for_error = task_id.clone();
51 let _ = std::thread::Builder::new()
52 .name(format!("vv-agent-background-{task_id}"))
53 .spawn(move || {
54 let update = catch_unwind(AssertUnwindSafe(|| {
55 match runner.run_blocking(
56 &agent,
57 NormalizedInput::from(input),
58 run_config,
59 None,
60 ) {
61 Ok(result) => {
62 let status = result.status();
63 let error = result.result().error.clone();
64 (status, Some(result), error)
65 }
66 Err(error) => (AgentStatus::Failed, None, Some(error)),
67 }
68 }))
69 .unwrap_or_else(|_| {
70 (
71 AgentStatus::Failed,
72 None,
73 Some(BACKGROUND_AGENT_TASK_WORKER_PANIC.to_string()),
74 )
75 });
76 if let Ok(mut state) = state_for_worker.lock() {
77 (state.status, state.result, state.error) = update;
78 }
79 })
80 .map_err(|error| {
81 if let Ok(mut state) = state.lock() {
82 state.status = AgentStatus::Failed;
83 state.error = Some(error.to_string());
84 }
85 format!("failed to spawn background agent task {task_id_for_error}: {error}")
86 })?;
87 let handle = BackgroundAgentTaskHandle {
88 task_id,
89 agent_name: self.agent.name().to_string(),
90 state,
91 };
92 self.handles
93 .lock()
94 .map_err(|_| "background task registry lock poisoned".to_string())?
95 .insert(handle.task_id.clone(), handle.clone());
96 Ok(handle)
97 }
98
99 pub fn get_handle(&self, task_id: &str) -> Result<BackgroundAgentTaskHandle, String> {
100 self.handles
101 .lock()
102 .map_err(|_| "background task registry lock poisoned".to_string())?
103 .get(task_id)
104 .cloned()
105 .ok_or_else(|| format!("unknown background agent task: {task_id}"))
106 }
107
108 fn input_from_arguments(&self, raw_arguments: Value) -> Result<String, String> {
109 let object = raw_arguments
110 .as_object()
111 .ok_or_else(|| "background task arguments must be an object".to_string())?;
112 object
113 .get("task_description")
114 .or_else(|| object.get("task"))
115 .or_else(|| object.get("input"))
116 .and_then(Value::as_str)
117 .map(str::trim)
118 .filter(|value| !value.is_empty())
119 .map(str::to_string)
120 .ok_or_else(|| "background task requires task_description".to_string())
121 }
122}
123
124impl Tool for BackgroundAgentTask {
125 fn name(&self) -> &str {
126 &self.name
127 }
128
129 fn description(&self) -> &str {
130 &self.description
131 }
132
133 fn parameters_schema(&self) -> &Value {
134 &self.parameters_schema
135 }
136
137 fn as_tool_spec(&self) -> ToolSpec {
138 let name = self.name.clone();
139 let description = self.description.clone();
140 let parameters_schema = self.parameters_schema.clone();
141 let task = self.clone();
142 let mut spec = ToolSpec::new(
143 name.clone(),
144 description.clone(),
145 Arc::new(
146 move |context: &mut ToolContext, arguments: &ToolArguments| {
147 let raw_arguments = Value::Object(arguments.clone().into_iter().collect());
148 let task_description = match task.input_from_arguments(raw_arguments.clone()) {
149 Ok(task_description) => task_description,
150 Err(error) => {
151 return ToolOutput::error(error)
152 .with_code("invalid_background_task_arguments")
153 .to_result(&context.tool_call_id)
154 }
155 };
156 let Some(model_provider) = context.model_provider.clone() else {
157 return ToolOutput::error("background agent runtime is not available")
158 .with_code("background_agent_runtime_unavailable")
159 .to_result(&context.tool_call_id);
160 };
161 let runner = match Runner::builder()
162 .model_provider_arc(model_provider)
163 .workspace(context.workspace.clone())
164 .build()
165 {
166 Ok(runner) => runner,
167 Err(error) => {
168 return ToolOutput::error(error)
169 .with_code("background_agent_runtime_unavailable")
170 .to_result(&context.tool_call_id)
171 }
172 };
173 match task.start(&runner, context, raw_arguments, None) {
174 Ok(handle) => ToolOutput::json(json!({
175 "agent_name": task.agent.name(),
176 "status": "background_task_started",
177 "task_description": task_description,
178 "task_id": handle.task_id(),
179 }))
180 .to_result(&context.tool_call_id),
181 Err(error) => ToolOutput::error(error)
182 .with_code("background_agent_start_failed")
183 .to_result(&context.tool_call_id),
184 }
185 },
186 ),
187 );
188 spec.kind = ToolSpecKind::BackgroundAgent;
189 spec.schema = json!({
190 "type": "function",
191 "function": {
192 "name": name,
193 "description": description,
194 "parameters": parameters_schema,
195 }
196 });
197 spec
198 }
199}
200
201pub struct BackgroundAgentTaskBuilder {
202 agent: Agent,
203 name: Option<String>,
204 description: Option<String>,
205}
206
207impl BackgroundAgentTaskBuilder {
208 pub fn new(agent: Agent) -> Self {
209 Self {
210 agent,
211 name: None,
212 description: None,
213 }
214 }
215
216 pub fn name(mut self, name: impl Into<String>) -> Self {
217 self.name = Some(name.into());
218 self
219 }
220
221 pub fn description(mut self, description: impl Into<String>) -> Self {
222 self.description = Some(description.into());
223 self
224 }
225
226 pub fn build(self) -> Result<BackgroundAgentTask, String> {
227 let name = self
228 .name
229 .unwrap_or_else(|| format!("{}_background_task", self.agent.name()));
230 if name.trim().is_empty() {
231 return Err("background task tool name cannot be empty".to_string());
232 }
233 let description = self.description.unwrap_or_else(|| {
234 format!(
235 "Start the {} agent as a background task.",
236 self.agent.name()
237 )
238 });
239 Ok(BackgroundAgentTask {
240 agent: self.agent,
241 name,
242 description,
243 parameters_schema: json!({
244 "type": "object",
245 "properties": {
246 "task_description": {
247 "type": "string",
248 "description": "Task for the background agent."
249 }
250 },
251 "required": ["task_description"],
252 "additionalProperties": false
253 }),
254 handles: Arc::new(Mutex::new(BTreeMap::new())),
255 })
256 }
257}
258
259pub(crate) fn inherited_run_config(
260 context: &ToolContext,
261 run_config: Option<RunConfig>,
262) -> RunConfig {
263 let explicit_shared_state = run_config
264 .as_ref()
265 .map(|config| config.initial_shared_state.clone())
266 .unwrap_or_default();
267 let explicit_metadata = run_config
268 .as_ref()
269 .map(|config| config.metadata.clone())
270 .unwrap_or_default();
271 let has_projected_parent = context.background_parent_run_config.is_some();
272 let mut config = context
273 .background_parent_run_config
274 .clone()
275 .or(run_config)
276 .unwrap_or_default();
277 if config.workspace.is_none() {
278 config.workspace = Some(context.workspace.clone());
279 }
280 if config.workspace_backend.is_none() {
281 config.workspace_backend = Some(context.workspace_backend.clone());
282 }
283 if config.model_provider.is_none() {
284 config.model_provider = context.model_provider.clone();
285 }
286 if config.execution_backend.is_none() {
287 config.execution_backend = context.execution_backend.clone();
288 }
289 if config.app_state.is_none() {
290 config.app_state = context.app_state.clone();
291 }
292 if config.cancellation_token.is_none() {
293 config.cancellation_token = context
294 .sub_task_turn_snapshot
295 .as_ref()
296 .and_then(|snapshot| snapshot.cancellation_token.as_ref())
297 .cloned();
298 }
299 let scoped_child_cancellation = config
300 .cancellation_token
301 .is_none()
302 .then(CancellationToken::child_of_current)
303 .flatten();
304
305 let mut shared_state = context.shared_state.clone();
306 shared_state.extend(explicit_shared_state);
307
308 if !has_projected_parent {
309 config.metadata = context.metadata.clone();
310 for key in [
311 "agent_name",
312 "session_id",
313 "approved_tool_interruption_ids",
314 "_vv_agent_run_id",
315 "_vv_agent_trace_id",
316 "_vv_agent_agent_name",
317 "_vv_agent_input",
318 "_vv_agent_session_id",
319 "_vv_agent_tool_use_behavior",
320 "_vv_agent_stop_at_tool_names",
321 ] {
322 config.metadata.remove(key);
323 }
324 }
325 config.metadata.extend(explicit_metadata);
326 let mut child = config.for_background_child(shared_state);
327 if child.cancellation_token.is_none() {
328 child.cancellation_token = scoped_child_cancellation;
329 }
330 child
331}
332
333#[derive(Clone)]
334pub struct BackgroundAgentTaskHandle {
335 task_id: String,
336 agent_name: String,
337 state: Arc<Mutex<BackgroundAgentTaskState>>,
338}
339
340impl BackgroundAgentTaskHandle {
341 pub fn task_id(&self) -> &str {
342 &self.task_id
343 }
344
345 pub fn agent_name(&self) -> &str {
346 &self.agent_name
347 }
348
349 pub fn status(&self) -> AgentStatus {
350 self.state
351 .lock()
352 .map(|state| state.status)
353 .unwrap_or(AgentStatus::Failed)
354 }
355
356 pub fn poll(&self) -> Result<BackgroundAgentTaskSnapshot, String> {
357 let state = self
358 .state
359 .lock()
360 .map_err(|_| "background task lock poisoned".to_string())?;
361 Ok(BackgroundAgentTaskSnapshot {
362 task_id: self.task_id.clone(),
363 agent_name: self.agent_name.clone(),
364 status: state.status,
365 final_output: state
366 .result
367 .as_ref()
368 .and_then(|result| result.final_output().map(str::to_string)),
369 error: state.error.clone(),
370 })
371 }
372
373 pub async fn wait(&self) -> Result<BackgroundAgentTaskSnapshot, String> {
374 loop {
375 let snapshot = self.poll()?;
376 if !matches!(snapshot.status, AgentStatus::Running | AgentStatus::Pending) {
377 return Ok(snapshot);
378 }
379 tokio::time::sleep(std::time::Duration::from_millis(10)).await;
380 }
381 }
382
383 pub async fn wait_with_timeout(
384 &self,
385 timeout: std::time::Duration,
386 ) -> Result<BackgroundAgentTaskSnapshot, String> {
387 tokio::time::timeout(timeout, self.wait())
388 .await
389 .map_err(|_| {
390 format!(
391 "background agent task {} was not ready before timeout",
392 self.task_id
393 )
394 })?
395 }
396}
397
398struct BackgroundAgentTaskState {
399 status: AgentStatus,
400 result: Option<RunResult>,
401 error: Option<String>,
402}
403
404#[derive(Debug, Clone, PartialEq, Eq)]
405pub struct BackgroundAgentTaskSnapshot {
406 task_id: String,
407 agent_name: String,
408 status: AgentStatus,
409 final_output: Option<String>,
410 error: Option<String>,
411}
412
413impl BackgroundAgentTaskSnapshot {
414 pub fn task_id(&self) -> &str {
415 &self.task_id
416 }
417
418 pub fn agent_name(&self) -> &str {
419 &self.agent_name
420 }
421
422 pub fn status(&self) -> AgentStatus {
423 self.status
424 }
425
426 pub fn final_output(&self) -> Option<&str> {
427 self.final_output.as_deref()
428 }
429
430 pub fn error(&self) -> Option<&str> {
431 self.error.as_deref()
432 }
433}
434
435#[cfg(test)]
436mod tests {
437 use std::sync::Arc;
438
439 use serde_json::json;
440
441 use super::inherited_run_config;
442 use crate::context_providers::{
443 ContextError, ContextFragment, ContextProvider, ContextRequest,
444 };
445 use crate::memory::{
446 MemoryFuture, MemoryProvider, MemorySaveRequest, MemorySaveResult, MemorySearchRequest,
447 MemorySearchResult,
448 };
449 use crate::model::ModelRef;
450 use crate::model_settings::ModelSettings;
451 use crate::run_config::RunConfig;
452 use crate::runtime::{CancellationToken, RuntimeHook, SubTaskManager, SubTaskTurnSnapshot};
453 use crate::sessions::MemorySession;
454 use crate::tools::{ApprovalPolicy, ToolContext, ToolPolicy, ToolRegistry};
455 use crate::types::Message;
456
457 struct NoopContextProvider;
458
459 impl ContextProvider for NoopContextProvider {
460 fn fragments(
461 &self,
462 _request: &ContextRequest<'_>,
463 ) -> Result<Vec<ContextFragment>, ContextError> {
464 Ok(Vec::new())
465 }
466 }
467
468 struct NoopMemoryProvider;
469
470 impl MemoryProvider for NoopMemoryProvider {
471 fn search(&self, _request: MemorySearchRequest) -> MemoryFuture<Vec<MemorySearchResult>> {
472 Box::pin(async { Ok(Vec::new()) })
473 }
474
475 fn save(&self, _request: MemorySaveRequest) -> MemoryFuture<MemorySaveResult> {
476 Box::pin(async { Ok(MemorySaveResult::default()) })
477 }
478 }
479
480 struct NoopHook;
481
482 impl RuntimeHook for NoopHook {}
483
484 #[test]
485 fn approval_resume_snapshot_derives_one_way_child_cancellation() {
486 let snapshot_parent = CancellationToken::default();
487 let fallback_parent = CancellationToken::default();
488 let mut context = ToolContext::new("./workspace");
489 context.sub_task_turn_snapshot = Some(SubTaskTurnSnapshot {
490 cancellation_token: Some(snapshot_parent.clone()),
491 ..SubTaskTurnSnapshot::default()
492 });
493 let child = {
494 let _scope = CancellationToken::enter_scope(Some(&fallback_parent));
495 inherited_run_config(&context, None)
496 .cancellation_token
497 .expect("derived child cancellation")
498 };
499
500 child.cancel();
501 assert!(!snapshot_parent.is_cancelled());
502 assert!(!fallback_parent.is_cancelled());
503
504 let second_child = {
505 let _scope = CancellationToken::enter_scope(Some(&fallback_parent));
506 inherited_run_config(&context, None)
507 .cancellation_token
508 .expect("derived child cancellation")
509 };
510 fallback_parent.cancel();
511 assert!(!second_child.is_cancelled());
512 snapshot_parent.cancel();
513 assert!(second_child.is_cancelled());
514 }
515
516 #[test]
517 fn thread_local_and_explicit_cancellation_are_derived_for_the_child() {
518 let context = ToolContext::new("./workspace");
519 let fallback_parent = CancellationToken::default();
520 let child = {
521 let _scope = CancellationToken::enter_scope(Some(&fallback_parent));
522 inherited_run_config(&context, None)
523 .cancellation_token
524 .expect("derived fallback child cancellation")
525 };
526 fallback_parent.cancel();
527 assert!(child.is_cancelled());
528
529 let unrelated_parent = CancellationToken::default();
530 let explicit = CancellationToken::default();
531 let config = {
532 let _scope = CancellationToken::enter_scope(Some(&unrelated_parent));
533 inherited_run_config(
534 &context,
535 Some(RunConfig {
536 cancellation_token: Some(explicit.clone()),
537 ..RunConfig::default()
538 }),
539 )
540 };
541 unrelated_parent.cancel();
542 assert!(!explicit.is_cancelled());
543 let explicit_child = config
544 .cancellation_token
545 .expect("derived explicit child cancellation");
546 explicit_child.cancel();
547 assert!(!explicit.is_cancelled());
548
549 let second_explicit_child = inherited_run_config(
550 &context,
551 Some(RunConfig {
552 cancellation_token: Some(explicit.clone()),
553 ..RunConfig::default()
554 }),
555 )
556 .cancellation_token
557 .expect("second explicit child cancellation");
558 explicit.cancel();
559 assert!(second_explicit_child.is_cancelled());
560 assert!(config.session.is_none());
561 assert!(config.approval_broker.is_none());
562 }
563
564 #[test]
565 fn projected_parent_preserves_capabilities_and_clears_run_instances() {
566 let parent_cancellation = CancellationToken::default();
567 let mut parent = RunConfig {
568 model: Some(ModelRef::named("parent-model")),
569 model_settings: Some(ModelSettings::default()),
570 session: Some(Arc::new(MemorySession::new("parent-session"))),
571 initial_messages: Some(vec![Message::user("parent history")]),
572 max_cycles: Some(7),
573 max_handoffs: Some(4),
574 tool_policy: ToolPolicy {
575 disallowed_tools: vec!["blocked".to_string()],
576 approval: ApprovalPolicy::OnRequest,
577 ..ToolPolicy::default()
578 },
579 cancellation_token: Some(parent_cancellation.clone()),
580 hooks: vec![Arc::new(NoopHook)],
581 context_providers: vec![Arc::new(NoopContextProvider)],
582 max_context_chars: Some(12_345),
583 memory_providers: vec![Arc::new(NoopMemoryProvider)],
584 app_state: Some(Arc::new("parent app state".to_string())),
585 tool_registry_factory: Some(Arc::new(ToolRegistry::default)),
586 log_preview_chars: Some(321),
587 before_cycle_messages: Some(Arc::new(|_, _, _| Vec::new())),
588 interruption_messages: Some(Arc::new(Vec::new)),
589 sub_task_manager: Some(SubTaskManager::default()),
590 runtime_log_handler: Some(Arc::new(|_, _| {})),
591 runtime_stream_callback: Some(Arc::new(|_| {})),
592 ..RunConfig::default()
593 };
594 parent
595 .metadata
596 .insert("parent_metadata".to_string(), json!("retained"));
597 parent
598 .initial_shared_state
599 .insert("stale".to_string(), json!(true));
600
601 let mut context = ToolContext::new("./workspace");
602 context.background_parent_run_config = Some(parent);
603 context
604 .shared_state
605 .insert("live".to_string(), json!("snapshot"));
606 let child = inherited_run_config(&context, None);
607
608 assert!(child.model.is_none());
609 assert!(child.model_settings.is_none());
610 assert!(child.session.is_none());
611 assert!(child.initial_messages.is_none());
612 assert!(child.before_cycle_messages.is_none());
613 assert!(child.interruption_messages.is_none());
614 assert!(child.sub_task_manager.is_none());
615 assert!(child.runtime_log_handler.is_none());
616 assert!(child.runtime_stream_callback.is_none());
617 assert_eq!(
618 child.initial_shared_state.get("live"),
619 Some(&json!("snapshot"))
620 );
621 assert!(!child.initial_shared_state.contains_key("stale"));
622
623 assert_eq!(child.max_cycles, Some(7));
624 assert_eq!(child.max_handoffs, Some(4));
625 assert_eq!(child.tool_policy.approval, ApprovalPolicy::OnRequest);
626 assert_eq!(child.tool_policy.disallowed_tools, ["blocked"]);
627 assert_eq!(child.hooks.len(), 1);
628 assert_eq!(child.context_providers.len(), 1);
629 assert_eq!(child.max_context_chars, Some(12_345));
630 assert_eq!(child.memory_providers.len(), 1);
631 assert!(child.app_state.is_some());
632 assert!(child.tool_registry_factory.is_some());
633 assert_eq!(child.log_preview_chars, Some(321));
634 assert_eq!(child.metadata["parent_metadata"], json!("retained"));
635
636 let child_cancellation = child
637 .cancellation_token
638 .expect("derived child cancellation");
639 child_cancellation.cancel();
640 assert!(!parent_cancellation.is_cancelled());
641 let second_child = inherited_run_config(&context, None)
642 .cancellation_token
643 .expect("second child cancellation");
644 parent_cancellation.cancel();
645 assert!(second_child.is_cancelled());
646 }
647}