1use std::sync::Arc;
2
3use crate::runtime::execution::{ExternalChildRunner, SessionInboxRuntimeBinding, 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, CodexRunTokenAuthority};
13use super::config::{parse_external_agents, ExternalAgentProtocol};
14
15#[async_trait]
18pub trait NativeToolCeilingSource: Send + Sync {
19 async fn observe(&self, session: &bamboo_domain::Session) -> Result<Vec<String>, String>;
20}
21
22pub fn uses_legacy_default_child_runtime(config: &Config) -> bool {
26 let subagents = config.subagents();
27 subagents.worker_bin.is_some()
28 || subagents.worker_args.is_some()
29 || !matches!(subagents.executor.as_deref(), None | Some("bamboo_runtime"))
30}
31
32pub fn requires_legacy_child_runtime(config: &Config) -> bool {
36 let subagents = config.subagents();
37 uses_legacy_default_child_runtime(config)
38 || !subagents.remote_placements.is_empty()
39 || !subagents.schedulable_placements.is_empty()
40 || !config.cluster_fabric.nodes.is_empty()
41 || parse_external_agents(config)
42 .values()
43 .any(|profile| matches!(profile.protocol, ExternalAgentProtocol::Actor))
44}
45
46fn codex_auth_mode_name(mode: bamboo_config::CodexAuthMode) -> String {
47 match mode {
48 bamboo_config::CodexAuthMode::Inherit => "inherit",
49 bamboo_config::CodexAuthMode::ApiKey => "api_key",
50 bamboo_config::CodexAuthMode::Custom => "custom",
51 bamboo_config::CodexAuthMode::Bamboo => "bamboo",
52 }
53 .to_string()
54}
55
56fn codex_wire_api_name(wire_api: bamboo_config::CodexWireApi) -> String {
57 match wire_api {
58 bamboo_config::CodexWireApi::Responses => "responses",
59 }
60 .to_string()
61}
62
63fn codex_mode_name(mode: bamboo_config::CodexMode) -> String {
64 match mode {
65 bamboo_config::CodexMode::Exec => "exec",
66 bamboo_config::CodexMode::AppServer => "app_server",
67 }
68 .to_string()
69}
70
71fn codex_sandbox_name(sandbox: bamboo_config::CodexSandbox) -> String {
72 match sandbox {
73 bamboo_config::CodexSandbox::ReadOnly => "read-only",
74 bamboo_config::CodexSandbox::WorkspaceWrite => "workspace-write",
75 bamboo_config::CodexSandbox::DangerFullAccess => "danger-full-access",
76 }
77 .to_string()
78}
79
80fn codex_approval_policy_name(policy: bamboo_config::CodexApprovalPolicy) -> String {
81 match policy {
82 bamboo_config::CodexApprovalPolicy::Never => "never",
83 bamboo_config::CodexApprovalPolicy::OnFailure => "on-failure",
84 bamboo_config::CodexApprovalPolicy::OnRequest => "on-request",
85 }
86 .to_string()
87}
88
89fn codex_base_url(
90 config: &Config,
91 mode: bamboo_config::CodexAuthMode,
92 custom: Option<String>,
93) -> Option<String> {
94 match mode {
95 bamboo_config::CodexAuthMode::Custom => custom,
96 bamboo_config::CodexAuthMode::Bamboo => {
97 let scheme = if config.server.tls.is_some() {
98 "https"
99 } else {
100 "http"
101 };
102 Some(format!(
103 "{scheme}://127.0.0.1:{}/openai/v1",
104 config.server.port
105 ))
106 }
107 bamboo_config::CodexAuthMode::Inherit | bamboo_config::CodexAuthMode::ApiKey => None,
108 }
109}
110
111pub struct CompositeExternalChildRunner {
113 runners: Vec<Arc<dyn ExternalChildRunner>>,
114 store: std::sync::RwLock<Option<Arc<bamboo_storage::SessionStoreV2>>>,
115}
116
117impl CompositeExternalChildRunner {
118 pub fn new(runners: Vec<Arc<dyn ExternalChildRunner>>) -> Self {
119 Self {
120 runners,
121 store: Default::default(),
122 }
123 }
124
125 async fn check_observer_route(&self, session: &bamboo_domain::Session) -> Result<(), String> {
126 use bamboo_domain::Storage;
127 let store = self
128 .store
129 .read()
130 .map_err(|_| "Child store lock is poisoned")?
131 .clone();
132 if let Some(store) = store {
133 if store
134 .supervisor_observer_activation(session)
135 .await
136 .map_err(|error| error.to_string())?
137 && !self.supports_native_ticket_budget()
138 {
139 return Err(
140 "registered Observers require the supported native-only backend".into(),
141 );
142 }
143 }
144 Ok(())
145 }
146}
147
148#[async_trait]
149impl ExternalChildRunner for CompositeExternalChildRunner {
150 fn supports_native_ticket_budget(&self) -> bool {
151 self.runners.len() == 1 && self.runners[0].supports_native_ticket_budget()
154 }
155
156 fn set_local_child_skill_factory(
157 &self,
158 factory: Option<Arc<dyn super::local_task::LocalChildSkillFactory>>,
159 ) {
160 for runner in &self.runners {
161 runner.set_local_child_skill_factory(factory.clone());
162 }
163 }
164
165 fn set_local_task_runtime(&self, runtime: Option<super::local_task::LocalTaskRuntime>) {
166 for runner in &self.runners {
167 runner.set_local_task_runtime(runtime.clone());
168 }
169 }
170
171 async fn prepare_child_execution(
172 &self,
173 session: &bamboo_agent_core::Session,
174 run_id: &str,
175 cancel: CancellationToken,
176 ) -> Result<Option<Arc<dyn bamboo_domain::RuntimeSessionPersistence>>, String> {
177 self.check_observer_route(session).await?;
178 for runner in &self.runners {
179 if runner.should_handle(session).await {
180 return runner
181 .prepare_child_execution(session, run_id, cancel)
182 .await;
183 }
184 }
185 Err("No child execution backend matched the canonical Session".into())
186 }
187
188 async fn abandon_child_execution(
189 &self,
190 session: &bamboo_agent_core::Session,
191 run_id: &str,
192 ) -> Result<(), String> {
193 for runner in &self.runners {
194 if runner.should_handle(session).await {
195 return runner.abandon_child_execution(session, run_id).await;
196 }
197 }
198 Err("Child execution backend disappeared before rollback".into())
199 }
200
201 fn quarantine_child_execution(&self, session: &bamboo_agent_core::Session, run_id: &str) {
202 for runner in &self.runners {
203 runner.quarantine_child_execution(session, run_id);
204 }
205 }
206
207 fn track_child_task(
208 &self,
209 session_id: &str,
210 run_id: &str,
211 mut handle: tokio::task::JoinHandle<()>,
212 ) -> Option<tokio::task::JoinHandle<()>> {
213 for runner in &self.runners {
214 handle = runner.track_child_task(session_id, run_id, handle)?;
215 }
216 Some(handle)
217 }
218
219 async fn finish_child_execution(
220 &self,
221 session: &bamboo_agent_core::Session,
222 run_id: &str,
223 history_committed: bool,
224 ) -> Result<(), String> {
225 for runner in &self.runners {
226 if runner.should_handle(session).await {
227 return runner
228 .finish_child_execution(session, run_id, history_committed)
229 .await;
230 }
231 }
232 Err("Child execution backend disappeared before finalization".into())
233 }
234
235 async fn shutdown(&self) -> Result<(), String> {
236 let mut failures = Vec::new();
237 for runner in &self.runners {
238 if let Err(error) = runner.shutdown().await {
239 failures.push(error);
240 }
241 }
242 if failures.is_empty() {
243 Ok(())
244 } else {
245 Err(failures.join("; "))
246 }
247 }
248
249 fn set_ticket_service(&self, service: Option<Arc<bamboo_tickets::TicketService>>) {
250 for runner in &self.runners {
251 runner.set_ticket_service(service.clone());
252 }
253 }
254
255 async fn validate_required_child_context_route(
256 &self,
257 session: &bamboo_agent_core::Session,
258 ) -> Result<(), String> {
259 for runner in &self.runners {
260 if runner.should_handle(session).await {
261 return runner.validate_required_child_context_route(session).await;
262 }
263 }
264 Err("required_child_context_unsupported: no matching registered worker route".into())
265 }
266 async fn should_handle(&self, session: &bamboo_agent_core::Session) -> bool {
267 for runner in &self.runners {
268 if runner.should_handle(session).await {
269 return true;
270 }
271 }
272 false
273 }
274
275 async fn execute_external_child(
276 &self,
277 session: &mut bamboo_agent_core::Session,
278 job: &SpawnJob,
279 event_tx: mpsc::Sender<AgentEvent>,
280 cancel_token: CancellationToken,
281 ) -> crate::runtime::runner::Result<()> {
282 self.check_observer_route(session)
283 .await
284 .map_err(AgentError::LLM)?;
285 for runner in &self.runners {
286 if runner.should_handle(session).await {
287 return runner
288 .execute_external_child(session, job, event_tx, cancel_token)
289 .await;
290 }
291 }
292 Err(AgentError::LLM(
293 "No matching external child runner found for session metadata".to_string(),
294 ))
295 }
296
297 async fn commit_durable_child_delivery(
298 &self,
299 session: &bamboo_agent_core::Session,
300 activation_run_id: &str,
301 ) -> Result<(), String> {
302 for runner in &self.runners {
303 runner
304 .commit_durable_child_delivery(session, activation_run_id)
305 .await?;
306 }
307 Ok(())
308 }
309
310 async fn confirm_durable_child_delivery(
311 &self,
312 session: &bamboo_agent_core::Session,
313 activation_run_id: &str,
314 history_committed: bool,
315 ) -> Result<(), String> {
316 for runner in &self.runners {
317 runner
318 .confirm_durable_child_delivery(session, activation_run_id, history_committed)
319 .await?;
320 }
321 Ok(())
322 }
323
324 async fn prepare_durable_child_delivery(
325 &self,
326 session: &bamboo_agent_core::Session,
327 activation_run_id: &str,
328 ) -> Result<bool, String> {
329 let mut prepared = false;
330 for runner in &self.runners {
331 prepared |= runner
332 .prepare_durable_child_delivery(session, activation_run_id)
333 .await?;
334 }
335 Ok(prepared)
336 }
337
338 fn set_escalation_bridge(&self, bridge: Option<bamboo_subagent::executor::HostBridge>) {
343 for runner in &self.runners {
344 runner.set_escalation_bridge(bridge.clone());
345 }
346 }
347
348 fn set_session_inbox_runtime(&self, binding: Option<SessionInboxRuntimeBinding>) {
349 for runner in &self.runners {
350 runner.set_session_inbox_runtime(binding.clone());
351 }
352 }
353
354 fn set_actor_directory_store(&self, store: Option<Arc<bamboo_storage::SessionStoreV2>>) {
355 if let Ok(mut current) = self.store.write() {
356 *current = store.clone();
357 }
358 for runner in &self.runners {
359 runner.set_actor_directory_store(store.clone());
360 }
361 }
362
363 fn set_canonical_subagent_tool(&self, tool: Option<Arc<dyn bamboo_agent_core::tools::Tool>>) {
364 for runner in &self.runners {
365 runner.set_canonical_subagent_tool(tool.clone());
366 }
367 }
368
369 fn set_actor_event_observer(
370 &self,
371 observer: Option<Arc<dyn super::actor_event_stream::ActorEventObserver>>,
372 ) {
373 for runner in &self.runners {
374 runner.set_actor_event_observer(observer.clone());
375 }
376 }
377}
378
379pub fn build_external_child_runner(config: &Config) -> Arc<dyn ExternalChildRunner> {
388 build_external_child_runner_with_registry(config, None)
389}
390
391pub fn build_external_child_runner_with_registry(
393 config: &Config,
394 approval_registry: Option<super::approval_registry::SharedApprovalRegistry>,
395) -> Arc<dyn ExternalChildRunner> {
396 build_external_child_runner_with_registry_and_reviewer(config, approval_registry, None, None)
397}
398
399pub fn build_external_child_runner_with_registry_and_reviewer(
402 config: &Config,
403 approval_registry: Option<super::approval_registry::SharedApprovalRegistry>,
404 approval_reviewer: Option<Arc<dyn ChildApprovalReviewer>>,
405 permission_config: Option<Arc<bamboo_tools::permission::PermissionConfig>>,
406) -> Arc<dyn ExternalChildRunner> {
407 build_external_child_runner_with_codex_tokens(
408 config,
409 approval_registry,
410 approval_reviewer,
411 permission_config,
412 None,
413 )
414}
415
416pub fn build_legacy_worker_child_runner(
420 config: &Config,
421 approval_registry: Option<super::approval_registry::SharedApprovalRegistry>,
422 approval_reviewer: Option<Arc<dyn ChildApprovalReviewer>>,
423 permission_config: Option<Arc<bamboo_tools::permission::PermissionConfig>>,
424) -> Arc<dyn ExternalChildRunner> {
425 let mut config = config.clone();
426 let subagents = config.subagents_mut();
427 if subagents.worker_bin.is_none() && subagents.worker_args.is_none() {
428 subagents.worker_args = Some(vec!["subagent-worker".into()]);
430 }
431 build_external_child_runner_with_registry_and_reviewer(
432 &config,
433 approval_registry,
434 approval_reviewer,
435 permission_config,
436 )
437}
438
439pub fn build_external_child_runner_with_codex_tokens(
442 config: &Config,
443 approval_registry: Option<super::approval_registry::SharedApprovalRegistry>,
444 approval_reviewer: Option<Arc<dyn ChildApprovalReviewer>>,
445 permission_config: Option<Arc<bamboo_tools::permission::PermissionConfig>>,
446 codex_run_tokens: Option<Arc<dyn CodexRunTokenAuthority>>,
447) -> Arc<dyn ExternalChildRunner> {
448 build_external_child_runner_internal(
449 config,
450 None,
451 approval_registry,
452 approval_reviewer,
453 permission_config,
454 codex_run_tokens,
455 None,
456 )
457}
458
459pub fn build_external_child_runner_with_live_config_and_codex_tokens(
463 config: &Config,
464 live_provider_config: Arc<tokio::sync::RwLock<Config>>,
465 approval_registry: Option<super::approval_registry::SharedApprovalRegistry>,
466 approval_reviewer: Option<Arc<dyn ChildApprovalReviewer>>,
467 permission_config: Option<Arc<bamboo_tools::permission::PermissionConfig>>,
468 codex_run_tokens: Option<Arc<dyn CodexRunTokenAuthority>>,
469) -> Arc<dyn ExternalChildRunner> {
470 build_external_child_runner_internal(
471 config,
472 Some(live_provider_config),
473 approval_registry,
474 approval_reviewer,
475 permission_config,
476 codex_run_tokens,
477 None,
478 )
479}
480
481#[allow(clippy::too_many_arguments)]
484pub fn build_external_child_runner_with_native_tool_ceiling(
485 config: &Config,
486 live_provider_config: Arc<tokio::sync::RwLock<Config>>,
487 approval_registry: Option<super::approval_registry::SharedApprovalRegistry>,
488 approval_reviewer: Option<Arc<dyn ChildApprovalReviewer>>,
489 permission_config: Option<Arc<bamboo_tools::permission::PermissionConfig>>,
490 codex_run_tokens: Option<Arc<dyn CodexRunTokenAuthority>>,
491 native_tool_ceiling: Arc<dyn NativeToolCeilingSource>,
492) -> Arc<dyn ExternalChildRunner> {
493 build_external_child_runner_internal(
494 config,
495 Some(live_provider_config),
496 approval_registry,
497 approval_reviewer,
498 permission_config,
499 codex_run_tokens,
500 Some(native_tool_ceiling),
501 )
502}
503
504fn build_external_child_runner_internal(
505 config: &Config,
506 live_provider_config: Option<Arc<tokio::sync::RwLock<Config>>>,
507 approval_registry: Option<super::approval_registry::SharedApprovalRegistry>,
508 approval_reviewer: Option<Arc<dyn ChildApprovalReviewer>>,
509 permission_config: Option<Arc<bamboo_tools::permission::PermissionConfig>>,
510 codex_run_tokens: Option<Arc<dyn CodexRunTokenAuthority>>,
511 native_tool_ceiling: Option<Arc<dyn NativeToolCeilingSource>>,
512) -> Arc<dyn ExternalChildRunner> {
513 let agents = parse_external_agents(config);
514
515 let mut runners: Vec<Arc<dyn ExternalChildRunner>> = Vec::new();
516
517 if uses_legacy_default_child_runtime(config) {
521 match build_local_actor_runner(
522 config,
523 approval_registry.clone(),
524 approval_reviewer.clone(),
525 permission_config.clone(),
526 codex_run_tokens.clone(),
527 live_provider_config.clone(),
528 native_tool_ceiling.clone(),
529 ) {
530 Ok(runner) => runners.push(runner),
531 Err(error) => tracing::error!("explicit legacy child runner unavailable: {error}"),
532 }
533 } else {
534 runners.push(Arc::new(
535 super::local_task::LocalTaskChildRunner::new(
536 config,
537 native_tool_ceiling.clone(),
538 permission_config.clone(),
539 )
540 .with_approval(approval_registry.clone(), approval_reviewer.clone()),
541 ));
542 if !config.subagents().remote_placements.is_empty()
543 || !config.subagents().schedulable_placements.is_empty()
544 {
545 match build_local_actor_runner(
546 config,
547 approval_registry.clone(),
548 approval_reviewer.clone(),
549 permission_config.clone(),
550 codex_run_tokens.clone(),
551 live_provider_config.clone(),
552 native_tool_ceiling.clone(),
553 ) {
554 Ok(runner) => runners.push(runner),
555 Err(error) => tracing::error!("explicit remote child route unavailable: {error}"),
556 }
557 }
558 }
559
560 for (_agent_id, profile) in agents {
561 if matches!(profile.protocol, ExternalAgentProtocol::Actor) {
563 let Some(worker_bin) = profile.worker_bin.as_ref() else {
564 tracing::error!(
565 "Actor agent profile {} has no worker_bin; skipping",
566 profile.agent_id
567 );
568 continue;
569 };
570 let fabric_dir = profile
574 .fabric_dir
575 .clone()
576 .map(std::path::PathBuf::from)
577 .unwrap_or_else(bamboo_config::paths::subagents_dir);
578 let executor = match profile.executor.as_deref() {
579 Some("echo") => bamboo_subagent::provision::ExecutorSpec::Echo,
580 Some("bamboo_runtime") | None => {
581 bamboo_subagent::provision::ExecutorSpec::BambooRuntime
582 }
583 Some("claude_code") => bamboo_subagent::provision::ExecutorSpec::ClaudeCode {
586 binary: profile.claude_code_binary.clone(),
587 model: profile.claude_code_model.clone(),
588 permission_mode: profile.claude_code_permission_mode.clone(),
589 inherit_user_config: profile.claude_code_inherit_user_config,
590 forward_env: profile.claude_code_forward_env.clone(),
591 },
592 Some("codex") => bamboo_subagent::provision::ExecutorSpec::Codex {
593 binary: profile.codex_binary.clone(),
594 model: profile.codex_model.clone(),
595 mode: profile.codex_mode.map(codex_mode_name),
596 sandbox: profile.codex_sandbox.map(codex_sandbox_name),
597 inherit_user_config: None,
598 auth_mode: Some(codex_auth_mode_name(
599 profile.codex_auth_mode.unwrap_or_default(),
600 )),
601 base_url: codex_base_url(
602 config,
603 profile.codex_auth_mode.unwrap_or_default(),
604 profile.codex_base_url.clone(),
605 ),
606 wire_api: profile.codex_wire_api.map(codex_wire_api_name),
607 provider_key_ref: profile
608 .codex_provider_key_ref
609 .as_ref()
610 .map(|reference| reference.as_str().to_string()),
611 forward_env: profile.codex_forward_env.clone(),
612 approval_policy: profile
613 .codex_approval_policy
614 .map(codex_approval_policy_name),
615 network_access: profile.codex_network_access,
616 allow_danger_bypass: profile.codex_allow_danger_bypass,
617 permission_profile: Some(profile.permission_profile.clone()),
618 workspace_owned: None,
619 },
620 Some(other) => {
621 tracing::error!(
622 "Actor agent profile {} has unknown executor '{}'; skipping",
623 profile.agent_id,
624 other
625 );
626 continue;
627 }
628 };
629 let mut runner = ActorChildRunner::new(
630 profile.agent_id.clone(),
631 std::path::PathBuf::from(worker_bin),
632 profile.worker_args.clone(),
633 fabric_dir,
634 executor,
635 extract_provider_credentials(config),
636 config.effective_default_provider().to_string(),
637 config
638 .subagents()
639 .max_concurrent
640 .unwrap_or(super::actor_adapter::DEFAULT_MAX_CONCURRENT_ACTORS),
641 )
642 .with_max_spawn_depth(config.subagents().effective_max_spawn_depth());
643 if let Some(registry) = approval_registry.clone() {
644 runner = runner.with_approval_registry(registry);
645 }
646 if let Some(reviewer) = approval_reviewer.clone() {
647 runner = runner.with_approval_reviewer(reviewer);
648 }
649 if let Some(config) = permission_config.clone() {
650 runner = runner.with_permission_config(config);
651 }
652 if let Some(config) = live_provider_config.clone() {
653 runner = runner.with_live_provider_config(config);
654 }
655 runner = runner.with_codex_run_tokens(codex_run_tokens.clone());
656 runners.push(Arc::new(runner));
657 continue;
658 }
659
660 if !matches!(profile.protocol, ExternalAgentProtocol::A2aJsonRpc) {
661 tracing::warn!(
662 "External agent profile {} uses unsupported protocol {:?}",
663 profile.agent_id,
664 profile.protocol
665 );
666 continue;
667 }
668
669 let auth_token = match profile.auth_ref.as_ref() {
670 Some(ref_name) => match std::env::var(ref_name) {
671 Ok(token) => Some(token),
672 Err(_) => {
673 tracing::error!(
674 "External agent profile {} auth_ref env var {} is not set",
675 profile.agent_id,
676 ref_name
677 );
678 continue;
679 }
680 },
681 None => None,
682 };
683
684 let client_config = match A2AExternalChildRunner::build_client_config(&profile, auth_token)
685 {
686 Ok(cfg) => cfg,
687 Err(e) => {
688 tracing::error!(
689 "Failed to build A2A client config for profile {}: {}",
690 profile.agent_id,
691 e
692 );
693 continue;
694 }
695 };
696
697 let client = match A2AJsonRpcClient::new(client_config) {
698 Ok(c) => c,
699 Err(e) => {
700 tracing::error!(
701 "Failed to create A2A JSON-RPC client for profile {}: {}",
702 profile.agent_id,
703 e
704 );
705 continue;
706 }
707 };
708
709 runners.push(Arc::new(A2AExternalChildRunner::new(client, profile)));
710 }
711
712 Arc::new(CompositeExternalChildRunner::new(runners))
713}
714
715fn builtin_local_worker_launch(worker_bin: &std::path::Path, worker_args: &[String]) -> bool {
717 if worker_args != ["subagent-worker".to_string()] {
718 return false;
719 }
720 let Ok(current) = std::env::current_exe().and_then(|path| path.canonicalize()) else {
725 return false;
726 };
727 worker_bin
728 .canonicalize()
729 .is_ok_and(|worker| worker == current)
730}
731
732fn build_local_actor_runner(
734 config: &Config,
735 approval_registry: Option<super::approval_registry::SharedApprovalRegistry>,
736 approval_reviewer: Option<Arc<dyn ChildApprovalReviewer>>,
737 permission_config: Option<Arc<bamboo_tools::permission::PermissionConfig>>,
738 codex_run_tokens: Option<Arc<dyn CodexRunTokenAuthority>>,
739 live_provider_config: Option<Arc<tokio::sync::RwLock<Config>>>,
740 native_tool_ceiling: Option<Arc<dyn NativeToolCeilingSource>>,
741) -> Result<Arc<dyn ExternalChildRunner>, String> {
742 let sub = config.subagents();
743
744 let (worker_bin, worker_args) = match &sub.worker_bin {
745 Some(custom) => (
746 std::path::PathBuf::from(custom),
747 sub.worker_args.clone().unwrap_or_default(),
748 ),
749 None => (
750 std::env::current_exe().map_err(|e| format!("cannot locate own executable: {e}"))?,
751 sub.worker_args
752 .clone()
753 .unwrap_or_else(|| vec!["subagent-worker".to_string()]),
754 ),
755 };
756
757 let fabric_dir = sub
760 .fabric_dir
761 .clone()
762 .map(std::path::PathBuf::from)
763 .unwrap_or_else(bamboo_config::paths::subagents_dir);
764
765 let executor = subagent_executor_spec(config)?;
766 let builtin_required_context_route = builtin_local_worker_launch(&worker_bin, &worker_args);
767
768 let mut runner = ActorChildRunner::new(
769 super::config::LOCAL_ACTOR_AGENT_ID.to_string(),
770 worker_bin,
771 worker_args,
772 fabric_dir,
773 executor,
774 extract_provider_credentials(config),
775 config.effective_default_provider().to_string(),
776 sub.max_concurrent
777 .unwrap_or(super::actor_adapter::DEFAULT_MAX_CONCURRENT_ACTORS),
778 )
779 .with_max_spawn_depth(sub.effective_max_spawn_depth())
780 .with_builtin_required_context_route(builtin_required_context_route)
781 .with_remote_placements(resolve_remote_placements(
782 &sub.remote_placements,
783 &config.cluster_fabric.nodes,
784 ))
785 .with_schedulable_placements(resolve_schedulable_placements(
786 &sub.schedulable_placements,
787 &config.cluster_fabric.nodes,
788 ))
789 .with_bus(sub.broker.as_ref().map(|b| bamboo_subagent::BusEndpoint {
790 endpoint: b.endpoint.clone(),
791 token: b.token.clone(),
792 }))
793 .with_codex_run_tokens(codex_run_tokens)
794 .with_native_tool_ceiling_source(native_tool_ceiling);
795 if let Some(registry) = approval_registry {
796 runner = runner.with_approval_registry(registry);
797 }
798 if let Some(reviewer) = approval_reviewer {
799 runner = runner.with_approval_reviewer(reviewer);
800 }
801 if let Some(config) = permission_config {
802 runner = runner.with_permission_config(config);
803 }
804 if let Some(config) = live_provider_config {
805 runner = runner.with_live_provider_config(config);
806 }
807 Ok(Arc::new(runner))
808}
809
810fn subagent_executor_spec(
814 config: &Config,
815) -> Result<bamboo_subagent::provision::ExecutorSpec, String> {
816 let sub = config.subagents();
817 Ok(match sub.executor.as_deref() {
818 Some("echo") => bamboo_subagent::provision::ExecutorSpec::Echo,
819 Some("bamboo_runtime") | None => bamboo_subagent::provision::ExecutorSpec::BambooRuntime,
820 Some("claude_code") => bamboo_subagent::provision::ExecutorSpec::ClaudeCode {
821 binary: sub.claude_code_binary.clone(),
822 model: sub.claude_code_model.clone(),
823 permission_mode: sub.claude_code_permission_mode.clone(),
824 inherit_user_config: sub.claude_code_inherit_user_config,
825 forward_env: sub.claude_code_forward_env.clone(),
826 },
827 Some("codex") => bamboo_subagent::provision::ExecutorSpec::Codex {
828 binary: sub.codex_binary.clone(),
829 model: sub.codex_model.clone(),
830 mode: sub.codex_mode.map(codex_mode_name),
831 sandbox: sub.codex_sandbox.map(codex_sandbox_name),
832 inherit_user_config: None,
833 auth_mode: Some(codex_auth_mode_name(
834 sub.codex_auth_mode.unwrap_or_default(),
835 )),
836 base_url: codex_base_url(
837 config,
838 sub.codex_auth_mode.unwrap_or_default(),
839 sub.codex_base_url.clone(),
840 ),
841 wire_api: sub.codex_wire_api.map(codex_wire_api_name),
842 provider_key_ref: sub
843 .codex_provider_key_ref
844 .as_ref()
845 .map(|reference| reference.as_str().to_string()),
846 forward_env: sub.codex_forward_env.clone(),
847 approval_policy: sub.codex_approval_policy.map(codex_approval_policy_name),
848 network_access: sub.codex_network_access,
849 allow_danger_bypass: sub.codex_allow_danger_bypass,
850 permission_profile: None,
851 workspace_owned: None,
852 },
853 Some(other) => return Err(format!("unknown subagents.executor '{other}'")),
854 })
855}
856
857fn valid_placement_identifier(value: &str) -> bool {
858 !value.is_empty()
859 && value.len() <= 128
860 && value
861 .bytes()
862 .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.'))
863}
864
865fn valid_placement_requirements(value: &bamboo_config::OperatorPlacementRequirements) -> bool {
866 valid_placement_identifier(&value.trust_zone)
867 && value
868 .workspace_label
869 .as_deref()
870 .is_none_or(valid_placement_identifier)
871 && value
872 .network_zone
873 .as_deref()
874 .is_none_or(valid_placement_identifier)
875 && value.required_tools.len() <= 256
876 && value
877 .required_tools
878 .iter()
879 .all(|tool| valid_placement_identifier(tool))
880}
881
882fn valid_scoped_broker_endpoint(value: &str) -> bool {
883 value.len() <= 2048
884 && url::Url::parse(value).is_ok_and(|url| {
885 url.scheme() == "wss"
886 && url.host_str().is_some()
887 && url.username().is_empty()
888 && url.password().is_none()
889 && url.query().is_none()
890 && url.fragment().is_none()
891 })
892}
893
894fn resolved_broker_token(name: Option<&str>) -> Option<String> {
895 name.filter(|name| !name.is_empty() && name.len() <= 256)
896 .and_then(|name| std::env::var(name).ok())
897 .filter(|token| {
898 (32..=256).contains(&token.len()) && token.bytes().all(|byte| byte.is_ascii_graphic())
899 })
900}
901
902fn resolve_schedulable_placements(
905 placements: &[bamboo_config::SchedulablePlacement],
906 nodes: &[bamboo_config::cluster_fabric::Node],
907) -> std::collections::HashMap<String, super::actor_adapter::ResolvedSchedulablePlacement> {
908 let mut out = std::collections::HashMap::new();
909 for p in placements {
910 let token = resolved_broker_token(p.token_env.as_deref());
911 let valid = valid_placement_identifier(&p.pool)
912 && valid_scoped_broker_endpoint(&p.registry_url)
913 && p.token_env.is_some()
914 && token.is_some()
915 && p.ca_cert_file.as_ref().is_some_and(|path| {
916 bamboo_broker::client_config_trusting_cert(std::path::Path::new(path)).is_ok()
917 })
918 && p.broker_parent.as_ref().is_some_and(|parent| {
919 valid_placement_identifier(&parent.parent_mailbox)
920 && valid_placement_identifier(&parent.parent_role)
921 })
922 && p.placement_requirements
923 .as_ref()
924 .is_some_and(valid_placement_requirements);
925 let resolved = super::actor_adapter::ResolvedSchedulablePlacement {
926 pool: p.pool.clone(),
927 host_label: node_label_for_role(nodes, &p.pool),
928 endpoint: p.registry_url.clone(),
929 token: if valid { token } else { None },
930 ca_cert_file: p.ca_cert_file.as_ref().map(std::path::PathBuf::from),
931 broker_parent: if valid { p.broker_parent.clone() } else { None },
932 requirements: if valid {
933 p.placement_requirements.clone()
934 } else {
935 None
936 },
937 };
938 if out.insert(p.role.clone(), resolved).is_some() {
939 if let Some(route) = out.get_mut(&p.role) {
941 route.broker_parent = None;
942 }
943 }
944 }
945 out
946}
947
948fn node_label_for_role(
952 nodes: &[bamboo_config::cluster_fabric::Node],
953 role: &str,
954) -> Option<String> {
955 nodes
956 .iter()
957 .find(|n| n.deploy.default_role.as_deref() == Some(role))
958 .map(node_display_name)
959}
960
961fn node_label_for_endpoint(
965 nodes: &[bamboo_config::cluster_fabric::Node],
966 endpoint: &str,
967) -> Option<String> {
968 let host = endpoint
969 .trim()
970 .trim_start_matches("wss://")
971 .trim_start_matches("ws://")
972 .split(['/', ':'])
973 .next()
974 .unwrap_or("");
975 if host.is_empty() {
976 return None;
977 }
978 nodes
979 .iter()
980 .find(|n| match &n.placement {
981 bamboo_config::cluster_fabric::NodePlacement::Ssh(t) => t.host == host,
982 bamboo_config::cluster_fabric::NodePlacement::Local => false,
983 })
984 .map(node_display_name)
985}
986
987fn node_display_name(n: &bamboo_config::cluster_fabric::Node) -> String {
988 if !n.label.trim().is_empty() {
989 return n.label.clone();
990 }
991 match &n.placement {
992 bamboo_config::cluster_fabric::NodePlacement::Ssh(t) => t.host.clone(),
993 bamboo_config::cluster_fabric::NodePlacement::Local => "local".to_string(),
994 }
995}
996
997fn resolve_remote_placements(
1001 placements: &[bamboo_config::RemoteActorPlacement],
1002 nodes: &[bamboo_config::cluster_fabric::Node],
1003) -> std::collections::HashMap<String, super::actor_adapter::ResolvedRemotePlacement> {
1004 let mut out = std::collections::HashMap::new();
1005 for p in placements {
1006 let duplicate_strict = placements
1007 .iter()
1008 .any(|other| other.role == p.role && other.broker_peer.is_some())
1009 && placements
1010 .iter()
1011 .filter(|other| other.role == p.role)
1012 .count()
1013 != 1;
1014 if duplicate_strict {
1015 out.insert(
1016 p.role.clone(),
1017 super::actor_adapter::ResolvedRemotePlacement {
1018 endpoint: String::new(),
1019 token: None,
1020 ca_cert_file: None,
1021 host_label: Some("remote".into()),
1022 broker_peer: Some(Err(())),
1023 requirements: None,
1024 },
1025 );
1026 continue;
1027 }
1028 if let Some(peer) = &p.broker_peer {
1029 let token = resolved_broker_token(p.token_env.as_deref());
1030 let valid = peer.valid()
1031 && valid_scoped_broker_endpoint(&p.endpoint)
1032 && p.token_env.is_some()
1033 && token.is_some()
1034 && p.ca_cert_file.as_ref().is_some_and(|path| {
1035 bamboo_broker::client_config_trusting_cert(std::path::Path::new(path)).is_ok()
1036 })
1037 && p.placement_requirements
1038 .as_ref()
1039 .is_some_and(valid_placement_requirements);
1040 out.insert(
1041 p.role.clone(),
1042 super::actor_adapter::ResolvedRemotePlacement {
1043 endpoint: p.endpoint.clone(),
1044 token: if valid { token } else { None },
1045 ca_cert_file: p.ca_cert_file.as_ref().map(std::path::PathBuf::from),
1046 host_label: Some("remote".into()),
1047 broker_peer: Some(if valid { Ok(peer.clone()) } else { Err(()) }),
1048 requirements: if valid {
1049 p.placement_requirements.clone()
1050 } else {
1051 None
1052 },
1053 },
1054 );
1055 continue; }
1057 out.insert(
1058 p.role.clone(),
1059 super::actor_adapter::ResolvedRemotePlacement {
1060 endpoint: p.endpoint.clone(),
1061 token: None,
1062 ca_cert_file: p.ca_cert_file.as_ref().map(std::path::PathBuf::from),
1063 host_label: node_label_for_endpoint(nodes, &p.endpoint),
1066 broker_peer: Some(Err(())),
1067 requirements: None,
1068 },
1069 );
1070 }
1071 out
1072}
1073
1074pub fn extract_provider_credentials(
1082 config: &Config,
1083) -> Vec<bamboo_subagent::provision::ScopedCredential> {
1084 let mut out = Vec::new();
1085
1086 fn push_instance(
1087 out: &mut Vec<bamboo_subagent::provision::ScopedCredential>,
1088 id: &str,
1089 instance: &bamboo_config::ProviderInstanceConfig,
1090 ) {
1091 if !instance.enabled {
1092 return;
1093 }
1094 let api_key = instance.api_key.trim().to_string();
1095 if api_key.is_empty() {
1096 return;
1097 }
1098 out.push(bamboo_subagent::provision::ScopedCredential {
1099 provider: id.to_string(),
1100 api_key,
1101 base_url: instance.base_url.clone(),
1102 provider_type: Some(instance.provider_type.clone()),
1103 credential_ref: instance
1104 .credential_ref
1105 .as_ref()
1106 .map(|reference| reference.as_str().to_string()),
1107 });
1108 }
1109
1110 if !config.provider_instances.is_empty() {
1111 for (id, instance) in &config.provider_instances {
1114 push_instance(&mut out, id, instance);
1115 }
1116
1117 let default_id = config.effective_default_provider();
1121 if !config.provider_instances.contains_key(default_id) {
1122 if let Some((_, instance)) = bamboo_config::synthesize_legacy_instances(config)
1123 .into_iter()
1124 .find(|(id, _)| id == default_id)
1125 {
1126 push_instance(&mut out, default_id, &instance);
1127 }
1128 }
1129 return out;
1130 }
1131
1132 let mut push_legacy =
1136 |name: &str, api_key: &str, base_url: Option<String>, credential_ref: Option<String>| {
1137 let api_key = api_key.trim().to_string();
1138 if api_key.is_empty() {
1139 return;
1140 }
1141 out.push(bamboo_subagent::provision::ScopedCredential {
1142 provider: name.to_string(),
1143 api_key,
1144 base_url,
1145 provider_type: Some(name.to_string()),
1146 credential_ref,
1147 });
1148 };
1149 if let Some(c) = &config.providers().openai {
1150 push_legacy(
1151 "openai",
1152 &c.api_key,
1153 c.base_url.clone(),
1154 c.credential_ref
1155 .as_ref()
1156 .map(|reference| reference.as_str().to_string()),
1157 );
1158 }
1159 if let Some(c) = &config.providers().anthropic {
1160 push_legacy(
1161 "anthropic",
1162 &c.api_key,
1163 c.base_url.clone(),
1164 c.credential_ref
1165 .as_ref()
1166 .map(|reference| reference.as_str().to_string()),
1167 );
1168 }
1169 if let Some(c) = &config.providers().gemini {
1170 push_legacy(
1171 "gemini",
1172 &c.api_key,
1173 c.base_url.clone(),
1174 c.credential_ref
1175 .as_ref()
1176 .map(|reference| reference.as_str().to_string()),
1177 );
1178 }
1179 if let Some(c) = &config.providers().bodhi {
1180 push_legacy(
1181 "bodhi",
1182 &c.api_key,
1183 c.base_url.clone(),
1184 c.credential_ref
1185 .as_ref()
1186 .map(|reference| reference.as_str().to_string()),
1187 );
1188 }
1189
1190 out
1191}
1192
1193#[cfg(test)]
1194mod codex_runtime_config_tests {
1195 #[test]
1196 fn builtin_local_worker_launch_requires_current_executable_and_exact_arguments() {
1197 let current = std::env::current_exe().unwrap();
1198 let exact = vec!["subagent-worker".to_string()];
1199 assert!(super::builtin_local_worker_launch(¤t, &exact));
1200 assert!(super::builtin_local_worker_launch(
1201 ¤t.canonicalize().unwrap(),
1202 &exact
1203 ));
1204 for args in [
1205 vec![],
1206 vec!["serve".into()],
1207 vec!["subagent-worker".into(), "--extra".into()],
1208 ] {
1209 assert!(!super::builtin_local_worker_launch(¤t, &args));
1210 }
1211 let temp = tempfile::tempdir().unwrap();
1212 assert!(!super::builtin_local_worker_launch(
1213 &temp.path().join("missing"),
1214 &exact
1215 ));
1216 let foreign = temp.path().join("foreign-worker");
1217 std::fs::write(
1218 &foreign,
1219 "claimed worker capabilities do not establish executable identity",
1220 )
1221 .unwrap();
1222 assert!(!super::builtin_local_worker_launch(&foreign, &exact));
1223 }
1224
1225 #[tokio::test]
1226 async fn required_context_uses_registered_launch_snapshot_after_live_config_flip() {
1227 let mut candidate = bamboo_agent_core::Session::new("preflight", "model");
1228 candidate.metadata = super::super::config::resolve_runtime_metadata(
1229 &bamboo_llm::Config::default(),
1230 "worker",
1231 );
1232 candidate
1233 .metadata
1234 .insert("subagent_type".into(), "worker".into());
1235 for kind in ["custom", "args", "codex", "remote"] {
1236 let mut config = bamboo_llm::Config::default();
1237 let sub = config.subagents_mut();
1238 sub.broker = Some(bamboo_config::BrokerClientConfig {
1239 endpoint: "ws://127.0.0.1:9998".into(),
1240 token: "fixture".into(),
1241 ..Default::default()
1242 });
1243 match kind {
1244 "custom" => sub.worker_bin = Some("/bin/false".into()),
1245 "args" => sub.worker_args = Some(vec!["subagent-worker".into()]),
1246 "codex" => sub.executor = Some("codex".into()),
1247 _ => sub
1248 .remote_placements
1249 .push(bamboo_config::RemoteActorPlacement {
1250 role: "worker".into(),
1251 endpoint: "ws://127.0.0.1:9999".into(),
1252 ..Default::default()
1253 }),
1254 }
1255 let live = std::sync::Arc::new(tokio::sync::RwLock::new(config.clone()));
1256 let runner = super::build_external_child_runner_with_live_config_and_codex_tokens(
1257 &config,
1258 live.clone(),
1259 None,
1260 None,
1261 None,
1262 None,
1263 );
1264 *live.write().await = bamboo_llm::Config::default();
1265 let error = runner
1266 .validate_required_child_context_route(&candidate)
1267 .await
1268 .unwrap_err();
1269 assert!(
1270 error.contains("required_child_context_unsupported"),
1271 "{kind}"
1272 );
1273 }
1274 }
1275
1276 use super::{
1277 codex_approval_policy_name, codex_auth_mode_name, codex_base_url, codex_mode_name,
1278 codex_sandbox_name, codex_wire_api_name, subagent_executor_spec,
1279 };
1280 use bamboo_config::{
1281 CodexApprovalPolicy, CodexAuthMode, CodexMode, CodexSandbox, CodexWireApi, CredentialRef,
1282 };
1283 use bamboo_llm::Config;
1284 use bamboo_subagent::provision::ExecutorSpec;
1285
1286 #[test]
1287 fn codex_runtime_mapping_keeps_parent_loopback_and_custom_url_unambiguous() {
1288 let mut config = Config::default();
1289 config.server.port = 5700;
1290
1291 assert_eq!(codex_auth_mode_name(CodexAuthMode::Bamboo), "bamboo");
1292 assert_eq!(codex_mode_name(CodexMode::AppServer), "app_server");
1293 assert_eq!(codex_wire_api_name(CodexWireApi::Responses), "responses");
1294 assert_eq!(codex_sandbox_name(CodexSandbox::ReadOnly), "read-only");
1295 assert_eq!(
1296 codex_sandbox_name(CodexSandbox::WorkspaceWrite),
1297 "workspace-write"
1298 );
1299 assert_eq!(
1300 codex_approval_policy_name(CodexApprovalPolicy::OnFailure),
1301 "on-failure"
1302 );
1303 assert_eq!(
1304 codex_base_url(&config, CodexAuthMode::Bamboo, None).as_deref(),
1305 Some("http://127.0.0.1:5700/openai/v1")
1306 );
1307 assert_eq!(
1308 codex_base_url(
1309 &config,
1310 CodexAuthMode::Custom,
1311 Some("https://provider.example/v1".to_string()),
1312 )
1313 .as_deref(),
1314 Some("https://provider.example/v1")
1315 );
1316 assert_eq!(codex_base_url(&config, CodexAuthMode::Inherit, None), None);
1317 assert_eq!(codex_base_url(&config, CodexAuthMode::ApiKey, None), None);
1318 }
1319
1320 #[test]
1321 fn durable_codex_fields_map_without_loss_to_worker_spawn_spec() {
1322 let mut config = Config::default();
1323 let subagents = config.subagents_mut();
1324 subagents.executor = Some("codex".to_string());
1325 subagents.codex_binary = Some("/opt/codex/bin/codex".to_string());
1326 subagents.codex_model = Some("gpt-5.4".to_string());
1327 subagents.codex_mode = Some(CodexMode::AppServer);
1328 subagents.codex_auth_mode = Some(CodexAuthMode::Custom);
1329 subagents.codex_base_url = Some("https://provider.example/v1".to_string());
1330 subagents.codex_wire_api = Some(CodexWireApi::Responses);
1331 subagents.codex_provider_key_ref = Some(
1332 CredentialRef::parse("provider.codex-work.api_key").expect("valid credential ref"),
1333 );
1334 subagents.codex_forward_env = Some(vec!["HTTPS_PROXY".to_string()]);
1335 subagents.codex_sandbox = Some(CodexSandbox::WorkspaceWrite);
1336 subagents.codex_approval_policy = Some(CodexApprovalPolicy::OnRequest);
1337 subagents.codex_network_access = Some(true);
1338 subagents.codex_allow_danger_bypass = Some(false);
1339
1340 let spec = subagent_executor_spec(&config).expect("Codex config maps to executor spec");
1341 let ExecutorSpec::Codex {
1342 binary,
1343 model,
1344 mode,
1345 sandbox,
1346 auth_mode,
1347 base_url,
1348 wire_api,
1349 provider_key_ref,
1350 forward_env,
1351 approval_policy,
1352 network_access,
1353 allow_danger_bypass,
1354 ..
1355 } = spec
1356 else {
1357 panic!("expected Codex executor spec");
1358 };
1359 assert_eq!(binary.as_deref(), Some("/opt/codex/bin/codex"));
1360 assert_eq!(model.as_deref(), Some("gpt-5.4"));
1361 assert_eq!(mode.as_deref(), Some("app_server"));
1362 assert_eq!(sandbox.as_deref(), Some("workspace-write"));
1363 assert_eq!(auth_mode.as_deref(), Some("custom"));
1364 assert_eq!(base_url.as_deref(), Some("https://provider.example/v1"));
1365 assert_eq!(wire_api.as_deref(), Some("responses"));
1366 assert_eq!(
1367 provider_key_ref.as_deref(),
1368 Some("provider.codex-work.api_key")
1369 );
1370 assert_eq!(forward_env, Some(vec!["HTTPS_PROXY".to_string()]));
1371 assert_eq!(approval_policy.as_deref(), Some("on-request"));
1372 assert_eq!(network_access, Some(true));
1373 assert_eq!(allow_danger_bypass, Some(false));
1374 }
1375}
1376
1377#[cfg(test)]
1378mod extract_provider_credentials_tests {
1379 use super::extract_provider_credentials;
1380 use bamboo_config::{
1381 AnthropicConfig, BodhiConfig, Config, OpenAIConfig, ProviderInstanceConfig,
1382 };
1383
1384 fn instance(provider_type: &str, api_key: &str) -> ProviderInstanceConfig {
1385 ProviderInstanceConfig {
1386 provider_type: provider_type.to_string(),
1387 label: None,
1388 api_key: api_key.to_string(),
1389 api_key_encrypted: None,
1390 credential_ref: None,
1391 base_url: None,
1392 model: None,
1393 fast_model: None,
1394 vision_model: None,
1395 reasoning_effort: None,
1396 responses_only_models: Vec::new(),
1397 request_overrides: None,
1398 enabled: true,
1399 extra: Default::default(),
1400 }
1401 }
1402
1403 #[test]
1404 fn no_config_yields_no_credentials() {
1405 let config = Config::default();
1406 assert!(extract_provider_credentials(&config).is_empty());
1407 }
1408
1409 #[test]
1414 fn legacy_only_config_yields_credential() {
1415 let mut config = Config::default();
1416 config.providers_mut().anthropic = Some(AnthropicConfig {
1417 api_key: "sk-ant-legacy".to_string(),
1418 base_url: Some("https://api.anthropic.com".to_string()),
1419 ..Default::default()
1420 });
1421
1422 let creds = extract_provider_credentials(&config);
1423 assert_eq!(creds.len(), 1);
1424 let c = &creds[0];
1425 assert_eq!(c.provider, "anthropic");
1426 assert_eq!(c.api_key, "sk-ant-legacy");
1427 assert_eq!(c.base_url.as_deref(), Some("https://api.anthropic.com"));
1428 assert_eq!(c.provider_type.as_deref(), Some("anthropic"));
1429 }
1430
1431 #[test]
1435 fn legacy_bodhi_config_yields_credential() {
1436 let mut config = Config::default();
1437 config.providers_mut().bodhi = Some(BodhiConfig {
1438 api_key: "bhi_sk_legacy".to_string(),
1439 api_key_encrypted: None,
1440 credential_ref: None,
1441 base_url: None,
1442 target_provider: None,
1443 reasoning_effort: None,
1444 extra: Default::default(),
1445 });
1446
1447 let creds = extract_provider_credentials(&config);
1448 assert_eq!(creds.len(), 1);
1449 assert_eq!(creds[0].provider, "bodhi");
1450 assert_eq!(creds[0].api_key, "bhi_sk_legacy");
1451 }
1452
1453 #[test]
1456 fn legacy_config_with_empty_api_key_is_skipped() {
1457 let mut config = Config::default();
1458 config.providers_mut().openai = Some(OpenAIConfig::default());
1459 assert!(extract_provider_credentials(&config).is_empty());
1460 }
1461
1462 #[test]
1465 fn instance_mode_omits_unrelated_legacy_credentials() {
1466 let mut config = Config::default();
1467 config.providers_mut().anthropic = Some(AnthropicConfig {
1468 api_key: "sk-ant-legacy".to_string(),
1469 ..Default::default()
1470 });
1471 let mut openai_work = instance("openai", "sk-oai-work");
1472 openai_work.credential_ref = Some(
1473 bamboo_config::CredentialRef::parse("provider.openai-work.api_key")
1474 .expect("valid provider credential reference"),
1475 );
1476 config
1477 .provider_instances
1478 .insert("openai-work".to_string(), openai_work);
1479 config.default_provider_instance = Some("openai-work".to_string());
1480
1481 let creds = extract_provider_credentials(&config);
1482
1483 assert_eq!(creds.len(), 1);
1484 assert_eq!(creds[0].provider, "openai-work");
1485 assert_eq!(creds[0].api_key, "sk-oai-work");
1486 assert_eq!(creds[0].provider_type.as_deref(), Some("openai"));
1487 assert_eq!(
1488 creds[0].credential_ref.as_deref(),
1489 Some("provider.openai-work.api_key")
1490 );
1491 }
1492
1493 #[test]
1494 fn hybrid_legacy_default_is_the_only_legacy_credential_exported() {
1495 let mut config = Config::default();
1496 config.providers_mut().anthropic = Some(AnthropicConfig {
1497 api_key: "sk-ant-default".to_string(),
1498 ..Default::default()
1499 });
1500 config.providers_mut().openai = Some(OpenAIConfig {
1501 api_key: "sk-oai-stale".to_string(),
1502 ..Default::default()
1503 });
1504 config
1505 .provider_instances
1506 .insert("work".to_string(), instance("openai", "sk-oai-work"));
1507 config.default_provider_instance = Some("anthropic".to_string());
1508
1509 let mut creds = extract_provider_credentials(&config);
1510 creds.sort_by(|a, b| a.provider.cmp(&b.provider));
1511
1512 assert_eq!(creds.len(), 2);
1513 assert_eq!(creds[0].provider, "anthropic");
1514 assert_eq!(creds[0].api_key, "sk-ant-default");
1515 assert_eq!(creds[1].provider, "work");
1516 assert_eq!(creds[1].api_key, "sk-oai-work");
1517 assert!(creds
1518 .iter()
1519 .all(|credential| credential.api_key != "sk-oai-stale"));
1520 }
1521
1522 #[test]
1523 fn hybrid_legacy_provider_fallback_without_explicit_default_remains_exportable() {
1524 let mut config = Config::default();
1525 config.provider = "anthropic".to_string();
1526 config.providers_mut().anthropic = Some(AnthropicConfig {
1527 api_key: "sk-ant-effective-default".to_string(),
1528 ..Default::default()
1529 });
1530 config
1531 .provider_instances
1532 .insert("work".to_string(), instance("openai", "sk-oai-work"));
1533
1534 let mut creds = extract_provider_credentials(&config);
1535 creds.sort_by(|a, b| a.provider.cmp(&b.provider));
1536
1537 assert_eq!(creds.len(), 2);
1538 assert_eq!(creds[0].provider, "anthropic");
1539 assert_eq!(creds[1].provider, "work");
1540 }
1541
1542 #[test]
1543 fn disabled_instance_credential_is_not_exported() {
1544 let mut config = Config::default();
1545 let mut disabled = instance("openai", "sk-disabled");
1546 disabled.enabled = false;
1547 config
1548 .provider_instances
1549 .insert("disabled".to_string(), disabled);
1550
1551 assert!(extract_provider_credentials(&config).is_empty());
1552 }
1553}
1554
1555#[cfg(test)]
1556mod placement_resolver_tests {
1557 use super::{node_display_name, resolve_remote_placements, resolve_schedulable_placements};
1558 use bamboo_config::cluster_fabric::{
1559 DeployProfile, Node, NodePlacement, SshAuth, SshTarget, TrustLevel,
1560 };
1561 use bamboo_config::{RemoteActorPlacement, SchedulablePlacement};
1562
1563 #[test]
1564 fn explicit_broker_route_retains_unavailable_selection_and_redacted_debug() {
1565 let route = bamboo_config::RemoteBrokerPeer {
1566 parent_mailbox: "host".into(),
1567 worker_mailbox: "worker".into(),
1568 parent_role: Some("host".into()),
1569 worker_role: Some("worker".into()),
1570 };
1571 assert!(route.valid());
1572 let mut invalid = route.clone();
1573 invalid.worker_mailbox = "Worker".into();
1574 assert!(!invalid.valid());
1575 invalid.worker_mailbox = "../worker".into();
1576 assert!(!invalid.valid());
1577 let placement = RemoteActorPlacement {
1578 role: "worker".into(),
1579 endpoint: "ws://private.invalid".into(),
1580 token_env: Some("BAMBOO_1431_MISSING_FIXTURE_TOKEN".into()),
1581 ca_cert_file: None,
1582 broker_peer: Some(route),
1583 placement_requirements: None,
1584 };
1585 let parsed: RemoteActorPlacement =
1586 serde_json::from_value(serde_json::to_value(&placement).unwrap()).unwrap();
1587 assert_eq!(parsed, placement);
1588 let resolved = resolve_remote_placements(&[placement], &[]);
1589 let selected = &resolved["worker"];
1590 assert!(matches!(selected.broker_peer, Some(Err(()))));
1591 assert!(selected.token.is_none());
1592 assert_eq!(selected.host_label.as_deref(), Some("remote"));
1593 assert!(!format!("{selected:?}").contains("private.invalid"));
1594 assert!(serde_json::from_value::<bamboo_config::RemoteBrokerPeer>(serde_json::json!({
1595 "parent_mailbox":"host", "worker_mailbox":"worker", "parent_role":null, "worker_role":null,"credential":"secret"
1596 })).is_err());
1597 }
1598
1599 fn ssh_node(id: &str, label: &str, host: &str, default_role: Option<&str>) -> Node {
1600 Node {
1601 id: id.into(),
1602 label: label.into(),
1603 placement: NodePlacement::Ssh(SshTarget {
1604 host: host.into(),
1605 port: 22,
1606 username: "u".into(),
1607 auth: SshAuth::SystemSshConfig,
1608 host_key_fingerprint: None,
1609 }),
1610 trust_level: TrustLevel::default(),
1611 deploy: DeployProfile {
1612 default_role: default_role.map(String::from),
1613 ..Default::default()
1614 },
1615 state: None,
1616 enabled: true,
1617 }
1618 }
1619
1620 #[test]
1621 fn node_display_name_prefers_label_then_ssh_host() {
1622 let n = ssh_node("n1", "mini", "mini.local", None);
1623 assert_eq!(node_display_name(&n), "mini");
1624 let mut unlabeled = n.clone();
1625 unlabeled.label = String::new();
1626 assert_eq!(node_display_name(&unlabeled), "mini.local");
1627 }
1628
1629 #[test]
1630 fn schedulable_placement_takes_host_label_from_node_by_default_role() {
1631 let nodes = vec![ssh_node(
1632 "n1",
1633 "mini",
1634 "mini.local",
1635 Some("mac-mini-monitor"),
1636 )];
1637 let placements = vec![SchedulablePlacement {
1638 role: "mac-mini-monitor".into(),
1639 pool: "mac-mini-monitor".into(),
1640 ..Default::default()
1641 }];
1642 let out = resolve_schedulable_placements(&placements, &nodes);
1643 let r = out.get("mac-mini-monitor").expect("role resolved");
1644 assert_eq!(r.pool, "mac-mini-monitor");
1645 assert_eq!(r.host_label.as_deref(), Some("mini"));
1646 }
1647
1648 #[test]
1649 fn remote_placement_takes_host_label_from_node_by_ssh_host() {
1650 let nodes = vec![ssh_node("n1", "mini", "mini.local", None)];
1651 let placements = vec![RemoteActorPlacement {
1652 role: "explorer".into(),
1653 endpoint: "ws://mini.local:8899".into(),
1654 ..Default::default()
1655 }];
1656 let out = resolve_remote_placements(&placements, &nodes);
1657 assert_eq!(
1658 out.get("explorer").unwrap().host_label.as_deref(),
1659 Some("mini")
1660 );
1661 assert!(matches!(out["explorer"].broker_peer, Some(Err(()))));
1662 assert!(out["explorer"].token.is_none());
1663 }
1664
1665 #[test]
1666 fn no_host_label_when_no_node_matches() {
1667 let nodes = vec![ssh_node("n1", "mini", "mini.local", Some("other-role"))];
1668 let sched = vec![SchedulablePlacement {
1669 role: "x".into(),
1670 pool: "unmatched".into(),
1671 ..Default::default()
1672 }];
1673 assert_eq!(
1674 resolve_schedulable_placements(&sched, &nodes)
1675 .get("x")
1676 .unwrap()
1677 .host_label,
1678 None
1679 );
1680 let remote = vec![RemoteActorPlacement {
1681 role: "y".into(),
1682 endpoint: "ws://other-host:9000".into(),
1683 ..Default::default()
1684 }];
1685 assert_eq!(
1686 resolve_remote_placements(&remote, &nodes)
1687 .get("y")
1688 .unwrap()
1689 .host_label,
1690 None
1691 );
1692 }
1693}