1use std::sync::Arc;
2
3use crate::runtime::execution::{ExternalChildRunner, SpawnJob};
4use async_trait::async_trait;
5use bamboo_a2a::A2AJsonRpcClient;
6use bamboo_agent_core::{AgentError, AgentEvent};
7use bamboo_llm::Config;
8use tokio::sync::mpsc;
9use tokio_util::sync::CancellationToken;
10
11use super::a2a_adapter::A2AExternalChildRunner;
12use super::actor_adapter::{ActorChildRunner, ChildApprovalReviewer};
13use super::config::{parse_external_agents, ExternalAgentProtocol};
14
15pub struct CompositeExternalChildRunner {
17 runners: Vec<Arc<dyn ExternalChildRunner>>,
18}
19
20impl CompositeExternalChildRunner {
21 pub fn new(runners: Vec<Arc<dyn ExternalChildRunner>>) -> Self {
22 Self { runners }
23 }
24}
25
26#[async_trait]
27impl ExternalChildRunner for CompositeExternalChildRunner {
28 async fn should_handle(&self, session: &bamboo_agent_core::Session) -> bool {
29 for runner in &self.runners {
30 if runner.should_handle(session).await {
31 return true;
32 }
33 }
34 false
35 }
36
37 async fn execute_external_child(
38 &self,
39 session: &mut bamboo_agent_core::Session,
40 job: &SpawnJob,
41 event_tx: mpsc::Sender<AgentEvent>,
42 cancel_token: CancellationToken,
43 ) -> crate::runtime::runner::Result<()> {
44 for runner in &self.runners {
45 if runner.should_handle(session).await {
46 return runner
47 .execute_external_child(session, job, event_tx, cancel_token)
48 .await;
49 }
50 }
51 Err(AgentError::LLM(
52 "No matching external child runner found for session metadata".to_string(),
53 ))
54 }
55
56 fn set_escalation_bridge(&self, bridge: Option<bamboo_subagent::executor::HostBridge>) {
61 for runner in &self.runners {
62 runner.set_escalation_bridge(bridge.clone());
63 }
64 }
65}
66
67pub fn build_external_child_runner(config: &Config) -> Arc<dyn ExternalChildRunner> {
76 build_external_child_runner_with_registry(config, None)
77}
78
79pub fn build_external_child_runner_with_registry(
81 config: &Config,
82 approval_registry: Option<super::approval_registry::SharedApprovalRegistry>,
83) -> Arc<dyn ExternalChildRunner> {
84 build_external_child_runner_with_registry_and_reviewer(config, approval_registry, None, None)
85}
86
87pub fn build_external_child_runner_with_registry_and_reviewer(
90 config: &Config,
91 approval_registry: Option<super::approval_registry::SharedApprovalRegistry>,
92 approval_reviewer: Option<Arc<dyn ChildApprovalReviewer>>,
93 permission_config: Option<Arc<bamboo_tools::permission::PermissionConfig>>,
94) -> Arc<dyn ExternalChildRunner> {
95 let agents = parse_external_agents(config);
96
97 let mut runners: Vec<Arc<dyn ExternalChildRunner>> = Vec::new();
98
99 match build_local_actor_runner(
103 config,
104 approval_registry.clone(),
105 approval_reviewer.clone(),
106 permission_config.clone(),
107 ) {
108 Ok(runner) => runners.push(runner),
109 Err(e) => tracing::error!("local actor sub-agent runner unavailable: {e}"),
110 }
111
112 for (_agent_id, profile) in agents {
113 if matches!(profile.protocol, ExternalAgentProtocol::Actor) {
115 let Some(worker_bin) = profile.worker_bin.as_ref() else {
116 tracing::error!(
117 "Actor agent profile {} has no worker_bin; skipping",
118 profile.agent_id
119 );
120 continue;
121 };
122 let fabric_dir = profile
126 .fabric_dir
127 .clone()
128 .map(std::path::PathBuf::from)
129 .unwrap_or_else(bamboo_config::paths::subagents_dir);
130 let executor = match profile.executor.as_deref() {
131 Some("echo") => bamboo_subagent::provision::ExecutorSpec::Echo,
132 Some("bamboo_runtime") | None => {
133 bamboo_subagent::provision::ExecutorSpec::BambooRuntime
134 }
135 Some("claude_code") => bamboo_subagent::provision::ExecutorSpec::ClaudeCode {
138 binary: profile.claude_code_binary.clone(),
139 model: profile.claude_code_model.clone(),
140 permission_mode: profile.claude_code_permission_mode.clone(),
141 inherit_user_config: profile.claude_code_inherit_user_config,
142 forward_env: profile.claude_code_forward_env.clone(),
143 },
144 Some(other) => {
145 tracing::error!(
146 "Actor agent profile {} has unknown executor '{}'; skipping",
147 profile.agent_id,
148 other
149 );
150 continue;
151 }
152 };
153 let mut runner = ActorChildRunner::new(
154 profile.agent_id.clone(),
155 std::path::PathBuf::from(worker_bin),
156 profile.worker_args.clone(),
157 fabric_dir,
158 executor,
159 extract_provider_credentials(config),
160 config.provider.clone(),
161 config
162 .subagents()
163 .max_concurrent
164 .unwrap_or(super::actor_adapter::DEFAULT_MAX_CONCURRENT_ACTORS),
165 );
166 if let Some(registry) = approval_registry.clone() {
167 runner = runner.with_approval_registry(registry);
168 }
169 if let Some(reviewer) = approval_reviewer.clone() {
170 runner = runner.with_approval_reviewer(reviewer);
171 }
172 if let Some(config) = permission_config.clone() {
173 runner = runner.with_permission_config(config);
174 }
175 runners.push(Arc::new(runner));
176 continue;
177 }
178
179 if !matches!(profile.protocol, ExternalAgentProtocol::A2aJsonRpc) {
180 tracing::warn!(
181 "External agent profile {} uses unsupported protocol {:?}",
182 profile.agent_id,
183 profile.protocol
184 );
185 continue;
186 }
187
188 let auth_token = match profile.auth_ref.as_ref() {
189 Some(ref_name) => match std::env::var(ref_name) {
190 Ok(token) => Some(token),
191 Err(_) => {
192 tracing::error!(
193 "External agent profile {} auth_ref env var {} is not set",
194 profile.agent_id,
195 ref_name
196 );
197 continue;
198 }
199 },
200 None => None,
201 };
202
203 let client_config = match A2AExternalChildRunner::build_client_config(&profile, auth_token)
204 {
205 Ok(cfg) => cfg,
206 Err(e) => {
207 tracing::error!(
208 "Failed to build A2A client config for profile {}: {}",
209 profile.agent_id,
210 e
211 );
212 continue;
213 }
214 };
215
216 let client = match A2AJsonRpcClient::new(client_config) {
217 Ok(c) => c,
218 Err(e) => {
219 tracing::error!(
220 "Failed to create A2A JSON-RPC client for profile {}: {}",
221 profile.agent_id,
222 e
223 );
224 continue;
225 }
226 };
227
228 runners.push(Arc::new(A2AExternalChildRunner::new(client, profile)));
229 }
230
231 Arc::new(CompositeExternalChildRunner::new(runners))
232}
233
234fn build_local_actor_runner(
239 config: &Config,
240 approval_registry: Option<super::approval_registry::SharedApprovalRegistry>,
241 approval_reviewer: Option<Arc<dyn ChildApprovalReviewer>>,
242 permission_config: Option<Arc<bamboo_tools::permission::PermissionConfig>>,
243) -> Result<Arc<dyn ExternalChildRunner>, String> {
244 let sub = config.subagents();
245
246 let (worker_bin, worker_args) = match &sub.worker_bin {
247 Some(custom) => (
248 std::path::PathBuf::from(custom),
249 sub.worker_args.clone().unwrap_or_default(),
250 ),
251 None => (
252 std::env::current_exe().map_err(|e| format!("cannot locate own executable: {e}"))?,
253 sub.worker_args
254 .clone()
255 .unwrap_or_else(|| vec!["subagent-worker".to_string()]),
256 ),
257 };
258
259 let fabric_dir = sub
262 .fabric_dir
263 .clone()
264 .map(std::path::PathBuf::from)
265 .unwrap_or_else(bamboo_config::paths::subagents_dir);
266
267 let executor = match sub.executor.as_deref() {
268 Some("echo") => bamboo_subagent::provision::ExecutorSpec::Echo,
269 Some("bamboo_runtime") | None => bamboo_subagent::provision::ExecutorSpec::BambooRuntime,
270 Some("claude_code") => bamboo_subagent::provision::ExecutorSpec::ClaudeCode {
274 binary: sub.claude_code_binary.clone(),
275 model: sub.claude_code_model.clone(),
276 permission_mode: sub.claude_code_permission_mode.clone(),
277 inherit_user_config: sub.claude_code_inherit_user_config,
278 forward_env: sub.claude_code_forward_env.clone(),
279 },
280 Some(other) => return Err(format!("unknown subagents.executor '{other}'")),
281 };
282
283 let mut runner = ActorChildRunner::new(
284 super::config::LOCAL_ACTOR_AGENT_ID.to_string(),
285 worker_bin,
286 worker_args,
287 fabric_dir,
288 executor,
289 extract_provider_credentials(config),
290 config.provider.clone(),
291 sub.max_concurrent
292 .unwrap_or(super::actor_adapter::DEFAULT_MAX_CONCURRENT_ACTORS),
293 )
294 .with_remote_placements(resolve_remote_placements(
295 &sub.remote_placements,
296 &config.cluster_fabric.nodes,
297 ))
298 .with_schedulable_placements(resolve_schedulable_placements(
299 &sub.schedulable_placements,
300 &config.cluster_fabric.nodes,
301 ))
302 .with_bus(sub.broker.as_ref().map(|b| bamboo_subagent::BusEndpoint {
303 endpoint: b.endpoint.clone(),
304 token: b.token.clone(),
305 }));
306 if let Some(registry) = approval_registry {
307 runner = runner.with_approval_registry(registry);
308 }
309 if let Some(reviewer) = approval_reviewer {
310 runner = runner.with_approval_reviewer(reviewer);
311 }
312 if let Some(config) = permission_config {
313 runner = runner.with_permission_config(config);
314 }
315 Ok(Arc::new(runner))
316}
317
318fn resolve_schedulable_placements(
327 placements: &[bamboo_config::SchedulablePlacement],
328 nodes: &[bamboo_config::cluster_fabric::Node],
329) -> std::collections::HashMap<String, super::actor_adapter::ResolvedSchedulablePlacement> {
330 placements
333 .iter()
334 .map(|p| {
335 (
336 p.role.clone(),
337 super::actor_adapter::ResolvedSchedulablePlacement {
338 pool: p.pool.clone(),
339 host_label: node_label_for_role(nodes, &p.pool),
342 },
343 )
344 })
345 .collect()
346}
347
348fn node_label_for_role(
352 nodes: &[bamboo_config::cluster_fabric::Node],
353 role: &str,
354) -> Option<String> {
355 nodes
356 .iter()
357 .find(|n| n.deploy.default_role.as_deref() == Some(role))
358 .map(node_display_name)
359}
360
361fn node_label_for_endpoint(
365 nodes: &[bamboo_config::cluster_fabric::Node],
366 endpoint: &str,
367) -> Option<String> {
368 let host = endpoint
369 .trim()
370 .trim_start_matches("wss://")
371 .trim_start_matches("ws://")
372 .split(['/', ':'])
373 .next()
374 .unwrap_or("");
375 if host.is_empty() {
376 return None;
377 }
378 nodes
379 .iter()
380 .find(|n| match &n.placement {
381 bamboo_config::cluster_fabric::NodePlacement::Ssh(t) => t.host == host,
382 bamboo_config::cluster_fabric::NodePlacement::Local => false,
383 })
384 .map(node_display_name)
385}
386
387fn node_display_name(n: &bamboo_config::cluster_fabric::Node) -> String {
388 if !n.label.trim().is_empty() {
389 return n.label.clone();
390 }
391 match &n.placement {
392 bamboo_config::cluster_fabric::NodePlacement::Ssh(t) => t.host.clone(),
393 bamboo_config::cluster_fabric::NodePlacement::Local => "local".to_string(),
394 }
395}
396
397fn endpoint_looks_public(endpoint: &str) -> bool {
408 if endpoint.starts_with("wss://") {
409 return true;
410 }
411 let host = endpoint
412 .strip_prefix("ws://")
413 .unwrap_or(endpoint)
414 .split(['/', ':'])
415 .next()
416 .unwrap_or("");
417 !(host == "localhost" || host == "127.0.0.1" || host == "::1" || host.is_empty())
418}
419
420fn resolve_remote_placements(
421 placements: &[bamboo_config::RemoteActorPlacement],
422 nodes: &[bamboo_config::cluster_fabric::Node],
423) -> std::collections::HashMap<String, super::actor_adapter::ResolvedRemotePlacement> {
424 let mut out = std::collections::HashMap::new();
425 for p in placements {
426 let token = match p.token_env.as_deref() {
427 Some(env_var) => match std::env::var(env_var) {
428 Ok(token) => Some(token),
429 Err(_) => {
430 tracing::error!(
431 "remote placement for role '{}' token_env '{}' is not set; \
432 skipping (role falls back to local, NOT unauthenticated remote)",
433 p.role,
434 env_var
435 );
436 continue;
437 }
438 },
439 None => {
440 if endpoint_looks_public(&p.endpoint) {
444 tracing::warn!(
445 "remote placement for role '{}' has no token_env but targets a \
446 public-looking endpoint '{}'; work will be dispatched with NO bearer. \
447 Set token_env (and use wss://) for any non-loopback worker.",
448 p.role,
449 p.endpoint
450 );
451 }
452 None
453 }
454 };
455 out.insert(
456 p.role.clone(),
457 super::actor_adapter::ResolvedRemotePlacement {
458 endpoint: p.endpoint.clone(),
459 token,
460 ca_cert_file: p.ca_cert_file.as_ref().map(std::path::PathBuf::from),
461 host_label: node_label_for_endpoint(nodes, &p.endpoint),
464 },
465 );
466 }
467 out
468}
469
470pub fn extract_provider_credentials(
478 config: &Config,
479) -> Vec<bamboo_subagent::provision::ScopedCredential> {
480 let mut out = Vec::new();
481
482 let mut push_legacy = |name: &str, api_key: &str, base_url: Option<String>| {
486 let api_key = api_key.trim().to_string();
487 if api_key.is_empty() {
488 return;
489 }
490 out.push(bamboo_subagent::provision::ScopedCredential {
491 provider: name.to_string(),
492 api_key,
493 base_url,
494 provider_type: Some(name.to_string()),
495 });
496 };
497 if let Some(c) = &config.providers().openai {
498 push_legacy("openai", &c.api_key, c.base_url.clone());
499 }
500 if let Some(c) = &config.providers().anthropic {
501 push_legacy("anthropic", &c.api_key, c.base_url.clone());
502 }
503 if let Some(c) = &config.providers().gemini {
504 push_legacy("gemini", &c.api_key, c.base_url.clone());
505 }
506 if let Some(c) = &config.providers().bodhi {
507 push_legacy("bodhi", &c.api_key, c.base_url.clone());
508 }
509
510 out.extend(config.provider_instances.iter().filter_map(|(id, inst)| {
515 let api_key = inst.api_key.trim().to_string();
516 if api_key.is_empty() {
517 return None;
518 }
519 Some(bamboo_subagent::provision::ScopedCredential {
520 provider: id.clone(),
521 api_key,
522 base_url: inst.base_url.clone(),
523 provider_type: Some(inst.provider_type.clone()),
524 })
525 }));
526
527 out
528}
529
530#[cfg(test)]
531mod extract_provider_credentials_tests {
532 use super::extract_provider_credentials;
533 use bamboo_config::{
534 AnthropicConfig, BodhiConfig, Config, OpenAIConfig, ProviderInstanceConfig,
535 };
536
537 fn instance(provider_type: &str, api_key: &str) -> ProviderInstanceConfig {
538 ProviderInstanceConfig {
539 provider_type: provider_type.to_string(),
540 label: None,
541 api_key: api_key.to_string(),
542 api_key_encrypted: None,
543 credential_ref: None,
544 base_url: None,
545 model: None,
546 fast_model: None,
547 vision_model: None,
548 reasoning_effort: None,
549 responses_only_models: Vec::new(),
550 request_overrides: None,
551 enabled: true,
552 extra: Default::default(),
553 }
554 }
555
556 #[test]
557 fn no_config_yields_no_credentials() {
558 let config = Config::default();
559 assert!(extract_provider_credentials(&config).is_empty());
560 }
561
562 #[test]
567 fn legacy_only_config_yields_credential() {
568 let mut config = Config::default();
569 config.providers_mut().anthropic = Some(AnthropicConfig {
570 api_key: "sk-ant-legacy".to_string(),
571 base_url: Some("https://api.anthropic.com".to_string()),
572 ..Default::default()
573 });
574
575 let creds = extract_provider_credentials(&config);
576 assert_eq!(creds.len(), 1);
577 let c = &creds[0];
578 assert_eq!(c.provider, "anthropic");
579 assert_eq!(c.api_key, "sk-ant-legacy");
580 assert_eq!(c.base_url.as_deref(), Some("https://api.anthropic.com"));
581 assert_eq!(c.provider_type.as_deref(), Some("anthropic"));
582 }
583
584 #[test]
588 fn legacy_bodhi_config_yields_credential() {
589 let mut config = Config::default();
590 config.providers_mut().bodhi = Some(BodhiConfig {
591 api_key: "bhi_sk_legacy".to_string(),
592 api_key_encrypted: None,
593 credential_ref: None,
594 base_url: None,
595 target_provider: None,
596 reasoning_effort: None,
597 extra: Default::default(),
598 });
599
600 let creds = extract_provider_credentials(&config);
601 assert_eq!(creds.len(), 1);
602 assert_eq!(creds[0].provider, "bodhi");
603 assert_eq!(creds[0].api_key, "bhi_sk_legacy");
604 }
605
606 #[test]
609 fn legacy_config_with_empty_api_key_is_skipped() {
610 let mut config = Config::default();
611 config.providers_mut().openai = Some(OpenAIConfig::default());
612 assert!(extract_provider_credentials(&config).is_empty());
613 }
614
615 #[test]
618 fn legacy_and_instances_both_present_no_duplicates() {
619 let mut config = Config::default();
620 config.providers_mut().anthropic = Some(AnthropicConfig {
621 api_key: "sk-ant-legacy".to_string(),
622 ..Default::default()
623 });
624 config
625 .provider_instances
626 .insert("openai-work".to_string(), instance("openai", "sk-oai-work"));
627
628 let mut creds = extract_provider_credentials(&config);
629 creds.sort_by(|a, b| a.provider.cmp(&b.provider));
630
631 assert_eq!(creds.len(), 2);
632 assert_eq!(creds[0].provider, "anthropic");
633 assert_eq!(creds[0].api_key, "sk-ant-legacy");
634 assert_eq!(creds[1].provider, "openai-work");
635 assert_eq!(creds[1].api_key, "sk-oai-work");
636 assert_eq!(creds[1].provider_type.as_deref(), Some("openai"));
637 }
638}
639
640#[cfg(test)]
641mod placement_resolver_tests {
642 use super::{node_display_name, resolve_remote_placements, resolve_schedulable_placements};
643 use bamboo_config::cluster_fabric::{
644 DeployProfile, Node, NodePlacement, SshAuth, SshTarget, TrustLevel,
645 };
646 use bamboo_config::{RemoteActorPlacement, SchedulablePlacement};
647
648 fn ssh_node(id: &str, label: &str, host: &str, default_role: Option<&str>) -> Node {
649 Node {
650 id: id.into(),
651 label: label.into(),
652 placement: NodePlacement::Ssh(SshTarget {
653 host: host.into(),
654 port: 22,
655 username: "u".into(),
656 auth: SshAuth::SystemSshConfig,
657 host_key_fingerprint: None,
658 }),
659 trust_level: TrustLevel::default(),
660 deploy: DeployProfile {
661 default_role: default_role.map(String::from),
662 ..Default::default()
663 },
664 state: None,
665 enabled: true,
666 }
667 }
668
669 #[test]
670 fn node_display_name_prefers_label_then_ssh_host() {
671 let n = ssh_node("n1", "mini", "mini.local", None);
672 assert_eq!(node_display_name(&n), "mini");
673 let mut unlabeled = n.clone();
674 unlabeled.label = String::new();
675 assert_eq!(node_display_name(&unlabeled), "mini.local");
676 }
677
678 #[test]
679 fn schedulable_placement_takes_host_label_from_node_by_default_role() {
680 let nodes = vec![ssh_node(
681 "n1",
682 "mini",
683 "mini.local",
684 Some("mac-mini-monitor"),
685 )];
686 let placements = vec![SchedulablePlacement {
687 role: "mac-mini-monitor".into(),
688 pool: "mac-mini-monitor".into(),
689 ..Default::default()
690 }];
691 let out = resolve_schedulable_placements(&placements, &nodes);
692 let r = out.get("mac-mini-monitor").expect("role resolved");
693 assert_eq!(r.pool, "mac-mini-monitor");
694 assert_eq!(r.host_label.as_deref(), Some("mini"));
695 }
696
697 #[test]
698 fn remote_placement_takes_host_label_from_node_by_ssh_host() {
699 let nodes = vec![ssh_node("n1", "mini", "mini.local", None)];
700 let placements = vec![RemoteActorPlacement {
701 role: "explorer".into(),
702 endpoint: "ws://mini.local:8899".into(),
703 ..Default::default()
704 }];
705 let out = resolve_remote_placements(&placements, &nodes);
706 assert_eq!(
707 out.get("explorer").unwrap().host_label.as_deref(),
708 Some("mini")
709 );
710 }
711
712 #[test]
713 fn no_host_label_when_no_node_matches() {
714 let nodes = vec![ssh_node("n1", "mini", "mini.local", Some("other-role"))];
715 let sched = vec![SchedulablePlacement {
716 role: "x".into(),
717 pool: "unmatched".into(),
718 ..Default::default()
719 }];
720 assert_eq!(
721 resolve_schedulable_placements(&sched, &nodes)
722 .get("x")
723 .unwrap()
724 .host_label,
725 None
726 );
727 let remote = vec![RemoteActorPlacement {
728 role: "y".into(),
729 endpoint: "ws://other-host:9000".into(),
730 ..Default::default()
731 }];
732 assert_eq!(
733 resolve_remote_placements(&remote, &nodes)
734 .get("y")
735 .unwrap()
736 .host_label,
737 None
738 );
739 }
740}