1use bamboo_domain::Session;
4use chrono::Utc;
5use serde_json::json;
6
7use super::helpers::{
8 compute_status_guidance, format_child_assignment, map_child_entry, metadata_text,
9 normalize_non_empty_optional, normalize_required_text, render_forked_parent_context,
10 replace_or_append_last_user_message, truncate_after_index, truncate_after_last_user,
11};
12use super::DELEGATION_NOTE;
13use super::{
14 ChildSessionEntry, ChildSessionError, ChildSessionPort, CreateChildInput, CreateChildResult,
15 QueuedInjectedMessage,
16};
17
18pub async fn create_child_action(
19 port: &dyn ChildSessionPort,
20 input: CreateChildInput,
21) -> Result<CreateChildResult, ChildSessionError> {
22 use crate::runner::refresh_prompt_snapshot;
23 use bamboo_agent_core::Message;
24
25 let mut child = Session::new_child_of(
30 input.child_id.clone(),
31 &input.parent_session,
32 input
33 .model_ref_override
34 .as_ref()
35 .map(|model_ref| model_ref.model.clone())
36 .or_else(|| input.model_override.clone())
37 .unwrap_or_else(|| input.parent_session.model.clone()),
38 input.title.clone(),
39 );
40
41 if let Some(model_ref) = input.model_ref_override.clone() {
42 child.model_ref = Some(model_ref.clone());
43 child
44 .metadata
45 .insert("provider_name".to_string(), model_ref.provider);
46 } else if let Some(parent_model_ref) = input.parent_session.model_ref.clone() {
47 child.model_ref = Some(parent_model_ref.clone());
48 child.set_provider_name(parent_model_ref.provider);
49 } else if let Some(parent_provider) = input.parent_session.provider_name() {
50 child.set_provider_name(parent_provider);
51 }
52
53 if let Some(effort) = input.reasoning_effort {
57 child.reasoning_effort = Some(effort);
58 }
59
60 if input
65 .parent_session
66 .agent_runtime_state
67 .as_ref()
68 .is_some_and(|state| state.bypass_permissions)
69 {
70 child
71 .agent_runtime_state
72 .get_or_insert_with(bamboo_domain::AgentRuntimeState::default)
73 .bypass_permissions = true;
74 }
75
76 if input
81 .parent_session
82 .agent_runtime_state
83 .as_ref()
84 .is_some_and(|state| state.no_human_approver)
85 {
86 child
87 .agent_runtime_state
88 .get_or_insert_with(bamboo_domain::AgentRuntimeState::default)
89 .no_human_approver = true;
90 }
91
92 let stored_workspace = bamboo_agent_core::workspace_state::set_workspace(
97 &child.id,
98 std::path::PathBuf::from(&input.workspace),
99 );
100 child.workspace = Some(stored_workspace.to_string_lossy().to_string());
101
102 child
103 .metadata
104 .insert("spawned_by".to_string(), "SubAgent".to_string());
105 child.set_subagent_type(input.subagent_type.clone());
106 child
107 .metadata
108 .insert("responsibility".to_string(), input.responsibility.clone());
109 child.metadata.insert(
110 "assignment_prompt".to_string(),
111 input.assignment_prompt.clone(),
112 );
113 if input.lifecycle.as_deref() == Some("resident") {
118 child
119 .metadata
120 .insert("lifecycle".to_string(), "resident".to_string());
121 if let Some(name) = input.resident_name.clone().filter(|n| !n.trim().is_empty()) {
122 child.metadata.insert("resident_name".to_string(), name);
123 }
124 child.metadata.insert(
125 "resident_context".to_string(),
126 input
127 .resident_context
128 .clone()
129 .filter(|c| matches!(c.as_str(), "reset" | "accumulate"))
130 .unwrap_or_else(|| "reset".to_string()),
131 );
132 }
133 child.set_last_run_status("pending");
134 child.clear_last_run_error();
135
136 for (key, value) in input.runtime_metadata {
138 child.metadata.insert(key, value);
139 }
140
141 let base_prompt = {
148 let global = crate::prompt_defaults::read_global_default_system_prompt_template();
149 if global.trim().is_empty() {
150 crate::context::DEFAULT_BASE_PROMPT.to_string()
151 } else {
152 global
153 }
154 };
155 let system_prompt = format!("{base_prompt}\n\n{DELEGATION_NOTE}");
156
157 child
158 .metadata
159 .insert("base_system_prompt".to_string(), system_prompt.clone());
160
161 child.add_message(Message::system(&system_prompt));
162
163 if let Some(parent_budget) = input.parent_session.effective_token_budget() {
167 let mut child_budget = parent_budget.clone();
168 child_budget.compression_trigger_percent = 70;
169 child_budget.compression_target_percent = 35;
170 child.token_budget = Some(child_budget);
171 }
172
173 refresh_prompt_snapshot(&mut child);
174 let assignment = format_child_assignment(
175 &input.title,
176 &input.responsibility,
177 &input.subagent_type,
178 &input.assignment_prompt,
179 );
180 let assignment = match input
184 .context_fork
185 .and_then(|n| render_forked_parent_context(&input.parent_session, n))
186 {
187 Some(forked) => format!("{forked}\n\n{assignment}"),
188 None => assignment,
189 };
190 child.add_message(Message::user(assignment));
191
192 if let Some(parent_task_list) = input.parent_session.task_list.clone() {
193 child.set_task_list(parent_task_list);
194 }
195
196 if let Some(ref disabled) = input.disabled_tools {
200 if !disabled.is_empty() {
201 child.metadata.insert(
202 "disabled_tools".to_string(),
203 serde_json::to_string(disabled).unwrap_or_default(),
204 );
205 }
206 }
207
208 let model = child.model.clone();
209 port.save_child_session(&mut child).await?;
210 if input.auto_run {
211 port.enqueue_child_run(&input.parent_session, &child)
212 .await?;
213 }
214
215 Ok(CreateChildResult {
216 child_session_id: child.id,
217 model,
218 })
219}
220
221pub async fn list_children_action(
222 port: &dyn ChildSessionPort,
223 parent_id: &str,
224) -> serde_json::Value {
225 let children = port.list_children(parent_id).await;
226 json!({
227 "parent_session_id": parent_id,
228 "children": children.iter().map(map_child_entry).collect::<Vec<_>>(),
229 "count": children.len(),
230 })
231}
232
233#[derive(Debug, Clone, PartialEq, serde::Serialize)]
236pub struct SessionTreeNode {
237 pub session_id: String,
238 pub title: String,
239 #[serde(skip_serializing_if = "Option::is_none")]
240 pub last_run_status: Option<String>,
241 pub depth: u32,
242 pub children: Vec<SessionTreeNode>,
243}
244
245pub fn assemble_session_tree(
250 root_id: &str,
251 root_title: &str,
252 adjacency: &std::collections::HashMap<String, Vec<ChildSessionEntry>>,
253 max_depth: u32,
254) -> SessionTreeNode {
255 fn build(
256 id: &str,
257 title: &str,
258 status: Option<String>,
259 depth: u32,
260 max_depth: u32,
261 adjacency: &std::collections::HashMap<String, Vec<ChildSessionEntry>>,
262 visited: &mut std::collections::HashSet<String>,
263 ) -> SessionTreeNode {
264 let first_visit = visited.insert(id.to_string());
265 let mut children = Vec::new();
266 if first_visit && depth < max_depth {
267 if let Some(kids) = adjacency.get(id) {
268 for kid in kids {
269 children.push(build(
270 &kid.child_session_id,
271 &kid.title,
272 kid.last_run_status.clone(),
273 depth + 1,
274 max_depth,
275 adjacency,
276 visited,
277 ));
278 }
279 }
280 }
281 SessionTreeNode {
282 session_id: id.to_string(),
283 title: title.to_string(),
284 last_run_status: status,
285 depth,
286 children,
287 }
288 }
289 let mut visited = std::collections::HashSet::new();
290 build(
291 root_id,
292 root_title,
293 None,
294 0,
295 max_depth,
296 adjacency,
297 &mut visited,
298 )
299}
300
301pub async fn build_session_tree_action(
307 port: &dyn ChildSessionPort,
308 root_id: &str,
309 max_depth: u32,
310) -> SessionTreeNode {
311 use std::collections::{HashMap, HashSet, VecDeque};
312 const NODE_CAP: usize = 5000;
313
314 let root_title = port
315 .load_root_session(root_id)
316 .await
317 .map(|s| s.title)
318 .unwrap_or_default();
319
320 let mut adjacency: HashMap<String, Vec<ChildSessionEntry>> = HashMap::new();
321 let mut visited: HashSet<String> = HashSet::new();
322 let mut queue: VecDeque<(String, u32)> = VecDeque::new();
323 queue.push_back((root_id.to_string(), 0));
324
325 while let Some((id, depth)) = queue.pop_front() {
326 if depth >= max_depth || adjacency.len() >= NODE_CAP || !visited.insert(id.clone()) {
327 continue;
328 }
329 let kids = port.list_children(&id).await;
330 for kid in &kids {
331 queue.push_back((kid.child_session_id.clone(), depth + 1));
332 }
333 adjacency.insert(id, kids);
334 }
335
336 assemble_session_tree(root_id, &root_title, &adjacency, max_depth)
337}
338
339pub async fn get_child_action(
340 port: &dyn ChildSessionPort,
341 parent_id: &str,
342 child_session_id: String,
343) -> Result<serde_json::Value, ChildSessionError> {
344 let child = port
345 .load_child_for_parent(parent_id, &child_session_id)
346 .await?;
347
348 let status = metadata_text(&child, "last_run_status");
349 let runner_info = port.get_child_runner_info(&child.id).await;
350
351 Ok(json!({
352 "child_session_id": child.id,
353 "title": child.title,
354 "model": child.model,
355 "pinned": child.pinned,
356 "message_count": child.messages.len(),
357 "is_running": port.is_child_running(&child.id).await,
358 "last_run_status": status,
359 "last_run_error": metadata_text(&child, "last_run_error"),
360 "responsibility": metadata_text(&child, "responsibility"),
361 "subagent_type": metadata_text(&child, "subagent_type"),
362 "prompt": metadata_text(&child, "assignment_prompt"),
363 "latest_user_message": child
364 .messages
365 .iter()
366 .rposition(|message| matches!(message.role, bamboo_agent_core::Role::User))
367 .and_then(|idx| child.messages.get(idx))
368 .map(|message| message.content.clone()),
369 "runtime_kind": metadata_text(&child, "runtime.kind"),
370 "external_protocol": metadata_text(&child, "external.protocol"),
371 "external_agent_id": metadata_text(&child, "external.agent_id"),
372 "a2a_context_id": metadata_text(&child, "a2a.context_id"),
373 "a2a_latest_task_id": metadata_text(&child, "a2a.latest_task_id"),
374 "a2a_last_state": metadata_text(&child, "a2a.last_state"),
375 "runner_started_at": runner_info.as_ref().and_then(|r| r.started_at.map(|t| t.to_rfc3339())),
376 "runner_completed_at": runner_info.as_ref().and_then(|r| r.completed_at.map(|t| t.to_rfc3339())),
377 "last_tool_name": runner_info.as_ref().and_then(|r| r.last_tool_name.clone()),
378 "last_tool_phase": runner_info.as_ref().and_then(|r| r.last_tool_phase.clone()),
379 "last_event_at": runner_info.as_ref().and_then(|r| r.last_event_at.map(|t| t.to_rfc3339())),
380 "round_count": runner_info.as_ref().map(|r| r.round_count).unwrap_or(0),
381 "has_pending_injected_messages": child.has_pending_injected_messages(),
382 "guidance": compute_status_guidance(status.as_deref(), runner_info.as_ref(), child.has_pending_injected_messages()),
383 }))
384}
385
386#[allow(clippy::too_many_arguments)]
387pub async fn update_child_action(
388 port: &dyn ChildSessionPort,
389 parent_id: &str,
390 child_session_id: String,
391 title: Option<String>,
392 responsibility: Option<String>,
393 prompt: Option<String>,
394 subagent_type: Option<String>,
395 reset_after_update: Option<bool>,
396 reasoning_effort: Option<bamboo_domain::ReasoningEffort>,
397) -> Result<serde_json::Value, ChildSessionError> {
398 let mut child = port
399 .load_child_for_parent(parent_id, &child_session_id)
400 .await?;
401
402 let title = normalize_non_empty_optional(title, "title")?;
403 let responsibility = normalize_non_empty_optional(responsibility, "responsibility")?;
404 let prompt = normalize_non_empty_optional(prompt, "prompt")?;
405 let subagent_type = normalize_non_empty_optional(subagent_type, "subagent_type")?;
406
407 let should_refresh_assignment =
408 responsibility.is_some() || prompt.is_some() || subagent_type.is_some();
409
410 if title.is_none() && !should_refresh_assignment && reasoning_effort.is_none() {
411 return Err(ChildSessionError::InvalidArguments(
412 "update requires at least one field: title/responsibility/prompt/subagent_type/reasoning_effort"
413 .to_string(),
414 ));
415 }
416
417 if let Some(effort) = reasoning_effort {
418 child.reasoning_effort = Some(effort);
419 }
420
421 if let Some(title) = title {
422 child.title = title;
423 }
424
425 let mut messages_removed = 0usize;
426
427 if should_refresh_assignment {
428 let effective_responsibility = normalize_required_text(
429 responsibility.or_else(|| metadata_text(&child, "responsibility")),
430 "responsibility",
431 )?;
432 let effective_subagent_type = normalize_required_text(
433 subagent_type.or_else(|| metadata_text(&child, "subagent_type")),
434 "subagent_type",
435 )?;
436 let effective_prompt = normalize_required_text(
437 prompt.or_else(|| metadata_text(&child, "assignment_prompt")),
438 "prompt",
439 )?;
440
441 child.metadata.insert(
442 "responsibility".to_string(),
443 effective_responsibility.clone(),
444 );
445 child
446 .metadata
447 .insert("subagent_type".to_string(), effective_subagent_type.clone());
448 child
449 .metadata
450 .insert("assignment_prompt".to_string(), effective_prompt.clone());
451 child.set_last_run_status("pending");
452 child.clear_last_run_error();
453
454 let assignment = format_child_assignment(
455 &child.title,
456 &effective_responsibility,
457 &effective_subagent_type,
458 &effective_prompt,
459 );
460 let user_index = replace_or_append_last_user_message(&mut child, assignment);
461
462 if reset_after_update.unwrap_or(true) {
463 messages_removed = truncate_after_index(&mut child, user_index);
464 }
465 }
466
467 child.updated_at = Utc::now();
468 port.save_child_session(&mut child).await?;
469
470 Ok(json!({
471 "child_session_id": child.id,
472 "title": child.title,
473 "messages_removed": messages_removed,
474 "last_run_status": metadata_text(&child, "last_run_status"),
475 "note": "Child session updated in place. Use action=run to execute the same child session.",
476 }))
477}
478
479pub async fn run_child_action(
480 port: &dyn ChildSessionPort,
481 parent: &Session,
482 child_session_id: String,
483 reset_to_last_user: Option<bool>,
484) -> Result<serde_json::Value, ChildSessionError> {
485 let mut child = port
486 .load_child_for_parent(&parent.id, &child_session_id)
487 .await?;
488
489 if port.is_child_running(&child.id).await {
490 return Ok(json!({
491 "child_session_id": child.id,
492 "status": "already_running",
493 "note": "Child session is already running.",
494 }));
495 }
496
497 let mut messages_removed = 0usize;
498 if reset_to_last_user.unwrap_or(true) {
499 messages_removed = truncate_after_last_user(&mut child)?;
500 }
501
502 child.set_last_run_status("pending");
503 child.clear_last_run_error();
504 child.updated_at = Utc::now();
505 port.save_child_session(&mut child).await?;
506
507 port.enqueue_child_run(parent, &child).await?;
508
509 Ok(json!({
510 "child_session_id": child.id,
511 "status": "queued",
512 "messages_removed": messages_removed,
513 "note": "Queued existing child session for retry in place.",
514 }))
515}
516
517pub async fn send_message_to_child_action(
518 port: &dyn ChildSessionPort,
519 parent: &Session,
520 child_session_id: String,
521 message: String,
522 auto_run: Option<bool>,
523 interrupt_running: Option<bool>,
524) -> Result<serde_json::Value, ChildSessionError> {
525 let mut child = port
526 .load_child_for_parent(&parent.id, &child_session_id)
527 .await?;
528
529 let is_running = port.is_child_running(&child.id).await;
530 let should_interrupt = interrupt_running.unwrap_or(false);
531
532 if is_running && should_interrupt {
533 port.cancel_child_run_and_wait(&child.id).await?;
534 child = port
535 .load_child_for_parent(&parent.id, &child_session_id)
536 .await?;
537 }
538
539 let message = normalize_required_text(Some(message), "message")?;
540
541 if is_running && !should_interrupt {
542 if crate::external_agents::live::deliver_message(&child.id, &message) {
548 child.add_message(bamboo_agent_core::Message::user(message.clone()));
549 port.save_child_session(&mut child).await?;
550 return Ok(json!({
551 "child_session_id": child.id,
552 "status": "message_delivered_live",
553 "auto_run": false,
554 "message": message,
555 "message_count": child.messages.len(),
556 "note": "Message delivered to the running actor in-band; it will be admitted at the next round boundary without canceling progress.",
557 }));
558 }
559
560 let mut pending = child.pending_injected_messages().unwrap_or_default();
565 let queued = QueuedInjectedMessage {
566 content: message.clone(),
567 created_at: Some(chrono::Utc::now()),
568 };
569 pending.push(serde_json::to_value(&queued).unwrap_or(serde_json::Value::Null));
570 child.set_pending_injected_messages(pending);
571 port.save_child_session(&mut child).await?;
572
573 if !port.is_child_running(&child.id).await {
578 port.enqueue_child_run(parent, &child).await?;
579 return Ok(json!({
580 "child_session_id": child.id,
581 "status": "queued",
582 "auto_run": true,
583 "message": message,
584 "message_count": child.messages.len(),
585 "note": "Child finished while the message was being queued; a new run was scheduled to process it.",
586 }));
587 }
588
589 return Ok(json!({
590 "child_session_id": child.id,
591 "status": "message_queued",
592 "auto_run": false,
593 "message": message,
594 "message_count": child.messages.len(),
595 "note": "Message queued for the child session. It will be picked up at the next turn boundary without canceling current progress.",
596 }));
597 }
598
599 child.add_message(bamboo_agent_core::Message::user(message.clone()));
600 child.set_last_run_status("pending");
601 child.clear_last_run_error();
602 port.save_child_session(&mut child).await?;
603
604 let should_auto_run = auto_run.unwrap_or(true);
605 if should_auto_run {
606 port.enqueue_child_run(parent, &child).await?;
607 }
608
609 Ok(json!({
610 "child_session_id": child.id,
611 "status": if should_auto_run { "queued" } else { "pending" },
612 "auto_run": should_auto_run,
613 "message": message,
614 "message_count": child.messages.len(),
615 "note": if should_auto_run {
616 "Follow-up message appended and child session queued."
617 } else {
618 "Follow-up message appended. Use action=run to execute the child session."
619 },
620 }))
621}
622
623pub async fn cancel_child_action(
624 port: &dyn ChildSessionPort,
625 parent_id: &str,
626 child_session_id: String,
627) -> Result<serde_json::Value, ChildSessionError> {
628 let _ = port
630 .load_child_for_parent(parent_id, &child_session_id)
631 .await?;
632 port.cancel_child_run_and_wait(&child_session_id).await?;
633
634 let mut child = port
639 .load_child_for_parent(parent_id, &child_session_id)
640 .await?;
641 let latest_status = child.last_run_status().unwrap_or_default();
642 if matches!(latest_status.as_str(), "completed" | "error") {
643 return Ok(json!({
644 "child_session_id": child_session_id,
645 "status": latest_status,
646 "note": "Child reached a natural terminal state while the cancel was in flight; its real outcome was kept.",
647 }));
648 }
649 child.set_last_run_status("cancelled");
650 child.set_last_run_error("Cancelled by parent");
651 port.save_child_session(&mut child).await?;
652 Ok(json!({
653 "child_session_id": child_session_id,
654 "status": "cancelled",
655 }))
656}
657
658pub async fn delete_child_action(
659 port: &dyn ChildSessionPort,
660 parent_id: &str,
661 child_session_id: String,
662) -> Result<serde_json::Value, ChildSessionError> {
663 let child = port
665 .load_child_for_parent(parent_id, &child_session_id)
666 .await?;
667 let result = port.delete_child_session(parent_id, &child.id).await?;
668
669 if !result.deleted {
670 return Err(ChildSessionError::Execution(format!(
671 "child session was not deleted: {}",
672 child.id
673 )));
674 }
675
676 Ok(json!({
677 "child_session_id": child.id,
678 "deleted": true,
679 "cancelled_running_child": result.cancelled_running_child,
680 }))
681}
682
683#[cfg(test)]
684mod tree_tests {
685 use super::super::ChildSessionEntry;
686 use super::assemble_session_tree;
687 use std::collections::HashMap;
688
689 fn entry(id: &str, title: &str) -> ChildSessionEntry {
690 ChildSessionEntry {
691 child_session_id: id.to_string(),
692 title: title.to_string(),
693 pinned: false,
694 message_count: 0,
695 updated_at: String::new(),
696 last_run_status: Some("completed".to_string()),
697 last_run_error: None,
698 }
699 }
700
701 #[test]
702 fn assembles_multi_level_tree() {
703 let mut adj: HashMap<String, Vec<ChildSessionEntry>> = HashMap::new();
704 adj.insert(
705 "root".into(),
706 vec![entry("c1", "child 1"), entry("c2", "child 2")],
707 );
708 adj.insert("c1".into(), vec![entry("g1", "grandchild")]);
709
710 let tree = assemble_session_tree("root", "Root", &adj, 8);
711 assert_eq!(tree.session_id, "root");
712 assert_eq!(tree.depth, 0);
713 assert_eq!(tree.children.len(), 2);
714 let c1 = tree.children.iter().find(|n| n.session_id == "c1").unwrap();
715 assert_eq!(c1.depth, 1);
716 assert_eq!(c1.children.len(), 1);
717 assert_eq!(c1.children[0].session_id, "g1");
718 assert_eq!(c1.children[0].depth, 2);
719 let c2 = tree.children.iter().find(|n| n.session_id == "c2").unwrap();
720 assert!(c2.children.is_empty());
721 }
722
723 #[test]
724 fn depth_cap_stops_descent() {
725 let mut adj: HashMap<String, Vec<ChildSessionEntry>> = HashMap::new();
726 adj.insert("root".into(), vec![entry("c1", "c1")]);
727 adj.insert("c1".into(), vec![entry("g1", "g1")]);
728 let tree = assemble_session_tree("root", "Root", &adj, 1);
729 assert_eq!(tree.children.len(), 1);
730 assert!(
731 tree.children[0].children.is_empty(),
732 "depth cap stops expansion at depth 1"
733 );
734 }
735
736 #[test]
737 fn cycle_is_broken_by_first_visit_guard() {
738 let mut adj: HashMap<String, Vec<ChildSessionEntry>> = HashMap::new();
739 adj.insert("a".into(), vec![entry("b", "b")]);
740 adj.insert("b".into(), vec![entry("a", "a")]); let tree = assemble_session_tree("a", "A", &adj, 100);
742 assert_eq!(tree.children.len(), 1);
743 let b = &tree.children[0];
744 assert_eq!(b.session_id, "b");
745 assert_eq!(b.children.len(), 1);
746 let a2 = &b.children[0];
747 assert_eq!(a2.session_id, "a");
748 assert!(a2.children.is_empty(), "cycle must terminate as a leaf");
749 }
750}