1use std::sync::Arc;
7
8use bamboo_agent_core::Session;
9
10use crate::runtime::{AgentRuntime, AgentRuntimeBuilder, ExecuteRequest};
11use bamboo_domain::RuntimeSessionPersistence;
12
13#[derive(Clone)]
25pub struct Agent {
26 runtime: Arc<AgentRuntime>,
27}
28
29pub struct DirectExecutionLease {
35 target_session_id: String,
36 router: Option<Arc<crate::session_activation::SessionActivationRouter>>,
37 registration: Option<crate::session_activation::SessionRunRegistration>,
38}
39
40impl DirectExecutionLease {
41 pub async fn abandon(mut self) {
44 if let Some(registration) = self.registration.take() {
45 registration.abandon().await;
46 }
47 }
48}
49
50pub struct SdkInputSubmission {
53 session_id: String,
54 execution_id: String,
55 user: bamboo_agent_core::Message,
56 hook_runner: Arc<crate::runtime::hooks::HookRunner>,
57}
58
59impl SdkInputSubmission {
60 pub fn execution_id(&self) -> &str {
61 &self.execution_id
62 }
63 pub fn user(&self) -> &bamboo_agent_core::Message {
64 &self.user
65 }
66}
67
68impl Agent {
69 pub fn from_runtime(runtime: Arc<AgentRuntime>) -> Self {
71 Agent { runtime }
72 }
73
74 #[doc(hidden)]
77 pub fn with_execution_persistence(
78 &self,
79 persistence: Arc<dyn RuntimeSessionPersistence>,
80 ) -> Self {
81 let mut runtime = (*self.runtime).clone();
82 runtime.persistence = persistence;
83 runtime.inherited_child_wait_captured = true;
84 Self::from_runtime(Arc::new(runtime))
85 }
86
87 #[doc(hidden)]
89 pub fn with_sdk_skill_execution_host(
90 &self,
91 host: Arc<dyn crate::runtime::config::SdkSkillExecutionHost>,
92 ) -> Self {
93 self.with_skill_execution_host(host)
94 }
95
96 #[doc(hidden)]
98 pub fn with_skill_execution_host(
99 &self,
100 host: Arc<dyn crate::runtime::config::SkillExecutionHost>,
101 ) -> Self {
102 let mut runtime = (*self.runtime).clone();
103 runtime.sdk_skill_execution_host = Some(host);
104 Self::from_runtime(Arc::new(runtime))
105 }
106
107 #[doc(hidden)]
109 pub fn with_workspace_execution_host(
110 &self,
111 host: Arc<dyn crate::runtime::config::WorkspaceExecutionHost>,
112 ) -> Self {
113 let mut runtime = (*self.runtime).clone();
114 runtime.workspace_execution_host = Some(host);
115 Self::from_runtime(Arc::new(runtime))
116 }
117
118 #[doc(hidden)]
121 pub async fn precheck_sdk_user_input(
122 &self,
123 session: &mut Session,
124 user: &bamboo_agent_core::Message,
125 app_data_dir: Option<std::path::PathBuf>,
126 ) -> crate::runtime::runner::Result<SdkInputSubmission> {
127 session
128 .check_execution_compatibility()
129 .map_err(|error| bamboo_agent_core::AgentError::Tool(error.to_string()))?;
130 if user.role != bamboo_agent_core::Role::User
131 || crate::runtime::config::UntrustedInputObservation::new(&user.id, None).is_none()
132 || session.messages.iter().any(|message| message.id == user.id)
133 {
134 return Err(bamboo_agent_core::AgentError::Tool(
135 "invalid new SDK User".into(),
136 ));
137 }
138 let hook_runner = {
139 let config = self.runtime.config.read().await;
140 Arc::new(
141 self.runtime
142 .hook_runner
143 .with_lifecycle_config(&config.lifecycle_hooks, app_data_dir),
144 )
145 };
146 session.metadata.remove("runtime.plugin_prompt_prechecked");
147 let prompt = hook_runner
148 .apply_portable_user_prompt(session, &user.content)
149 .await?;
150 let mut transformed = user.clone();
151 transformed.content = prompt.clone();
152 if let Some(parts) = transformed.content_parts.as_mut() {
153 if let Some(bamboo_domain::MessagePart::Text { text }) =
154 parts.iter_mut().find(|part| matches!(part, bamboo_domain::MessagePart::Text { text } if text == &user.content))
155 {
156 *text = prompt;
157 }
158 }
159 Ok(SdkInputSubmission {
160 session_id: session.id.clone(),
161 execution_id: crate::runtime::runner::round_prelude::new_execution_id(),
162 user: transformed,
163 hook_runner,
164 })
165 }
166
167 #[doc(hidden)]
170 pub fn append_sdk_user_input(
171 &self,
172 session: &mut Session,
173 user: bamboo_agent_core::Message,
174 receipt: Option<SdkInputSubmission>,
175 request: Option<bamboo_domain::SessionSkillRequest>,
176 ) -> crate::runtime::runner::Result<crate::runtime::config::UntrustedExecutionInputs> {
177 session
178 .check_execution_compatibility()
179 .map_err(|error| bamboo_agent_core::AgentError::Tool(error.to_string()))?;
180 use crate::runtime::config::{
181 sdk_message_digest, sdk_message_header_digest, SdkPendingInput,
182 UntrustedExecutionInputs, UntrustedInputObservation,
183 };
184 let invalid = || bamboo_agent_core::AgentError::Tool("invalid SDK append receipt".into());
185 let header_digest = sdk_message_header_digest(&user);
186 let input_id = user.id.clone();
187 if let Some(receipt) = receipt.as_ref() {
188 if receipt.session_id != session.id
189 || receipt.user.role != bamboo_agent_core::Role::User
190 || user.role != bamboo_agent_core::Role::User
191 || sdk_message_header_digest(&receipt.user) != header_digest
192 || session
193 .messages
194 .iter()
195 .any(|message| message.id == input_id)
196 {
197 return Err(invalid());
198 }
199 }
200 let final_digest = receipt.as_ref().map(|_| sdk_message_digest(&user));
203 if receipt.is_some() {
204 crate::runtime::hooks::HookRunner::mark_user_prompt_prechecked(session, &user.content);
205 }
206 session.add_message(user);
207 let observation =
208 UntrustedInputObservation::new(&input_id, request.as_ref()).ok_or_else(invalid)?;
209 let carrier = UntrustedExecutionInputs::new(vec![observation]).ok_or_else(invalid)?;
210 let (execution_id, hook_runner) = receipt
211 .map(|receipt| (receipt.execution_id, Some(receipt.hook_runner)))
212 .unwrap_or_else(|| {
213 (
214 crate::runtime::runner::round_prelude::new_execution_id(),
215 None,
216 )
217 });
218 Ok(carrier.bind_sdk_append(
219 execution_id,
220 SdkPendingInput {
221 session_id: session.id.clone(),
222 input_id,
223 header_digest,
224 final_digest,
225 },
226 hook_runner,
227 ))
228 }
229
230 pub fn builder() -> AgentBuilder {
232 AgentBuilder::new()
233 }
234
235 pub async fn execute(
237 &self,
238 session: &mut Session,
239 req: ExecuteRequest,
240 ) -> crate::runtime::runner::Result<()> {
241 self.runtime.execute(session, req).await
242 }
243
244 pub async fn execute_with_inputs(
247 &self,
248 session: &mut Session,
249 req: ExecuteRequest,
250 inputs: Option<crate::runtime::config::UntrustedExecutionInputs>,
251 ) -> crate::runtime::runner::Result<()> {
252 self.runtime.execute_with_inputs(session, req, inputs).await
253 }
254
255 pub async fn execute_direct(
266 &self,
267 session: &mut Session,
268 req: ExecuteRequest,
269 ) -> crate::runtime::runner::Result<()> {
270 session
271 .check_execution_compatibility()
272 .map_err(|error| bamboo_agent_core::AgentError::Tool(error.to_string()))?;
273 let lease = self.begin_direct_execution(&session.id).await?;
274 self.check_direct_execution_admission(session, &lease)
275 .await?;
276 self.prepare_external_session_for_execution(session).await?;
277 self.execute_direct_registered(session, req, lease).await
278 }
279
280 pub fn check_direct_execution_admission<'a>(
284 &'a self,
285 session: &'a mut Session,
286 lease: &'a DirectExecutionLease,
287 ) -> impl std::future::Future<Output = crate::runtime::runner::Result<()>> + Send + 'a {
288 let matches_target = lease.target_session_id == session.id;
289 let ownership = lease.registration.as_ref().map(|run| run.ownership());
293 async move {
294 if !matches_target {
295 return Err(bamboo_agent_core::AgentError::LLM(
296 "direct execution lease target does not match session".into(),
297 ));
298 }
299 session
300 .check_execution_compatibility()
301 .map_err(|error| bamboo_agent_core::AgentError::Tool(error.to_string()))?;
302 super::counted_loop::admit_registered(
303 session,
304 ownership.as_ref(),
305 self.activation_router(),
306 self.persistence().as_ref(),
307 )
308 .await
309 }
310 }
311
312 pub async fn prepare_external_session_for_execution(
320 &self,
321 session: &mut Session,
322 ) -> crate::runtime::runner::Result<()> {
323 crate::session_app::execution_prep::prepare_external_session_for_execution(
324 session,
325 self.runtime.project_context_resolver.as_deref(),
326 )
327 .await
328 }
329
330 pub async fn prepare_external_project_assignment_read_only(
334 &self,
335 session: &mut Session,
336 ) -> crate::runtime::runner::Result<()> {
337 let resolver = self
338 .runtime
339 .project_context_resolver
340 .as_deref()
341 .ok_or_else(|| {
342 bamboo_agent_core::AgentError::ProjectContext(
343 "Project assignment requires a ProjectContextResolver".to_string(),
344 )
345 })?;
346 resolver
347 .refresh_session_prompt_read_only(session)
348 .await
349 .map(|_| ())
350 .map_err(|error| bamboo_agent_core::AgentError::ProjectContext(error.to_string()))
351 }
352
353 pub async fn begin_direct_execution(
356 &self,
357 target_session_id: &str,
358 ) -> crate::runtime::runner::Result<DirectExecutionLease> {
359 let Some(router) = self.activation_router().cloned() else {
360 return Ok(DirectExecutionLease {
361 target_session_id: target_session_id.to_string(),
362 router: None,
363 registration: None,
364 });
365 };
366 let run_id = format!("sdk-direct-{}", uuid::Uuid::new_v4());
367 let registration = router
368 .register_run(target_session_id, &run_id)
369 .await
370 .map_err(|error| bamboo_agent_core::AgentError::LLM(error.to_string()))?;
371 Ok(DirectExecutionLease {
372 target_session_id: target_session_id.to_string(),
373 router: Some(router),
374 registration: Some(registration),
375 })
376 }
377
378 pub async fn execute_direct_registered(
381 &self,
382 session: &mut Session,
383 req: ExecuteRequest,
384 lease: DirectExecutionLease,
385 ) -> crate::runtime::runner::Result<()> {
386 self.execute_direct_registered_with_inputs(session, req, lease, None)
387 .await
388 }
389
390 pub async fn execute_direct_registered_with_inputs(
392 &self,
393 session: &mut Session,
394 req: ExecuteRequest,
395 lease: DirectExecutionLease,
396 inputs: Option<crate::runtime::config::UntrustedExecutionInputs>,
397 ) -> crate::runtime::runner::Result<()> {
398 self.check_direct_execution_admission(session, &lease)
399 .await?;
400 let inherited = self.persistence().inherited_child_wait().or_else(|| {
401 (!self.runtime.inherited_child_wait_captured)
402 .then(|| bamboo_domain::InheritedChildWait::capture(session))
403 .flatten()
404 });
405 let agent = if let Some(inherited) = inherited {
406 inherited
407 .validate_session(session)
408 .map_err(|error| bamboo_agent_core::AgentError::LLM(error.to_string()))?;
409 if self.persistence().inherited_child_wait().is_some() {
410 self.clone()
411 } else {
412 self.with_execution_persistence(
413 self.persistence()
414 .bind_inherited_child_wait(inherited)
415 .map_err(|error| bamboo_agent_core::AgentError::LLM(error.to_string()))?,
416 )
417 }
418 } else {
419 self.with_execution_persistence(self.persistence().clone())
420 };
421 agent
422 .execute_direct_registered_bound(session, req, lease, inputs)
423 .await
424 }
425
426 async fn execute_direct_registered_bound(
427 &self,
428 session: &mut Session,
429 mut req: ExecuteRequest,
430 mut lease: DirectExecutionLease,
431 inputs: Option<crate::runtime::config::UntrustedExecutionInputs>,
432 ) -> crate::runtime::runner::Result<()> {
433 if lease.target_session_id != session.id {
434 return Err(bamboo_agent_core::AgentError::LLM(format!(
435 "direct execution lease target {} does not match session {}",
436 lease.target_session_id, session.id
437 )));
438 }
439 let Some(router) = lease.router.take() else {
440 return self.execute_with_inputs(session, req, inputs).await;
441 };
442 let mut registration = lease.registration.take().ok_or_else(|| {
443 bamboo_agent_core::AgentError::LLM(
444 "direct execution lease is missing its router registration".to_string(),
445 )
446 })?;
447 if !router
448 .bind_run_lifetime(®istration.ownership(), session)
449 .await
450 {
451 return Err(bamboo_agent_core::AgentError::Tool(
452 "registered run Session lifetime differs".into(),
453 ));
454 }
455 if let Some(counted) = req.counted_loop.as_mut() {
456 counted.bind_owner(registration.ownership(), session);
457 }
458 if let Some(control) = req.loop_control.as_mut() {
459 control.bind_owner(registration.ownership(), session);
460 control.bind_transport_cleanup(
461 &mut registration,
462 self.storage().clone(),
463 self.session_inbox().cloned(),
464 self.persistence().root_actor_writer(),
465 );
466 }
467 let result = if req.loop_control.is_some() {
468 bamboo_agent_core::tools::observe_tool_suspension(false, async {
469 let token = bamboo_agent_core::tools::capture_suspension_restriction().ok_or_else(
470 || {
471 bamboo_agent_core::AgentError::Tool(
472 "direct loop_control execution is missing its host observation scope"
473 .into(),
474 )
475 },
476 )?;
477 registration.bind_tool_suspension_observation(token);
480 self.execute_with_inputs(session, req, inputs).await
481 })
482 .await
483 .0
484 } else {
485 self.execute_with_inputs(session, req, inputs).await
486 };
487 if registration.observed_forbidden_running() {
488 registration.quarantine().await;
493 return match result {
494 Err(error) => Err(error),
495 Ok(()) => Err(bamboo_agent_core::AgentError::Tool(
496 "suspension_disabled: detached Running outcome; stop unconfirmed".into(),
497 )),
498 };
499 }
500
501 let executed_admitted_generation = session
504 .session_inbox_admission()
505 .map_or(0, |state| state.last_admitted_sequence);
506 registration.begin_finalization().await;
507
508 let legacy_migration = crate::runtime::runner::state_bridge::migrate_legacy_pending_only(
509 session,
510 Some(self.storage()),
511 Some(self.persistence()),
512 self.session_inbox(),
513 )
514 .await;
515 if let Some(generation) = legacy_migration.highest_generation {
516 session.session_inbox_admission_mut().observe(generation);
517 }
518 let pending_generation = session
519 .session_inbox_admission()
520 .and_then(|state| state.pending_activation_generation());
521 if let Some(generation) = pending_generation {
522 let activation_ready = if let Some(inbox) = self.session_inbox() {
523 match inbox
524 .mark_activation_eligible(
525 &session.id,
526 generation,
527 bamboo_domain::SessionActivationPolicy::InterruptSpecificWait,
528 )
529 .await
530 {
531 Ok(()) => true,
532 Err(error) => {
533 tracing::error!(
534 session_id = %session.id,
535 %error,
536 "failed to persist direct SDK SessionInbox activation watermark"
537 );
538 false
539 }
540 }
541 } else {
542 false
543 };
544 if activation_ready {
545 if let Err(error) = bamboo_domain::SessionActivationPort::request_activation(
546 router.as_ref(),
547 &session.id,
548 generation,
549 )
550 .await
551 {
552 tracing::error!(
553 session_id = %session.id,
554 %error,
555 "failed to hand direct SDK SessionInbox generation to activation router"
556 );
557 }
558 }
559 }
560
561 if let Err(error) = self.persistence().checkpoint_runtime_session(session).await {
564 tracing::warn!(
565 session_id = %session.id,
566 %error,
567 "failed to checkpoint direct SDK terminal SessionInbox state"
568 );
569 }
570 if let Err(error) = registration.finish(executed_admitted_generation).await {
571 tracing::error!(
572 session_id = %session.id,
573 %error,
574 "direct SDK SessionInbox finalization failed"
575 );
576 }
577
578 result
579 }
580
581 pub fn storage(&self) -> &Arc<dyn bamboo_agent_core::storage::Storage> {
583 &self.runtime.storage
584 }
585
586 pub fn persistence(&self) -> &Arc<dyn RuntimeSessionPersistence> {
588 &self.runtime.persistence
589 }
590
591 pub fn session_inbox(&self) -> Option<&Arc<dyn bamboo_domain::SessionInboxPort>> {
592 self.runtime.session_inbox.as_ref()
593 }
594
595 pub async fn admit_session_inbox_at_safe_boundary(
600 &self,
601 session: &mut bamboo_agent_core::Session,
602 ) -> usize {
603 crate::runtime::runner::state_bridge::refresh_turn_boundary_with_inbox(
604 session,
605 Some(self.storage()),
606 Some(self.persistence()),
607 self.session_inbox(),
608 )
609 .await
610 .merged
611 }
612
613 pub async fn admit_session_inbox_at_safe_boundary_checked(
617 &self,
618 session: &mut bamboo_agent_core::Session,
619 ) -> Result<usize, bamboo_agent_core::AgentError> {
620 let refresh = crate::runtime::runner::state_bridge::refresh_turn_boundary_with_inbox(
621 session,
622 Some(self.storage()),
623 Some(self.persistence()),
624 self.session_inbox(),
625 )
626 .await;
627 if let Some(error) = refresh.admission_error {
628 return Err(bamboo_agent_core::AgentError::Tool(error));
629 }
630 Ok(refresh.merged)
631 }
632
633 pub fn activation_router(
634 &self,
635 ) -> Option<&Arc<crate::session_activation::SessionActivationRouter>> {
636 self.runtime.activation_router.as_ref()
637 }
638
639 pub fn session_messenger(&self) -> Option<&Arc<crate::SessionMessenger>> {
640 self.runtime.session_messenger.as_ref()
641 }
642
643 pub fn default_tools(&self) -> &Arc<dyn bamboo_agent_core::tools::ToolExecutor> {
653 &self.runtime.default_tools
654 }
655}
656
657pub struct AgentBuilder {
665 inner: AgentRuntimeBuilder,
666}
667
668impl AgentBuilder {
669 pub fn new() -> Self {
670 Self {
671 inner: AgentRuntimeBuilder::new(),
672 }
673 }
674
675 pub fn storage(mut self, v: Arc<dyn bamboo_agent_core::storage::Storage>) -> Self {
676 self.inner = self.inner.storage(v);
677 self
678 }
679
680 pub fn persistence(mut self, v: Arc<dyn RuntimeSessionPersistence>) -> Self {
681 self.inner = self.inner.persistence(v);
682 self
683 }
684
685 pub fn session_inbox(mut self, v: Arc<dyn bamboo_domain::SessionInboxPort>) -> Self {
686 self.inner = self.inner.session_inbox(v);
687 self
688 }
689
690 pub fn activation_router(
691 mut self,
692 v: Arc<crate::session_activation::SessionActivationRouter>,
693 ) -> Self {
694 self.inner = self.inner.activation_router(v);
695 self
696 }
697
698 pub fn session_messenger(mut self, v: Arc<crate::SessionMessenger>) -> Self {
699 self.inner = self.inner.session_messenger(v);
700 self
701 }
702
703 pub fn attachment_reader(
704 mut self,
705 v: Arc<dyn bamboo_agent_core::storage::AttachmentReader>,
706 ) -> Self {
707 self.inner = self.inner.attachment_reader(v);
708 self
709 }
710
711 pub fn skill_manager(mut self, v: Arc<bamboo_skills::SkillManager>) -> Self {
712 self.inner = self.inner.skill_manager(v);
713 self
714 }
715
716 pub fn project_context_resolver(
717 mut self,
718 v: Arc<crate::project_context::ProjectContextResolver>,
719 ) -> Self {
720 self.inner = self.inner.project_context_resolver(v);
721 self
722 }
723
724 pub fn metrics_collector(mut self, v: bamboo_metrics::MetricsCollector) -> Self {
725 self.inner = self.inner.metrics_collector(v);
726 self
727 }
728
729 pub fn config(mut self, v: Arc<tokio::sync::RwLock<bamboo_llm::Config>>) -> Self {
730 self.inner = self.inner.config(v);
731 self
732 }
733
734 pub fn model_limits_section(
737 mut self,
738 section: Arc<bamboo_config::LiveSection<bamboo_config::ModelLimitsSection>>,
739 ) -> Self {
740 self.inner = self.inner.model_limits_section(section);
741 self
742 }
743
744 pub fn permission_config(mut self, v: Arc<bamboo_tools::permission::PermissionConfig>) -> Self {
745 self.inner = self.inner.permission_config(v);
746 self
747 }
748
749 pub fn permission_mode(mut self, v: bamboo_domain::PermissionMode) -> Self {
750 self.inner = self.inner.permission_mode(v);
751 self
752 }
753
754 pub fn provider(mut self, v: Arc<dyn bamboo_llm::LLMProvider>) -> Self {
755 self.inner = self.inner.provider(v);
756 self
757 }
758
759 pub fn memory_store(mut self, v: bamboo_memory::memory_store::MemoryStore) -> Self {
760 self.inner = self.inner.memory_store(v);
761 self
762 }
763
764 pub fn default_tools(mut self, v: Arc<dyn bamboo_agent_core::tools::ToolExecutor>) -> Self {
765 self.inner = self.inner.default_tools(v);
766 self
767 }
768
769 pub fn hook_runner(mut self, v: Arc<crate::runtime::HookRunner>) -> Self {
771 self.inner = self.inner.hook_runner(v);
772 self
773 }
774
775 pub fn build(self) -> Result<Agent, &'static str> {
776 let runtime = self.inner.build()?;
777 Ok(Agent {
778 runtime: Arc::new(runtime),
779 })
780 }
781}
782
783impl Default for AgentBuilder {
784 fn default() -> Self {
785 Self::new()
786 }
787}