Skip to main content

bamboo_engine/external_agents/
runtime.rs

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/// Host-owned observation of its actual concrete tool surface at activation.
16/// The caller cannot submit a resolved ceiling through the public tool schema.
17#[async_trait]
18pub trait NativeToolCeilingSource: Send + Sync {
19    async fn observe(&self, session: &bamboo_domain::Session) -> Result<Vec<String>, String>;
20}
21
22/// Whether the default Child route explicitly requests the process runtime.
23/// Native Bamboo defaults, including tuning limits and a stale discovery path,
24/// do not select Worker IPC. An unknown explicit executor remains fail-closed.
25pub 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
32/// Whether existing explicit process/remote configuration needs legacy actor
33/// infrastructure. This is not a per-Child routing decision: a configured remote
34/// role or expert profile must not move unrelated native Children off Tokio.
35pub 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
111/// Composite router that delegates to the first matching external child runner.
112pub 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        // Deliberately conservative for this native-only first slice. Mixed
152        // process/remote routers retain their existing explicit dispatch path.
153        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    /// #68: fan the per-run escalation bridge out to every inner runner. The
339    /// composite is what `build_external_child_runner` returns and what the
340    /// worker retains, so without this forward the bind would hit the trait's
341    /// no-op default and the wrapped `ActorChildRunner`s would never see it.
342    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
379/// Build the child runner from the application config.
380///
381/// Sub-agents always run as actors (the in-process runtime was removed), so the
382/// built-in **local actor** worker is always part of the composite — its worker
383/// binary, arguments, and discovery dir are all derived; no expert tables
384/// needed. Expert `externalAgents` profiles add extra routers so
385/// `external.agent_id` metadata can pin specific roles to other agents. Returns
386/// a composite router that delegates to the first matching runner.
387pub fn build_external_child_runner(config: &Config) -> Arc<dyn ExternalChildRunner> {
388    build_external_child_runner_with_registry(config, None)
389}
390
391/// Build the child runner with an AppState-scoped durable approval registry.
392pub 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
399/// Build the child runner with durable approval state and an optional
400/// parent-agent model reviewer for forced-ask requests.
401pub 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
416/// Preserve the already-selected legacy Worker entrypoint's nested process
417/// backend. Its isolated store is not a Host canonical store for Tokio Children.
418/// This explicit compatibility assembly is never a native-runtime fallback.
419pub 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        // The old process backend already derived these exact arguments.
429        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
439/// Full server wiring, including the process-ephemeral Codex per-run token
440/// authority. Non-server callers keep using the compatibility wrapper above.
441pub 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
459/// Full server wiring with an AppState-owned live config source. Provider
460/// credentials and the default provider are resolved for each new child
461/// activation, so provider reloads do not require rebuilding the actor runner.
462pub 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/// The server binds this to the same Builtin Arc and complete base routing
482/// chain assembled for this AppState, before applying the Root role fence.
483#[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    // Ordinary native Child executions stay in this Host process. The old
518    // Worker/broker implementation remains available only for explicit legacy
519    // configuration; a local task failure never selects it as a fallback.
520    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        // Actor protocol: spawn a local worker binary over the bamboo-subagent WS protocol.
562        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            // #217: default under the persistent data-dir subagents home
571            // instead of `env::temp_dir()`, so fabric discovery state
572            // survives reboots and stays inside the tenant's data dir.
573            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                // #443: binary/model/permission_mode/isolation/env-forward
584                // are plumbed from the profile's `claude_code_*` fields.
585                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
715/// Classify the resolved local launch without granting execution authority.
716fn 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    // Explicit selection of the retained local process backend is still the
721    // built-in route only for this Host executable and its exact Worker argv.
722    // Failed path observations never qualify. Actor admission rechecks this
723    // identity and the actual Worker capabilities before any Child is created.
724    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
732/// Build the explicitly selected local process runner from typed configuration.
733fn 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    // #217: default under the persistent data-dir subagents home instead of
758    // `env::temp_dir()` (mirrors the `build_external_child_runner` arm above).
759    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
810/// Convert the durable typed `subagents` section into the exact worker
811/// provisioning executor. This is deliberately independent of actor launch so
812/// the settings-to-spawn contract can be tested directly.
813fn 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
902/// Resolve every configured role, including invalid legacy entries. An explicit
903/// scheduled role must never silently fall through to Local.
904fn 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            // Ambiguous authority is unavailable, regardless of entry order.
940            if let Some(route) = out.get_mut(&p.role) {
941                route.broker_parent = None;
942            }
943        }
944    }
945    out
946}
947
948/// Friendly display name for a cluster node whose worker serves `role`
949/// (`deploy.default_role`) — the operator `label`, else its ssh host. Used to
950/// stamp the UI placement badge from the node's own metadata.
951fn 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
961/// Friendly display name for a cluster node whose ssh host matches `endpoint`'s
962/// host — so a `remote_placements` endpoint pointing at a known node shows the
963/// node's label rather than a bare IP.
964fn 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
997/// Resolve pinned remote placements. Broker peer selection is required to run;
998/// old direct-worker configurations remain visible but unavailable, without
999/// exposing their credentials or silently rerouting to Local.
1000fn 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; // Invalid explicit routes remain selected, never Local fallback.
1056        }
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                // Badge from the node's own metadata when the endpoint points at
1064                // a known cluster node; else the endpoint host is used downstream.
1065                host_label: node_label_for_endpoint(nodes, &p.endpoint),
1066                broker_peer: Some(Err(())),
1067                requirements: None,
1068            },
1069        );
1070    }
1071    out
1072}
1073
1074/// Snapshot per-provider credentials from the parent config for actor
1075/// provisioning. `api_key` (plaintext, in-memory only) is `#[serde(skip_serializing)]`
1076/// on every legacy single-instance provider struct — it's hydrated from
1077/// `api_key_encrypted` at load time but deliberately never round-tripped
1078/// through serde, so a `serde_json::to_value` projection of `config.providers`
1079/// sees none of it (#495). Read each typed struct's `api_key` field directly
1080/// instead, mirroring how `provider_instances` below already has to.
1081pub 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        // Native instances are authoritative. Export enabled explicit
1112        // instances only; stale legacy slots must not leak into child workers.
1113        for (id, instance) in &config.provider_instances {
1114            push_instance(&mut out, id, instance);
1115        }
1116
1117        // Narrow #780 compatibility seam: a hybrid default may still name a
1118        // real legacy alias not yet materialized. Mirror the registry's exact
1119        // rule and add only that default, never every legacy credential.
1120        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    // Legacy single-instance slots: providers.anthropic / openai / gemini /
1133    // bodhi. `copilot` is intentionally omitted — it authenticates via device
1134    // flow and has no `api_key` field to extract.
1135    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(&current, &exact));
1200        assert!(super::builtin_local_worker_launch(
1201            &current.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(&current, &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    /// #495 — a legacy single-instance provider (`config.providers.anthropic`
1410    /// etc.) must yield its `api_key` even though the field is
1411    /// `#[serde(skip_serializing)]`, because the extraction now reads the
1412    /// typed struct instead of projecting through `serde_json::to_value`.
1413    #[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    /// `bodhi` doesn't derive `Default`, so it's exercised separately —
1432    /// covers the last of the four legacy structs the fix touches
1433    /// (openai/anthropic/gemini already share the `Default`-derive path).
1434    #[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    /// A legacy slot with an empty `api_key` (struct present but never
1454    /// configured) must not produce a bogus empty credential.
1455    #[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    /// Explicit instances are authoritative: unrelated stale legacy slots do
1463    /// not cross the actor provisioning boundary.
1464    #[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}