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;
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 let agents = parse_external_agents(config);
85
86 let mut runners: Vec<Arc<dyn ExternalChildRunner>> = Vec::new();
87
88 match build_local_actor_runner(config, approval_registry.clone()) {
92 Ok(runner) => runners.push(runner),
93 Err(e) => tracing::error!("local actor sub-agent runner unavailable: {e}"),
94 }
95
96 for (_agent_id, profile) in agents {
97 if matches!(profile.protocol, ExternalAgentProtocol::Actor) {
99 let Some(worker_bin) = profile.worker_bin.as_ref() else {
100 tracing::error!(
101 "Actor agent profile {} has no worker_bin; skipping",
102 profile.agent_id
103 );
104 continue;
105 };
106 let fabric_dir = profile
110 .fabric_dir
111 .clone()
112 .map(std::path::PathBuf::from)
113 .unwrap_or_else(bamboo_config::paths::subagents_dir);
114 let executor = match profile.executor.as_deref() {
115 Some("echo") => bamboo_subagent::provision::ExecutorSpec::Echo,
116 Some("bamboo_runtime") | None => {
117 bamboo_subagent::provision::ExecutorSpec::BambooRuntime
118 }
119 Some("claude_code") => bamboo_subagent::provision::ExecutorSpec::ClaudeCode {
122 binary: profile.claude_code_binary.clone(),
123 model: profile.claude_code_model.clone(),
124 permission_mode: profile.claude_code_permission_mode.clone(),
125 inherit_user_config: profile.claude_code_inherit_user_config,
126 forward_env: profile.claude_code_forward_env.clone(),
127 },
128 Some(other) => {
129 tracing::error!(
130 "Actor agent profile {} has unknown executor '{}'; skipping",
131 profile.agent_id,
132 other
133 );
134 continue;
135 }
136 };
137 let mut runner = ActorChildRunner::new(
138 profile.agent_id.clone(),
139 std::path::PathBuf::from(worker_bin),
140 profile.worker_args.clone(),
141 fabric_dir,
142 executor,
143 extract_provider_credentials(config),
144 config.provider.clone(),
145 config
146 .subagents()
147 .max_concurrent
148 .unwrap_or(super::actor_adapter::DEFAULT_MAX_CONCURRENT_ACTORS),
149 );
150 if let Some(registry) = approval_registry.clone() {
151 runner = runner.with_approval_registry(registry);
152 }
153 runners.push(Arc::new(runner));
154 continue;
155 }
156
157 if !matches!(profile.protocol, ExternalAgentProtocol::A2aJsonRpc) {
158 tracing::warn!(
159 "External agent profile {} uses unsupported protocol {:?}",
160 profile.agent_id,
161 profile.protocol
162 );
163 continue;
164 }
165
166 let auth_token = match profile.auth_ref.as_ref() {
167 Some(ref_name) => match std::env::var(ref_name) {
168 Ok(token) => Some(token),
169 Err(_) => {
170 tracing::error!(
171 "External agent profile {} auth_ref env var {} is not set",
172 profile.agent_id,
173 ref_name
174 );
175 continue;
176 }
177 },
178 None => None,
179 };
180
181 let client_config = match A2AExternalChildRunner::build_client_config(&profile, auth_token)
182 {
183 Ok(cfg) => cfg,
184 Err(e) => {
185 tracing::error!(
186 "Failed to build A2A client config for profile {}: {}",
187 profile.agent_id,
188 e
189 );
190 continue;
191 }
192 };
193
194 let client = match A2AJsonRpcClient::new(client_config) {
195 Ok(c) => c,
196 Err(e) => {
197 tracing::error!(
198 "Failed to create A2A JSON-RPC client for profile {}: {}",
199 profile.agent_id,
200 e
201 );
202 continue;
203 }
204 };
205
206 runners.push(Arc::new(A2AExternalChildRunner::new(client, profile)));
207 }
208
209 Arc::new(CompositeExternalChildRunner::new(runners))
210}
211
212fn build_local_actor_runner(
217 config: &Config,
218 approval_registry: Option<super::approval_registry::SharedApprovalRegistry>,
219) -> Result<Arc<dyn ExternalChildRunner>, String> {
220 let sub = config.subagents();
221
222 let (worker_bin, worker_args) = match &sub.worker_bin {
223 Some(custom) => (
224 std::path::PathBuf::from(custom),
225 sub.worker_args.clone().unwrap_or_default(),
226 ),
227 None => (
228 std::env::current_exe().map_err(|e| format!("cannot locate own executable: {e}"))?,
229 sub.worker_args
230 .clone()
231 .unwrap_or_else(|| vec!["subagent-worker".to_string()]),
232 ),
233 };
234
235 let fabric_dir = sub
238 .fabric_dir
239 .clone()
240 .map(std::path::PathBuf::from)
241 .unwrap_or_else(bamboo_config::paths::subagents_dir);
242
243 let executor = match sub.executor.as_deref() {
244 Some("echo") => bamboo_subagent::provision::ExecutorSpec::Echo,
245 Some("bamboo_runtime") | None => bamboo_subagent::provision::ExecutorSpec::BambooRuntime,
246 Some("claude_code") => bamboo_subagent::provision::ExecutorSpec::ClaudeCode {
250 binary: sub.claude_code_binary.clone(),
251 model: sub.claude_code_model.clone(),
252 permission_mode: sub.claude_code_permission_mode.clone(),
253 inherit_user_config: sub.claude_code_inherit_user_config,
254 forward_env: sub.claude_code_forward_env.clone(),
255 },
256 Some(other) => return Err(format!("unknown subagents.executor '{other}'")),
257 };
258
259 let mut runner = ActorChildRunner::new(
260 super::config::LOCAL_ACTOR_AGENT_ID.to_string(),
261 worker_bin,
262 worker_args,
263 fabric_dir,
264 executor,
265 extract_provider_credentials(config),
266 config.provider.clone(),
267 sub.max_concurrent
268 .unwrap_or(super::actor_adapter::DEFAULT_MAX_CONCURRENT_ACTORS),
269 )
270 .with_remote_placements(resolve_remote_placements(
271 &sub.remote_placements,
272 &config.cluster_fabric.nodes,
273 ))
274 .with_schedulable_placements(resolve_schedulable_placements(
275 &sub.schedulable_placements,
276 &config.cluster_fabric.nodes,
277 ))
278 .with_bus(sub.broker.as_ref().map(|b| bamboo_subagent::BusEndpoint {
279 endpoint: b.endpoint.clone(),
280 token: b.token.clone(),
281 }));
282 if let Some(registry) = approval_registry {
283 runner = runner.with_approval_registry(registry);
284 }
285 Ok(Arc::new(runner))
286}
287
288fn resolve_schedulable_placements(
297 placements: &[bamboo_config::SchedulablePlacement],
298 nodes: &[bamboo_config::cluster_fabric::Node],
299) -> std::collections::HashMap<String, super::actor_adapter::ResolvedSchedulablePlacement> {
300 placements
303 .iter()
304 .map(|p| {
305 (
306 p.role.clone(),
307 super::actor_adapter::ResolvedSchedulablePlacement {
308 pool: p.pool.clone(),
309 host_label: node_label_for_role(nodes, &p.pool),
312 },
313 )
314 })
315 .collect()
316}
317
318fn node_label_for_role(
322 nodes: &[bamboo_config::cluster_fabric::Node],
323 role: &str,
324) -> Option<String> {
325 nodes
326 .iter()
327 .find(|n| n.deploy.default_role.as_deref() == Some(role))
328 .map(node_display_name)
329}
330
331fn node_label_for_endpoint(
335 nodes: &[bamboo_config::cluster_fabric::Node],
336 endpoint: &str,
337) -> Option<String> {
338 let host = endpoint
339 .trim()
340 .trim_start_matches("wss://")
341 .trim_start_matches("ws://")
342 .split(['/', ':'])
343 .next()
344 .unwrap_or("");
345 if host.is_empty() {
346 return None;
347 }
348 nodes
349 .iter()
350 .find(|n| match &n.placement {
351 bamboo_config::cluster_fabric::NodePlacement::Ssh(t) => t.host == host,
352 bamboo_config::cluster_fabric::NodePlacement::Local => false,
353 })
354 .map(node_display_name)
355}
356
357fn node_display_name(n: &bamboo_config::cluster_fabric::Node) -> String {
358 if !n.label.trim().is_empty() {
359 return n.label.clone();
360 }
361 match &n.placement {
362 bamboo_config::cluster_fabric::NodePlacement::Ssh(t) => t.host.clone(),
363 bamboo_config::cluster_fabric::NodePlacement::Local => "local".to_string(),
364 }
365}
366
367fn endpoint_looks_public(endpoint: &str) -> bool {
378 if endpoint.starts_with("wss://") {
379 return true;
380 }
381 let host = endpoint
382 .strip_prefix("ws://")
383 .unwrap_or(endpoint)
384 .split(['/', ':'])
385 .next()
386 .unwrap_or("");
387 !(host == "localhost" || host == "127.0.0.1" || host == "::1" || host.is_empty())
388}
389
390fn resolve_remote_placements(
391 placements: &[bamboo_config::RemoteActorPlacement],
392 nodes: &[bamboo_config::cluster_fabric::Node],
393) -> std::collections::HashMap<String, super::actor_adapter::ResolvedRemotePlacement> {
394 let mut out = std::collections::HashMap::new();
395 for p in placements {
396 let token = match p.token_env.as_deref() {
397 Some(env_var) => match std::env::var(env_var) {
398 Ok(token) => Some(token),
399 Err(_) => {
400 tracing::error!(
401 "remote placement for role '{}' token_env '{}' is not set; \
402 skipping (role falls back to local, NOT unauthenticated remote)",
403 p.role,
404 env_var
405 );
406 continue;
407 }
408 },
409 None => {
410 if endpoint_looks_public(&p.endpoint) {
414 tracing::warn!(
415 "remote placement for role '{}' has no token_env but targets a \
416 public-looking endpoint '{}'; work will be dispatched with NO bearer. \
417 Set token_env (and use wss://) for any non-loopback worker.",
418 p.role,
419 p.endpoint
420 );
421 }
422 None
423 }
424 };
425 out.insert(
426 p.role.clone(),
427 super::actor_adapter::ResolvedRemotePlacement {
428 endpoint: p.endpoint.clone(),
429 token,
430 ca_cert_file: p.ca_cert_file.as_ref().map(std::path::PathBuf::from),
431 host_label: node_label_for_endpoint(nodes, &p.endpoint),
434 },
435 );
436 }
437 out
438}
439
440pub fn extract_provider_credentials(
448 config: &Config,
449) -> Vec<bamboo_subagent::provision::ScopedCredential> {
450 let mut out = Vec::new();
451
452 let mut push_legacy = |name: &str, api_key: &str, base_url: Option<String>| {
456 let api_key = api_key.trim().to_string();
457 if api_key.is_empty() {
458 return;
459 }
460 out.push(bamboo_subagent::provision::ScopedCredential {
461 provider: name.to_string(),
462 api_key,
463 base_url,
464 provider_type: Some(name.to_string()),
465 });
466 };
467 if let Some(c) = &config.providers().openai {
468 push_legacy("openai", &c.api_key, c.base_url.clone());
469 }
470 if let Some(c) = &config.providers().anthropic {
471 push_legacy("anthropic", &c.api_key, c.base_url.clone());
472 }
473 if let Some(c) = &config.providers().gemini {
474 push_legacy("gemini", &c.api_key, c.base_url.clone());
475 }
476 if let Some(c) = &config.providers().bodhi {
477 push_legacy("bodhi", &c.api_key, c.base_url.clone());
478 }
479
480 out.extend(config.provider_instances.iter().filter_map(|(id, inst)| {
485 let api_key = inst.api_key.trim().to_string();
486 if api_key.is_empty() {
487 return None;
488 }
489 Some(bamboo_subagent::provision::ScopedCredential {
490 provider: id.clone(),
491 api_key,
492 base_url: inst.base_url.clone(),
493 provider_type: Some(inst.provider_type.clone()),
494 })
495 }));
496
497 out
498}
499
500#[cfg(test)]
501mod extract_provider_credentials_tests {
502 use super::extract_provider_credentials;
503 use bamboo_config::{
504 AnthropicConfig, BodhiConfig, Config, OpenAIConfig, ProviderInstanceConfig,
505 };
506
507 fn instance(provider_type: &str, api_key: &str) -> ProviderInstanceConfig {
508 ProviderInstanceConfig {
509 provider_type: provider_type.to_string(),
510 label: None,
511 api_key: api_key.to_string(),
512 api_key_encrypted: None,
513 base_url: None,
514 model: None,
515 fast_model: None,
516 vision_model: None,
517 reasoning_effort: None,
518 responses_only_models: Vec::new(),
519 request_overrides: None,
520 enabled: true,
521 extra: Default::default(),
522 }
523 }
524
525 #[test]
526 fn no_config_yields_no_credentials() {
527 let config = Config::default();
528 assert!(extract_provider_credentials(&config).is_empty());
529 }
530
531 #[test]
536 fn legacy_only_config_yields_credential() {
537 let mut config = Config::default();
538 config.providers_mut().anthropic = Some(AnthropicConfig {
539 api_key: "sk-ant-legacy".to_string(),
540 base_url: Some("https://api.anthropic.com".to_string()),
541 ..Default::default()
542 });
543
544 let creds = extract_provider_credentials(&config);
545 assert_eq!(creds.len(), 1);
546 let c = &creds[0];
547 assert_eq!(c.provider, "anthropic");
548 assert_eq!(c.api_key, "sk-ant-legacy");
549 assert_eq!(c.base_url.as_deref(), Some("https://api.anthropic.com"));
550 assert_eq!(c.provider_type.as_deref(), Some("anthropic"));
551 }
552
553 #[test]
557 fn legacy_bodhi_config_yields_credential() {
558 let mut config = Config::default();
559 config.providers_mut().bodhi = Some(BodhiConfig {
560 api_key: "bhi_sk_legacy".to_string(),
561 api_key_encrypted: None,
562 base_url: None,
563 target_provider: None,
564 reasoning_effort: None,
565 extra: Default::default(),
566 });
567
568 let creds = extract_provider_credentials(&config);
569 assert_eq!(creds.len(), 1);
570 assert_eq!(creds[0].provider, "bodhi");
571 assert_eq!(creds[0].api_key, "bhi_sk_legacy");
572 }
573
574 #[test]
577 fn legacy_config_with_empty_api_key_is_skipped() {
578 let mut config = Config::default();
579 config.providers_mut().openai = Some(OpenAIConfig::default());
580 assert!(extract_provider_credentials(&config).is_empty());
581 }
582
583 #[test]
586 fn legacy_and_instances_both_present_no_duplicates() {
587 let mut config = Config::default();
588 config.providers_mut().anthropic = Some(AnthropicConfig {
589 api_key: "sk-ant-legacy".to_string(),
590 ..Default::default()
591 });
592 config
593 .provider_instances
594 .insert("openai-work".to_string(), instance("openai", "sk-oai-work"));
595
596 let mut creds = extract_provider_credentials(&config);
597 creds.sort_by(|a, b| a.provider.cmp(&b.provider));
598
599 assert_eq!(creds.len(), 2);
600 assert_eq!(creds[0].provider, "anthropic");
601 assert_eq!(creds[0].api_key, "sk-ant-legacy");
602 assert_eq!(creds[1].provider, "openai-work");
603 assert_eq!(creds[1].api_key, "sk-oai-work");
604 assert_eq!(creds[1].provider_type.as_deref(), Some("openai"));
605 }
606}
607
608#[cfg(test)]
609mod placement_resolver_tests {
610 use super::{node_display_name, resolve_remote_placements, resolve_schedulable_placements};
611 use bamboo_config::cluster_fabric::{
612 DeployProfile, Node, NodePlacement, SshAuth, SshTarget, TrustLevel,
613 };
614 use bamboo_config::{RemoteActorPlacement, SchedulablePlacement};
615
616 fn ssh_node(id: &str, label: &str, host: &str, default_role: Option<&str>) -> Node {
617 Node {
618 id: id.into(),
619 label: label.into(),
620 placement: NodePlacement::Ssh(SshTarget {
621 host: host.into(),
622 port: 22,
623 username: "u".into(),
624 auth: SshAuth::SystemSshConfig,
625 host_key_fingerprint: None,
626 }),
627 trust_level: TrustLevel::default(),
628 deploy: DeployProfile {
629 default_role: default_role.map(String::from),
630 ..Default::default()
631 },
632 state: None,
633 enabled: true,
634 }
635 }
636
637 #[test]
638 fn node_display_name_prefers_label_then_ssh_host() {
639 let n = ssh_node("n1", "mini", "mini.local", None);
640 assert_eq!(node_display_name(&n), "mini");
641 let mut unlabeled = n.clone();
642 unlabeled.label = String::new();
643 assert_eq!(node_display_name(&unlabeled), "mini.local");
644 }
645
646 #[test]
647 fn schedulable_placement_takes_host_label_from_node_by_default_role() {
648 let nodes = vec![ssh_node(
649 "n1",
650 "mini",
651 "mini.local",
652 Some("mac-mini-monitor"),
653 )];
654 let placements = vec![SchedulablePlacement {
655 role: "mac-mini-monitor".into(),
656 pool: "mac-mini-monitor".into(),
657 ..Default::default()
658 }];
659 let out = resolve_schedulable_placements(&placements, &nodes);
660 let r = out.get("mac-mini-monitor").expect("role resolved");
661 assert_eq!(r.pool, "mac-mini-monitor");
662 assert_eq!(r.host_label.as_deref(), Some("mini"));
663 }
664
665 #[test]
666 fn remote_placement_takes_host_label_from_node_by_ssh_host() {
667 let nodes = vec![ssh_node("n1", "mini", "mini.local", None)];
668 let placements = vec![RemoteActorPlacement {
669 role: "explorer".into(),
670 endpoint: "ws://mini.local:8899".into(),
671 ..Default::default()
672 }];
673 let out = resolve_remote_placements(&placements, &nodes);
674 assert_eq!(
675 out.get("explorer").unwrap().host_label.as_deref(),
676 Some("mini")
677 );
678 }
679
680 #[test]
681 fn no_host_label_when_no_node_matches() {
682 let nodes = vec![ssh_node("n1", "mini", "mini.local", Some("other-role"))];
683 let sched = vec![SchedulablePlacement {
684 role: "x".into(),
685 pool: "unmatched".into(),
686 ..Default::default()
687 }];
688 assert_eq!(
689 resolve_schedulable_placements(&sched, &nodes)
690 .get("x")
691 .unwrap()
692 .host_label,
693 None
694 );
695 let remote = vec![RemoteActorPlacement {
696 role: "y".into(),
697 endpoint: "ws://other-host:9000".into(),
698 ..Default::default()
699 }];
700 assert_eq!(
701 resolve_remote_placements(&remote, &nodes)
702 .get("y")
703 .unwrap()
704 .host_label,
705 None
706 );
707 }
708}