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)]
110 pub async fn precheck_sdk_user_input(
111 &self,
112 session: &mut Session,
113 user: &bamboo_agent_core::Message,
114 app_data_dir: Option<std::path::PathBuf>,
115 ) -> crate::runtime::runner::Result<SdkInputSubmission> {
116 if user.role != bamboo_agent_core::Role::User
117 || crate::runtime::config::UntrustedInputObservation::new(&user.id, None).is_none()
118 || session.messages.iter().any(|message| message.id == user.id)
119 {
120 return Err(bamboo_agent_core::AgentError::Tool(
121 "invalid new SDK User".into(),
122 ));
123 }
124 let hook_runner = {
125 let config = self.runtime.config.read().await;
126 Arc::new(
127 self.runtime
128 .hook_runner
129 .with_lifecycle_config(&config.lifecycle_hooks, app_data_dir),
130 )
131 };
132 session.metadata.remove("runtime.plugin_prompt_prechecked");
133 let prompt = hook_runner
134 .apply_portable_user_prompt(session, &user.content)
135 .await?;
136 let mut transformed = user.clone();
137 transformed.content = prompt.clone();
138 if let Some(parts) = transformed.content_parts.as_mut() {
139 if let Some(bamboo_domain::MessagePart::Text { text }) =
140 parts.iter_mut().find(|part| matches!(part, bamboo_domain::MessagePart::Text { text } if text == &user.content))
141 {
142 *text = prompt;
143 }
144 }
145 Ok(SdkInputSubmission {
146 session_id: session.id.clone(),
147 execution_id: crate::runtime::runner::round_prelude::new_execution_id(),
148 user: transformed,
149 hook_runner,
150 })
151 }
152
153 #[doc(hidden)]
156 pub fn append_sdk_user_input(
157 &self,
158 session: &mut Session,
159 user: bamboo_agent_core::Message,
160 receipt: Option<SdkInputSubmission>,
161 request: Option<bamboo_domain::SessionSkillRequest>,
162 ) -> crate::runtime::runner::Result<crate::runtime::config::UntrustedExecutionInputs> {
163 use crate::runtime::config::{
164 sdk_message_digest, sdk_message_header_digest, SdkPendingInput,
165 UntrustedExecutionInputs, UntrustedInputObservation,
166 };
167 let invalid = || bamboo_agent_core::AgentError::Tool("invalid SDK append receipt".into());
168 let header_digest = sdk_message_header_digest(&user);
169 let input_id = user.id.clone();
170 if let Some(receipt) = receipt.as_ref() {
171 if receipt.session_id != session.id
172 || receipt.user.role != bamboo_agent_core::Role::User
173 || user.role != bamboo_agent_core::Role::User
174 || sdk_message_header_digest(&receipt.user) != header_digest
175 || session
176 .messages
177 .iter()
178 .any(|message| message.id == input_id)
179 {
180 return Err(invalid());
181 }
182 }
183 let final_digest = receipt.as_ref().map(|_| sdk_message_digest(&user));
186 if receipt.is_some() {
187 crate::runtime::hooks::HookRunner::mark_user_prompt_prechecked(session, &user.content);
188 }
189 session.add_message(user);
190 let observation =
191 UntrustedInputObservation::new(&input_id, request.as_ref()).ok_or_else(invalid)?;
192 let carrier = UntrustedExecutionInputs::new(vec![observation]).ok_or_else(invalid)?;
193 let (execution_id, hook_runner) = receipt
194 .map(|receipt| (receipt.execution_id, Some(receipt.hook_runner)))
195 .unwrap_or_else(|| {
196 (
197 crate::runtime::runner::round_prelude::new_execution_id(),
198 None,
199 )
200 });
201 Ok(carrier.bind_sdk_append(
202 execution_id,
203 SdkPendingInput {
204 session_id: session.id.clone(),
205 input_id,
206 header_digest,
207 final_digest,
208 },
209 hook_runner,
210 ))
211 }
212
213 pub fn builder() -> AgentBuilder {
215 AgentBuilder::new()
216 }
217
218 pub async fn execute(
220 &self,
221 session: &mut Session,
222 req: ExecuteRequest,
223 ) -> crate::runtime::runner::Result<()> {
224 self.runtime.execute(session, req).await
225 }
226
227 pub async fn execute_with_inputs(
230 &self,
231 session: &mut Session,
232 req: ExecuteRequest,
233 inputs: Option<crate::runtime::config::UntrustedExecutionInputs>,
234 ) -> crate::runtime::runner::Result<()> {
235 self.runtime.execute_with_inputs(session, req, inputs).await
236 }
237
238 pub async fn execute_direct(
249 &self,
250 session: &mut Session,
251 req: ExecuteRequest,
252 ) -> crate::runtime::runner::Result<()> {
253 let lease = self.begin_direct_execution(&session.id).await?;
254 self.check_direct_execution_admission(session, &lease)
255 .await?;
256 self.prepare_external_session_for_execution(session).await?;
257 self.execute_direct_registered(session, req, lease).await
258 }
259
260 pub fn check_direct_execution_admission<'a>(
264 &'a self,
265 session: &'a mut Session,
266 lease: &'a DirectExecutionLease,
267 ) -> impl std::future::Future<Output = crate::runtime::runner::Result<()>> + Send + 'a {
268 let matches_target = lease.target_session_id == session.id;
269 let ownership = lease.registration.as_ref().map(|run| run.ownership());
273 async move {
274 if !matches_target {
275 return Err(bamboo_agent_core::AgentError::LLM(
276 "direct execution lease target does not match session".into(),
277 ));
278 }
279 super::counted_loop::admit_registered(
280 session,
281 ownership.as_ref(),
282 self.activation_router(),
283 self.persistence().as_ref(),
284 )
285 .await
286 }
287 }
288
289 pub async fn prepare_external_session_for_execution(
297 &self,
298 session: &mut Session,
299 ) -> crate::runtime::runner::Result<()> {
300 crate::session_app::execution_prep::prepare_external_session_for_execution(
301 session,
302 self.runtime.project_context_resolver.as_deref(),
303 )
304 .await
305 }
306
307 pub async fn prepare_external_project_assignment_read_only(
311 &self,
312 session: &mut Session,
313 ) -> crate::runtime::runner::Result<()> {
314 let resolver = self
315 .runtime
316 .project_context_resolver
317 .as_deref()
318 .ok_or_else(|| {
319 bamboo_agent_core::AgentError::ProjectContext(
320 "Project assignment requires a ProjectContextResolver".to_string(),
321 )
322 })?;
323 resolver
324 .refresh_session_prompt_read_only(session)
325 .await
326 .map(|_| ())
327 .map_err(|error| bamboo_agent_core::AgentError::ProjectContext(error.to_string()))
328 }
329
330 pub async fn begin_direct_execution(
333 &self,
334 target_session_id: &str,
335 ) -> crate::runtime::runner::Result<DirectExecutionLease> {
336 let Some(router) = self.activation_router().cloned() else {
337 return Ok(DirectExecutionLease {
338 target_session_id: target_session_id.to_string(),
339 router: None,
340 registration: None,
341 });
342 };
343 let run_id = format!("sdk-direct-{}", uuid::Uuid::new_v4());
344 let registration = router
345 .register_run(target_session_id, &run_id)
346 .await
347 .map_err(|error| bamboo_agent_core::AgentError::LLM(error.to_string()))?;
348 Ok(DirectExecutionLease {
349 target_session_id: target_session_id.to_string(),
350 router: Some(router),
351 registration: Some(registration),
352 })
353 }
354
355 pub async fn execute_direct_registered(
358 &self,
359 session: &mut Session,
360 req: ExecuteRequest,
361 lease: DirectExecutionLease,
362 ) -> crate::runtime::runner::Result<()> {
363 self.execute_direct_registered_with_inputs(session, req, lease, None)
364 .await
365 }
366
367 pub async fn execute_direct_registered_with_inputs(
369 &self,
370 session: &mut Session,
371 req: ExecuteRequest,
372 lease: DirectExecutionLease,
373 inputs: Option<crate::runtime::config::UntrustedExecutionInputs>,
374 ) -> crate::runtime::runner::Result<()> {
375 self.check_direct_execution_admission(session, &lease)
376 .await?;
377 let inherited = self.persistence().inherited_child_wait().or_else(|| {
378 (!self.runtime.inherited_child_wait_captured)
379 .then(|| bamboo_domain::InheritedChildWait::capture(session))
380 .flatten()
381 });
382 let agent = if let Some(inherited) = inherited {
383 inherited
384 .validate_session(session)
385 .map_err(|error| bamboo_agent_core::AgentError::LLM(error.to_string()))?;
386 if self.persistence().inherited_child_wait().is_some() {
387 self.clone()
388 } else {
389 self.with_execution_persistence(
390 self.persistence()
391 .bind_inherited_child_wait(inherited)
392 .map_err(|error| bamboo_agent_core::AgentError::LLM(error.to_string()))?,
393 )
394 }
395 } else {
396 self.with_execution_persistence(self.persistence().clone())
397 };
398 agent
399 .execute_direct_registered_bound(session, req, lease, inputs)
400 .await
401 }
402
403 async fn execute_direct_registered_bound(
404 &self,
405 session: &mut Session,
406 mut req: ExecuteRequest,
407 mut lease: DirectExecutionLease,
408 inputs: Option<crate::runtime::config::UntrustedExecutionInputs>,
409 ) -> crate::runtime::runner::Result<()> {
410 if lease.target_session_id != session.id {
411 return Err(bamboo_agent_core::AgentError::LLM(format!(
412 "direct execution lease target {} does not match session {}",
413 lease.target_session_id, session.id
414 )));
415 }
416 let Some(router) = lease.router.take() else {
417 return self.execute_with_inputs(session, req, inputs).await;
418 };
419 let mut registration = lease.registration.take().ok_or_else(|| {
420 bamboo_agent_core::AgentError::LLM(
421 "direct execution lease is missing its router registration".to_string(),
422 )
423 })?;
424 if !router
425 .bind_run_lifetime(®istration.ownership(), session)
426 .await
427 {
428 return Err(bamboo_agent_core::AgentError::Tool(
429 "registered run Session lifetime differs".into(),
430 ));
431 }
432 if let Some(counted) = req.counted_loop.as_mut() {
433 counted.bind_owner(registration.ownership(), session);
434 }
435 if let Some(control) = req.loop_control.as_mut() {
436 control.bind_owner(registration.ownership(), session);
437 control.bind_transport_cleanup(
438 &mut registration,
439 self.storage().clone(),
440 self.session_inbox().cloned(),
441 self.persistence().root_actor_writer(),
442 );
443 }
444 let result = if req.loop_control.is_some() {
445 bamboo_agent_core::tools::observe_tool_suspension(false, async {
446 let token = bamboo_agent_core::tools::capture_suspension_restriction().ok_or_else(
447 || {
448 bamboo_agent_core::AgentError::Tool(
449 "direct loop_control execution is missing its host observation scope"
450 .into(),
451 )
452 },
453 )?;
454 registration.bind_tool_suspension_observation(token);
457 self.execute_with_inputs(session, req, inputs).await
458 })
459 .await
460 .0
461 } else {
462 self.execute_with_inputs(session, req, inputs).await
463 };
464 if registration.observed_forbidden_running() {
465 registration.quarantine().await;
470 return match result {
471 Err(error) => Err(error),
472 Ok(()) => Err(bamboo_agent_core::AgentError::Tool(
473 "suspension_disabled: detached Running outcome; stop unconfirmed".into(),
474 )),
475 };
476 }
477
478 let executed_admitted_generation = session
481 .session_inbox_admission()
482 .map_or(0, |state| state.last_admitted_sequence);
483 registration.begin_finalization().await;
484
485 let legacy_migration = crate::runtime::runner::state_bridge::migrate_legacy_pending_only(
486 session,
487 Some(self.storage()),
488 Some(self.persistence()),
489 self.session_inbox(),
490 )
491 .await;
492 if let Some(generation) = legacy_migration.highest_generation {
493 session.session_inbox_admission_mut().observe(generation);
494 }
495 let pending_generation = session
496 .session_inbox_admission()
497 .and_then(|state| state.pending_activation_generation());
498 if let Some(generation) = pending_generation {
499 let activation_ready = if let Some(inbox) = self.session_inbox() {
500 match inbox
501 .mark_activation_eligible(
502 &session.id,
503 generation,
504 bamboo_domain::SessionActivationPolicy::InterruptSpecificWait,
505 )
506 .await
507 {
508 Ok(()) => true,
509 Err(error) => {
510 tracing::error!(
511 session_id = %session.id,
512 %error,
513 "failed to persist direct SDK SessionInbox activation watermark"
514 );
515 false
516 }
517 }
518 } else {
519 false
520 };
521 if activation_ready {
522 if let Err(error) = bamboo_domain::SessionActivationPort::request_activation(
523 router.as_ref(),
524 &session.id,
525 generation,
526 )
527 .await
528 {
529 tracing::error!(
530 session_id = %session.id,
531 %error,
532 "failed to hand direct SDK SessionInbox generation to activation router"
533 );
534 }
535 }
536 }
537
538 if let Err(error) = self.persistence().checkpoint_runtime_session(session).await {
541 tracing::warn!(
542 session_id = %session.id,
543 %error,
544 "failed to checkpoint direct SDK terminal SessionInbox state"
545 );
546 }
547 if let Err(error) = registration.finish(executed_admitted_generation).await {
548 tracing::error!(
549 session_id = %session.id,
550 %error,
551 "direct SDK SessionInbox finalization failed"
552 );
553 }
554
555 result
556 }
557
558 pub fn storage(&self) -> &Arc<dyn bamboo_agent_core::storage::Storage> {
560 &self.runtime.storage
561 }
562
563 pub fn persistence(&self) -> &Arc<dyn RuntimeSessionPersistence> {
565 &self.runtime.persistence
566 }
567
568 pub fn session_inbox(&self) -> Option<&Arc<dyn bamboo_domain::SessionInboxPort>> {
569 self.runtime.session_inbox.as_ref()
570 }
571
572 pub async fn admit_session_inbox_at_safe_boundary(
577 &self,
578 session: &mut bamboo_agent_core::Session,
579 ) -> usize {
580 crate::runtime::runner::state_bridge::refresh_turn_boundary_with_inbox(
581 session,
582 Some(self.storage()),
583 Some(self.persistence()),
584 self.session_inbox(),
585 )
586 .await
587 .merged
588 }
589
590 pub async fn admit_session_inbox_at_safe_boundary_checked(
594 &self,
595 session: &mut bamboo_agent_core::Session,
596 ) -> Result<usize, bamboo_agent_core::AgentError> {
597 let refresh = crate::runtime::runner::state_bridge::refresh_turn_boundary_with_inbox(
598 session,
599 Some(self.storage()),
600 Some(self.persistence()),
601 self.session_inbox(),
602 )
603 .await;
604 if let Some(error) = refresh.admission_error {
605 return Err(bamboo_agent_core::AgentError::Tool(error));
606 }
607 Ok(refresh.merged)
608 }
609
610 pub fn activation_router(
611 &self,
612 ) -> Option<&Arc<crate::session_activation::SessionActivationRouter>> {
613 self.runtime.activation_router.as_ref()
614 }
615
616 pub fn session_messenger(&self) -> Option<&Arc<crate::SessionMessenger>> {
617 self.runtime.session_messenger.as_ref()
618 }
619
620 pub fn default_tools(&self) -> &Arc<dyn bamboo_agent_core::tools::ToolExecutor> {
630 &self.runtime.default_tools
631 }
632}
633
634pub struct AgentBuilder {
642 inner: AgentRuntimeBuilder,
643}
644
645impl AgentBuilder {
646 pub fn new() -> Self {
647 Self {
648 inner: AgentRuntimeBuilder::new(),
649 }
650 }
651
652 pub fn storage(mut self, v: Arc<dyn bamboo_agent_core::storage::Storage>) -> Self {
653 self.inner = self.inner.storage(v);
654 self
655 }
656
657 pub fn persistence(mut self, v: Arc<dyn RuntimeSessionPersistence>) -> Self {
658 self.inner = self.inner.persistence(v);
659 self
660 }
661
662 pub fn session_inbox(mut self, v: Arc<dyn bamboo_domain::SessionInboxPort>) -> Self {
663 self.inner = self.inner.session_inbox(v);
664 self
665 }
666
667 pub fn activation_router(
668 mut self,
669 v: Arc<crate::session_activation::SessionActivationRouter>,
670 ) -> Self {
671 self.inner = self.inner.activation_router(v);
672 self
673 }
674
675 pub fn session_messenger(mut self, v: Arc<crate::SessionMessenger>) -> Self {
676 self.inner = self.inner.session_messenger(v);
677 self
678 }
679
680 pub fn attachment_reader(
681 mut self,
682 v: Arc<dyn bamboo_agent_core::storage::AttachmentReader>,
683 ) -> Self {
684 self.inner = self.inner.attachment_reader(v);
685 self
686 }
687
688 pub fn skill_manager(mut self, v: Arc<bamboo_skills::SkillManager>) -> Self {
689 self.inner = self.inner.skill_manager(v);
690 self
691 }
692
693 pub fn project_context_resolver(
694 mut self,
695 v: Arc<crate::project_context::ProjectContextResolver>,
696 ) -> Self {
697 self.inner = self.inner.project_context_resolver(v);
698 self
699 }
700
701 pub fn metrics_collector(mut self, v: bamboo_metrics::MetricsCollector) -> Self {
702 self.inner = self.inner.metrics_collector(v);
703 self
704 }
705
706 pub fn config(mut self, v: Arc<tokio::sync::RwLock<bamboo_llm::Config>>) -> Self {
707 self.inner = self.inner.config(v);
708 self
709 }
710
711 pub fn model_limits_section(
714 mut self,
715 section: Arc<bamboo_config::LiveSection<bamboo_config::ModelLimitsSection>>,
716 ) -> Self {
717 self.inner = self.inner.model_limits_section(section);
718 self
719 }
720
721 pub fn permission_config(mut self, v: Arc<bamboo_tools::permission::PermissionConfig>) -> Self {
722 self.inner = self.inner.permission_config(v);
723 self
724 }
725
726 pub fn permission_mode(mut self, v: bamboo_domain::PermissionMode) -> Self {
727 self.inner = self.inner.permission_mode(v);
728 self
729 }
730
731 pub fn provider(mut self, v: Arc<dyn bamboo_llm::LLMProvider>) -> Self {
732 self.inner = self.inner.provider(v);
733 self
734 }
735
736 pub fn memory_store(mut self, v: bamboo_memory::memory_store::MemoryStore) -> Self {
737 self.inner = self.inner.memory_store(v);
738 self
739 }
740
741 pub fn default_tools(mut self, v: Arc<dyn bamboo_agent_core::tools::ToolExecutor>) -> Self {
742 self.inner = self.inner.default_tools(v);
743 self
744 }
745
746 pub fn hook_runner(mut self, v: Arc<crate::runtime::HookRunner>) -> Self {
748 self.inner = self.inner.hook_runner(v);
749 self
750 }
751
752 pub fn build(self) -> Result<Agent, &'static str> {
753 let runtime = self.inner.build()?;
754 Ok(Agent {
755 runtime: Arc::new(runtime),
756 })
757 }
758}
759
760impl Default for AgentBuilder {
761 fn default() -> Self {
762 Self::new()
763 }
764}