funera_orchestrate/
runtime.rs

1use std::marker::PhantomData;
2#[cfg(feature = "skill")]
3use std::path::PathBuf;
4use std::sync::Arc;
5
6use async_openai::config::OpenAIConfig;
7use tokio::sync::{RwLock, broadcast, mpsc};
8use tokio::task::JoinHandle;
9
10#[cfg(test)]
11use funera_core::chat::session::FuneraSession;
12use funera_core::chat::session::{SessionCmd, spawn_session_actor};
13use funera_core::env::{FuneraEnv, FuneraEnvWatcher};
14use funera_core::event_bus::env_state_bus::EnvStateEvent;
15#[cfg(feature = "tool")]
16use funera_core::event_bus::tool_bus::ToolBus;
17use funera_core::provider::ChatProvider;
18#[cfg(feature = "deepseek")]
19use funera_core::provider::deepseek::DeepSeekProvider;
20#[cfg(feature = "skill")]
21use funera_core::re_act::skills::{Skill, SkillRegistry};
22#[cfg(feature = "tool")]
23use funera_core::re_act::tool::{Tool, ToolRegistry};
24#[cfg(feature = "tool")]
25use funera_core::re_act::tool_executor::ToolExecutor;
26#[cfg(feature = "security")]
27use funera_core::security::audit::{AuditBus, AuditEvent};
28#[cfg(all(feature = "sandbox", feature = "security"))]
29use funera_core::security::path_guard::PathGuard;
30#[cfg(feature = "security")]
31use funera_core::security::policy::ToolPolicy;
32#[cfg(feature = "security")]
33use funera_core::security::registry::ApprovalCallback;
34#[cfg(feature = "sandbox")]
35use funera_core::security::sandbox::SandboxPolicy;
36#[cfg(feature = "security")]
37use funera_core::security::secret::SecureApiKey;
38
39#[cfg(feature = "middleware")]
40use crate::event::AgentEvent;
41#[cfg(feature = "middleware")]
42use crate::middleware_bundle::MiddlewareBundle;
43#[cfg(feature = "middleware")]
44use funera_core::middleware::{ErrorsEnabled, MiddlewareChain};
45
46use crate::error::OrchestrateError;
47
48/// Builds an [`AgentRuntime`].
49///
50/// ```rust,no_run
51/// # use funera_orchestrate::{AgentRuntime, DeepSeekProvider};
52/// # fn example() -> Result<(), Box<dyn std::error::Error>> {
53/// let runtime = AgentRuntime::<DeepSeekProvider>::builder()
54///     .api_key(std::env::var("DEEPSEEK_API_KEY")?)
55///     .model("deepseek-v4-flash")
56///     .build()?;
57/// # Ok(())
58/// # }
59/// ```
60pub struct AgentRuntimeBuilder {
61    api_key: Option<String>,
62    base_url: Option<String>,
63    client: Option<async_openai::Client<OpenAIConfig>>,
64    model: Option<String>,
65    max_iterations: usize,
66    channel_buffer: usize,
67    #[cfg(feature = "tool")]
68    tools: Vec<Box<dyn Tool>>,
69    #[cfg(feature = "skill")]
70    skills: Vec<Skill>,
71    #[cfg(feature = "skill")]
72    skill_names_to_activate: Vec<String>,
73    #[cfg(feature = "skill")]
74    load_default_skills: bool,
75    #[cfg(feature = "sandbox")]
76    sandbox_policy: Option<SandboxPolicy>,
77    #[cfg(feature = "security")]
78    tool_policy: Option<ToolPolicy>,
79    #[cfg(feature = "security")]
80    secure_api_key: Option<SecureApiKey>,
81    #[cfg(feature = "security")]
82    approval_callback: Option<ApprovalCallback>,
83    #[cfg(feature = "security")]
84    approval_timeout: Option<std::time::Duration>,
85    #[cfg(feature = "middleware")]
86    middleware_bundle: Option<MiddlewareBundle<AgentEvent>>,
87}
88
89impl Default for AgentRuntimeBuilder {
90    fn default() -> Self {
91        Self::new()
92    }
93}
94
95impl AgentRuntimeBuilder {
96    pub fn new() -> Self {
97        Self {
98            api_key: None,
99            base_url: None,
100            client: None,
101            model: None,
102            max_iterations: 10,
103            channel_buffer: 32,
104            #[cfg(feature = "tool")]
105            tools: Vec::new(),
106            #[cfg(feature = "skill")]
107            skills: Vec::new(),
108            #[cfg(feature = "skill")]
109            skill_names_to_activate: Vec::new(),
110            #[cfg(feature = "skill")]
111            load_default_skills: false,
112            #[cfg(feature = "sandbox")]
113            sandbox_policy: None,
114            #[cfg(feature = "security")]
115            tool_policy: None,
116            #[cfg(feature = "security")]
117            secure_api_key: None,
118            #[cfg(feature = "security")]
119            approval_callback: None,
120            #[cfg(feature = "security")]
121            approval_timeout: None,
122            #[cfg(feature = "middleware")]
123            middleware_bundle: None,
124        }
125    }
126
127    /// OpenAI API key. Falls back to `OPENAI_API_KEY` env var.
128    pub fn api_key(mut self, key: impl Into<String>) -> Self {
129        let key = key.into();
130        #[cfg(feature = "security")]
131        {
132            self.secure_api_key = Some(SecureApiKey::new(key.clone()));
133        }
134        self.api_key = Some(key);
135        self
136    }
137
138    /// Custom base URL (proxy, compatible API, etc.).
139    /// Pass e.g. `std::env::var("OPENAI_BASE_URL").ok()`.
140    pub fn base_url(mut self, url: Option<String>) -> Self {
141        if let Some(u) = url {
142            self.base_url = Some(u);
143        }
144        self
145    }
146
147    /// LLM model name. Falls back to `OPENAI_MODEL` env var, then `"gpt-4o"`.
148    pub fn model(mut self, model: impl Into<String>) -> Self {
149        self.model = Some(model.into());
150        self
151    }
152
153    /// Directly provide an OpenAI client (overrides api_key + base_url).
154    pub fn client(mut self, client: async_openai::Client<OpenAIConfig>) -> Self {
155        self.client = Some(client);
156        self
157    }
158
159    /// Maximum number of ReAct iterations per call (default 10).
160    pub fn max_iterations(mut self, n: usize) -> Self {
161        self.max_iterations = n;
162        self
163    }
164
165    /// Internal channel buffer size (default 32).
166    pub fn channel_buffer(mut self, n: usize) -> Self {
167        self.channel_buffer = n;
168        self
169    }
170
171    /// Load a skill from a SKILL.md file.
172    #[cfg(feature = "skill")]
173    pub fn with_skill_file(mut self, path: impl Into<PathBuf>) -> Self {
174        let path = path.into();
175        match Skill::from_file(&path) {
176            Ok(skill) => {
177                self.skills.push(skill);
178            }
179            Err(e) => {
180                eprintln!("warn: failed to load skill from {:?}: {}", path, e);
181            }
182        }
183        self
184    }
185
186    /// Load all SKILL.md files from a directory.
187    #[cfg(feature = "skill")]
188    pub fn with_skills_dir(mut self, path: impl Into<PathBuf>) -> Self {
189        let path = path.into();
190        match Skill::from_dir(&path) {
191            Ok(skills) => self.skills.extend(skills),
192            Err(e) => {
193                eprintln!("warn: failed to load skills from {:?}: {}", path, e);
194            }
195        }
196        self
197    }
198
199    /// Register an inline skill definition.
200    #[cfg(feature = "skill")]
201    pub fn with_skill(
202        mut self,
203        name: impl Into<String>,
204        description: impl Into<String>,
205        content: impl Into<String>,
206    ) -> Self {
207        self.skills.push(Skill::new(name, description, content));
208        self
209    }
210
211    /// Activate a previously loaded skill by name.
212    /// If the skill does not exist, the call is silently ignored.
213    #[cfg(feature = "skill")]
214    pub fn with_skill_active(mut self, name: impl Into<String>) -> Self {
215        self.skill_names_to_activate.push(name.into());
216        self
217    }
218
219    /// Auto-discover and load skills from default paths
220    /// (`$SKILLS_HOME` or `~/.agents/skills/`), then activate them.
221    #[cfg(feature = "skill")]
222    pub fn with_skills_default_path(mut self) -> Self {
223        self.load_default_skills = true;
224        self
225    }
226
227    /// Register a tool by its type (requires `Tool + Default`).
228    #[cfg(feature = "tool")]
229    pub fn with_tool<T: Tool + Default + 'static>(mut self) -> Self {
230        self.tools.push(Box::new(T::default()));
231        self
232    }
233
234    /// Register a pre-constructed tool.
235    #[cfg(feature = "tool")]
236    pub fn with_tool_instance(mut self, tool: Box<dyn Tool>) -> Self {
237        self.tools.push(tool);
238        self
239    }
240
241    /// Attach a middleware chain with error channel.
242    ///
243    /// The runtime will spawn a task to consume inspector errors via `tracing::warn`.
244    #[cfg(feature = "middleware")]
245    pub fn with_middleware_bundle(mut self, bundle: MiddlewareBundle<AgentEvent>) -> Self {
246        self.middleware_bundle = Some(bundle);
247        self
248    }
249
250    /// Set a kernel-enforced sandbox policy for tool subprocesses.
251    ///
252    /// When enabled, tool subprocesses are isolated via Landlock
253    /// (Linux 5.13+), Seatbelt (macOS), or Write-Restricted Token
254    /// (Windows 8+). Unsupported configurations gracefully degrade
255    /// without full isolation.
256    #[cfg(feature = "sandbox")]
257    pub fn with_sandbox_policy(mut self, policy: SandboxPolicy) -> Self {
258        self.sandbox_policy = Some(policy);
259        self
260    }
261
262    /// Set an application-level tool policy for controlling which tools
263    /// are allowed/denied, shell command restrictions, argument size
264    /// limits, timeout bounds, and working directory restrictions.
265    ///
266    /// The policy is enforced by the guarded tool registry before each
267    /// tool call.  Combine with [`with_sandbox_policy`](Self::with_sandbox_policy)
268    /// for defence-in-depth (application-level + kernel-enforced isolation).
269    ///
270    /// # Example
271    ///
272    /// ```rust,no_run
273    /// # use funera_orchestrate::{AgentRuntimeBuilder, ToolPolicy, ShellPolicy};
274    /// # fn example() -> Result<(), Box<dyn std::error::Error>> {
275    /// let policy = ToolPolicy {
276    ///     denied_tools: ["shell".into()].into_iter().collect(),
277    ///     shell_policy: Some(ShellPolicy::strict()),
278    ///     ..ToolPolicy::default()
279    /// };
280    ///
281    /// let runtime = AgentRuntimeBuilder::new()
282    ///     .api_key(std::env::var("DEEPSEEK_API_KEY")?)
283    ///     .model("deepseek-v4-flash")
284    ///     .with_tool_policy(policy)
285    ///     .build()?;
286    /// # Ok(())
287    /// # }
288    /// ```
289    #[cfg(feature = "security")]
290    pub fn with_tool_policy(mut self, policy: ToolPolicy) -> Self {
291        self.tool_policy = Some(policy);
292        self
293    }
294
295    /// Register a notification callback fired when a tool call requires user approval.
296    ///
297    /// When a tool call targets a path outside the [`PathGuard`] trusted zone but
298    /// within the sandbox boundary, the registry pauses execution and invokes this
299    /// callback with `(call_id, tool_name, reason)`. **This is a notification only — do not
300    /// call [`AgentRuntime::approve_tool_call`] inside this callback.** Instead, store the
301    /// call_id and call `approve_tool_call` from your own async context (e.g. a channel
302    /// receiver task or an event handler).
303    ///
304    /// Requires the `security` feature.
305    #[cfg(feature = "security")]
306    pub fn on_approval_required(
307        mut self,
308        cb: impl Fn(Arc<str>, String, String) + Send + Sync + 'static,
309    ) -> Self {
310        self.approval_callback = Some(std::sync::Arc::new(
311            move |call_id: &str, tool_name: &str, reason: &str, _paths: &[std::path::PathBuf]| {
312                cb(
313                    Arc::from(call_id),
314                    tool_name.to_string(),
315                    reason.to_string(),
316                );
317            },
318        ));
319        self
320    }
321
322    /// Set a timeout for tool call approval. If not set, the registry waits indefinitely.
323    /// When the timeout elapses, the tool call is automatically rejected.
324    ///
325    /// Requires the `security` feature.
326    #[cfg(feature = "security")]
327    pub fn with_approval_timeout(mut self, timeout: std::time::Duration) -> Self {
328        self.approval_timeout = Some(timeout);
329        self
330    }
331
332    /// Register all builtin tools (Read, Write, Edit, Shell).
333    /// Requires the `funera-builtin-tools` feature.
334    ///
335    /// If a sandbox policy was configured via [`with_sandbox_policy`],
336    /// the `shell` tool will apply kernel-level isolation to each subprocess.
337    #[cfg(feature = "funera-builtin-tools")]
338    pub fn with_builtin_tools(mut self) -> Self {
339        use funera_builtin_tools::{EditTool, ReadTool, ShellTool, WriteTool};
340        self.tools.push(Box::new(ReadTool));
341        self.tools.push(Box::new(WriteTool));
342        self.tools.push(Box::new(EditTool));
343        #[cfg(feature = "sandbox")]
344        if let Some(ref policy) = self.sandbox_policy {
345            self.tools
346                .push(Box::new(ShellTool::with_sandbox(policy.clone())));
347        } else {
348            self.tools.push(Box::new(ShellTool::new()));
349        }
350        #[cfg(not(feature = "sandbox"))]
351        self.tools.push(Box::new(ShellTool::new()));
352        self
353    }
354
355    /// Build the runtime with the default DeepSeek provider.
356    ///
357    /// Spawns a background `ToolExecutor` task that lives for the runtime's
358    /// lifetime and processes tool calls from the ReAct loop.
359    #[cfg(feature = "deepseek")]
360    pub fn build(self) -> Result<AgentRuntime<DeepSeekProvider>, OrchestrateError> {
361        self.build_with::<DeepSeekProvider>()
362    }
363
364    /// Build the runtime with a custom LLM provider.
365    ///
366    /// ```rust,no_run
367    /// # use funera_orchestrate::{AgentRuntime, DeepSeekProvider};
368    /// # fn example() -> Result<(), Box<dyn std::error::Error>> {
369    /// let rt = AgentRuntime::<DeepSeekProvider>::builder().api_key(std::env::var("DEEPSEEK_API_KEY")?).model("deepseek-v4-flash").build_with::<DeepSeekProvider>()?;
370    /// # Ok(())
371    /// # }
372    /// ```
373    pub fn build_with<P: ChatProvider>(
374        #[allow(unused_mut)] mut self,
375    ) -> Result<AgentRuntime<P>, OrchestrateError> {
376        #[cfg(feature = "security")]
377        let api_key = {
378            self.secure_api_key
379                .take()
380                .map(|k| k.expose_secret().to_string())
381                .or_else(|| self.api_key.take())
382                .or_else(|| std::env::var("OPENAI_API_KEY").ok())
383        };
384        #[cfg(not(feature = "security"))]
385        let api_key = self
386            .api_key
387            .or_else(|| std::env::var("OPENAI_API_KEY").ok());
388        let model = self
389            .model
390            .or_else(|| std::env::var("OPENAI_MODEL").ok())
391            .unwrap_or_else(|| "gpt-4o".into());
392
393        let client = match self.client {
394            Some(c) => c,
395            None => {
396                let key = api_key.ok_or_else(|| {
397                    OrchestrateError::Config(
398                        "no API key; set OPENAI_API_KEY or call .api_key()".into(),
399                    )
400                })?;
401                let mut cfg = OpenAIConfig::default().with_api_key(key);
402                if let Some(url) = &self.base_url {
403                    cfg = cfg.with_api_base(url);
404                }
405                async_openai::Client::with_config(cfg)
406            }
407        };
408
409        // Sync sandbox policy into tool policy when only sandbox was configured.
410        #[cfg(all(feature = "sandbox", feature = "security"))]
411        {
412            if self.tool_policy.is_none()
413                && let Some(ref sp) = self.sandbox_policy
414            {
415                self.tool_policy = Some(ToolPolicy {
416                    sandbox: sp.clone(),
417                    ..Default::default()
418                });
419            }
420        }
421
422        #[cfg(feature = "security")]
423        let audit_bus = AuditBus::default();
424
425        #[cfg(feature = "tool")]
426        let registry = {
427            #[cfg(feature = "security")]
428            let mut reg = match self.tool_policy {
429                Some(ref policy) => ToolRegistry::new_from_policy(policy.clone()),
430                None => ToolRegistry::new(),
431            };
432            #[cfg(not(feature = "security"))]
433            let mut reg = ToolRegistry::new();
434            for t in self.tools {
435                reg.add_tool(t);
436            }
437
438            // ── Security wiring ─────────────────────────────────
439            #[cfg(feature = "security")]
440            reg.set_audit_bus(audit_bus.clone());
441
442            #[cfg(all(feature = "sandbox", feature = "security"))]
443            if let Some(ref sp) = self.sandbox_policy {
444                if sp.enabled && (!sp.read_paths.is_empty() || !sp.read_write_paths.is_empty()) {
445                    let all_paths: Vec<_> = sp
446                        .read_paths
447                        .iter()
448                        .chain(sp.read_write_paths.iter())
449                        .cloned()
450                        .collect();
451                    if !all_paths.is_empty() {
452                        let path_guard = PathGuard::new(all_paths.iter().map(|p| p.as_path()));
453                        reg.set_path_guard(path_guard);
454                    }
455                }
456                reg.set_sandbox_paths(sp.read_paths.clone(), sp.read_write_paths.clone());
457            }
458
459            #[cfg(feature = "security")]
460            {
461                if let Some(ref cb) = self.approval_callback {
462                    reg.set_approval_callback(cb.clone());
463                }
464                if let Some(dur) = self.approval_timeout {
465                    reg.set_approval_timeout(Some(dur));
466                }
467            }
468            // ── End security wiring ─────────────────────────────
469
470            reg
471        };
472
473        #[cfg(feature = "skill")]
474        let mut skill_registry = SkillRegistry::new();
475
476        #[cfg(feature = "skill")]
477        {
478            if self.load_default_skills {
479                let default_skills = Skill::from_default_path();
480                for skill in default_skills {
481                    let name = skill.name.clone();
482                    skill_registry.add(skill);
483                    self.skill_names_to_activate.push(name);
484                }
485            }
486            for skill in self.skills {
487                skill_registry.add(skill);
488            }
489            for name in &self.skill_names_to_activate {
490                skill_registry.activate(name);
491            }
492        }
493
494        let (env, env_watcher) = FuneraEnv::new(client, &model);
495
496        #[cfg(feature = "sandbox")]
497        let env = if let Some(ref sp) = self.sandbox_policy {
498            env.with_sandbox_policy(sp.clone())
499        } else {
500            env
501        };
502
503        #[cfg(feature = "tool")]
504        let env = env.with_tool_registry(registry);
505        #[cfg(feature = "skill")]
506        let env = env.with_skill_registry(skill_registry);
507
508        let (env_state_tx, _) = broadcast::channel(32);
509
510        #[cfg(feature = "tool")]
511        if let Ok(guard) = env.tool_registry.try_read() {
512            let tools = guard.get_all_tools();
513            for name in tools.keys() {
514                let _ = env_state_tx.send(EnvStateEvent::ToolAdded(name.clone()));
515            }
516        }
517
518        #[cfg(feature = "skill")]
519        if let Ok(guard) = env.skill_registry.try_read() {
520            let skills = guard.all_skills();
521            for name in skills.keys() {
522                let _ = env_state_tx.send(EnvStateEvent::SkillAdded(name.clone()));
523            }
524        }
525
526        #[cfg(feature = "middleware")]
527        let middleware_chain = if let Some(bundle) = self.middleware_bundle.take() {
528            let MiddlewareBundle { chain, error_rx } = bundle;
529            tokio::spawn(async move {
530                let mut rx = error_rx;
531                while let Some((name, err)) = rx.recv().await {
532                    tracing::warn!("[middleware:{name}] inspector error: {err}");
533                }
534            });
535            Arc::new(chain)
536        } else {
537            let (chain, error_rx) = MiddlewareChain::<AgentEvent>::new().activate_error_channel();
538            tokio::spawn(async move {
539                let mut rx = error_rx;
540                while let Some((name, err)) = rx.recv().await {
541                    tracing::warn!("[middleware:{name}] inspector error: {err}");
542                }
543            });
544            Arc::new(chain)
545        };
546
547        #[cfg(feature = "tool")]
548        let (tool_bus, exec_rx) = ToolBus::new();
549        #[cfg(feature = "tool")]
550        let reg = env.tool_registry.clone();
551        #[cfg(feature = "tool")]
552        let handle = tokio::spawn(async move {
553            ToolExecutor::new(reg, exec_rx).run().await;
554        });
555
556        let session_tx = spawn_session_actor();
557
558        #[cfg(feature = "security")]
559        let tool_policy_val = self.tool_policy.clone().unwrap_or_default();
560
561        Ok(AgentRuntime::<P> {
562            env,
563            env_watcher,
564            #[cfg(feature = "tool")]
565            tool_bus,
566            model,
567            max_iterations: self.max_iterations,
568            channel_buffer: self.channel_buffer,
569            env_state_tx,
570            #[cfg(feature = "tool")]
571            _executor_handle: handle,
572            session_tx,
573            _state: PhantomData,
574            _phantom: PhantomData,
575            #[cfg(feature = "middleware")]
576            middleware_chain,
577            #[cfg(feature = "security")]
578            tool_policy: tool_policy_val,
579            #[cfg(feature = "security")]
580            audit_bus,
581        })
582    }
583}
584
585/// A runtime context for executing agent interactions.
586///
587/// Marker type-state: the runtime is available for a `send`/`send_stream` call.
588pub struct Idle;
589
590/// Marker type-state: a `send`/`send_stream` call is in progress.
591pub struct Acquired;
592
593/// `AgentRuntime` owns the shared infrastructure (LLM client, tool registry,
594/// tool executor) and a persistent session (backed by a session actor).
595///
596/// The generic parameter `S` is a type-state marker — [`Idle`] means
597/// no `send`/`send_stream` is in progress, [`Acquired`] means one is active.
598/// Send operations consume `AgentRuntime<P, Idle>` and return a handle that
599/// eventually yields back `AgentRuntime<P, Idle>`.
600pub struct AgentRuntime<P: ChatProvider, S = Idle> {
601    env: FuneraEnv,
602    pub(crate) env_watcher: FuneraEnvWatcher,
603    #[cfg(feature = "tool")]
604    pub(crate) tool_bus: ToolBus,
605    pub(crate) model: String,
606    pub(crate) max_iterations: usize,
607    pub(crate) channel_buffer: usize,
608    env_state_tx: broadcast::Sender<EnvStateEvent>,
609    #[cfg(feature = "tool")]
610    _executor_handle: JoinHandle<()>,
611    pub(crate) session_tx: mpsc::UnboundedSender<SessionCmd>,
612    _state: PhantomData<S>,
613    _phantom: PhantomData<fn() -> P>,
614    #[cfg(feature = "middleware")]
615    middleware_chain: Arc<MiddlewareChain<AgentEvent, ErrorsEnabled>>,
616    #[cfg(feature = "security")]
617    tool_policy: ToolPolicy,
618    #[cfg(feature = "security")]
619    audit_bus: AuditBus,
620}
621
622// ── All state markers share these methods ─────────────────────
623
624impl<P: ChatProvider, S> AgentRuntime<P, S> {
625    /// Create a new builder.
626    pub fn builder() -> AgentRuntimeBuilder {
627        AgentRuntimeBuilder::new()
628    }
629
630    /// Reset the conversation session (clear message history).
631    pub fn reset(&self) {
632        let _ = self.session_tx.send(SessionCmd::Clear);
633    }
634
635    /// Access the session control channel.
636    pub fn session_tx(&self) -> mpsc::UnboundedSender<SessionCmd> {
637        self.session_tx.clone()
638    }
639
640    /// The LLM model name configured for this runtime.
641    pub fn model(&self) -> &str {
642        &self.model
643    }
644
645    /// Maximum ReAct iterations per call.
646    pub fn max_iterations(&self) -> usize {
647        self.max_iterations
648    }
649
650    /// Channel buffer size.
651    pub fn channel_buffer(&self) -> usize {
652        self.channel_buffer
653    }
654
655    /// Clone the env watcher for a session.
656    pub(crate) fn env_watcher(&self) -> FuneraEnvWatcher {
657        self.env_watcher.clone()
658    }
659
660    /// Subscribe to runtime-level environment state events.
661    ///
662    /// The returned receiver yields [`EnvStateEvent`] notifications about
663    /// tool/skill registration changes, LLM model changes, etc. that occur
664    /// during the runtime's lifetime.
665    ///
666    /// Unlike [`Agent::subscribe_raw_events`](crate::Agent::subscribe_raw_events)
667    /// which only delivers events during a `fire`/`send` call, this subscription
668    /// is persistent and independent of any agent call.
669    pub fn subscribe_env_state(&self) -> broadcast::Receiver<EnvStateEvent> {
670        self.env_state_tx.subscribe()
671    }
672
673    /// Subscribe to security audit events.
674    ///
675    /// The returned receiver yields [`AuditEvent`] notifications for every
676    /// tool execution, denial, policy violation, and sandbox action. This is
677    /// an independent, persistent subscription that is not tied to any
678    /// particular agent call.
679    ///
680    /// Requires the `security` feature.
681    #[cfg(feature = "security")]
682    pub fn subscribe_audit(&self) -> broadcast::Receiver<AuditEvent> {
683        self.audit_bus.subscribe()
684    }
685
686    /// Access the middleware chain for event filtering.
687    #[cfg(feature = "middleware")]
688    pub fn middleware_chain(&self) -> Arc<MiddlewareChain<AgentEvent, ErrorsEnabled>> {
689        self.middleware_chain.clone()
690    }
691
692    /// Transform the runtime into `Acquired` state (internal use).
693    pub(crate) fn into_acquired(self) -> AgentRuntime<P, Acquired> {
694        AgentRuntime::<P, Acquired> {
695            env: self.env,
696            env_watcher: self.env_watcher,
697            #[cfg(feature = "tool")]
698            tool_bus: self.tool_bus,
699            model: self.model,
700            max_iterations: self.max_iterations,
701            channel_buffer: self.channel_buffer,
702            env_state_tx: self.env_state_tx,
703            #[cfg(feature = "tool")]
704            _executor_handle: self._executor_handle,
705            session_tx: self.session_tx,
706            _state: PhantomData,
707            _phantom: PhantomData,
708            #[cfg(feature = "middleware")]
709            middleware_chain: self.middleware_chain,
710            #[cfg(feature = "security")]
711            tool_policy: self.tool_policy,
712            #[cfg(feature = "security")]
713            audit_bus: self.audit_bus,
714        }
715    }
716
717    /// The tool registry (for dynamic tool management).
718    #[cfg(feature = "tool")]
719    pub fn tool_registry(&self) -> Arc<RwLock<ToolRegistry>> {
720        self.env.tool_registry.clone()
721    }
722
723    /// The skill registry (for dynamic skill management).
724    #[cfg(feature = "skill")]
725    pub fn skill_registry(&self) -> Arc<RwLock<SkillRegistry>> {
726        self.env.skill_registry.clone()
727    }
728
729    /// The sandbox policy configured for this runtime.
730    #[cfg(feature = "sandbox")]
731    pub fn sandbox_policy(&self) -> SandboxPolicy {
732        self.env.sandbox_policy().clone()
733    }
734
735    /// The application-level tool policy configured for this runtime.
736    ///
737    /// Returns the [`ToolPolicy`] that controls which tools are allowed,
738    /// shell command restrictions, argument size limits, timeout bounds,
739    /// and working directory restrictions.
740    #[cfg(feature = "security")]
741    pub fn tool_policy(&self) -> &ToolPolicy {
742        &self.tool_policy
743    }
744
745    /// Approve or reject a pending tool call that is awaiting user approval.
746    ///
747    /// Call this from your async context (channel receiver, event handler) using
748    /// the `call_id` received via [`on_approval_required`] or
749    /// [`AgentEvent::ToolApprovalRequired`].
750    ///
751    /// Returns `Ok(())` if the approval was delivered, or `Err(String)` if no
752    /// pending approval was found for the given `call_id`.
753    #[cfg(all(feature = "tool", feature = "security"))]
754    pub fn approve_tool_call(&self, call_id: &str, approved: bool) -> Result<(), String> {
755        self.tool_registry()
756            .blocking_read()
757            .approve_tool_call(call_id, approved)
758    }
759}
760
761// ── Acquired → Idle ─────────────────────────────────────────
762
763impl<P: ChatProvider> AgentRuntime<P, Acquired> {
764    pub(crate) fn into_idle(self) -> AgentRuntime<P, Idle> {
765        AgentRuntime::<P, Idle> {
766            env: self.env,
767            env_watcher: self.env_watcher,
768            #[cfg(feature = "tool")]
769            tool_bus: self.tool_bus,
770            model: self.model,
771            max_iterations: self.max_iterations,
772            channel_buffer: self.channel_buffer,
773            env_state_tx: self.env_state_tx,
774            #[cfg(feature = "tool")]
775            _executor_handle: self._executor_handle,
776            session_tx: self.session_tx,
777            _state: PhantomData,
778            _phantom: PhantomData,
779            #[cfg(feature = "middleware")]
780            middleware_chain: self.middleware_chain,
781            #[cfg(feature = "security")]
782            tool_policy: self.tool_policy,
783            #[cfg(feature = "security")]
784            audit_bus: self.audit_bus,
785        }
786    }
787}
788
789#[cfg(test)]
790mod tests {
791    use super::*;
792    use funera_core::chat::message::{FuneraMessage, MsgVariant, Role, TextMessage};
793
794    // ── builder defaults ───────────────────────────────────────────
795
796    #[test]
797    fn builder_defaults() {
798        let b = AgentRuntimeBuilder::new();
799        assert_eq!(b.max_iterations, 10);
800        assert_eq!(b.channel_buffer, 32);
801        assert!(b.api_key.is_none());
802        assert!(b.model.is_none());
803    }
804
805    #[cfg(feature = "tool")]
806    mod tool_tests {
807        use super::*;
808        use funera_core::re_act::tool::ToolCallError;
809
810        #[derive(Default)]
811        struct MockTool;
812
813        #[async_trait::async_trait]
814        impl Tool for MockTool {
815            fn name(&self) -> &str {
816                "mock_tool"
817            }
818            fn description(&self) -> &str {
819                "A mock tool for testing"
820            }
821            fn schema(&self) -> serde_json::Value {
822                serde_json::json!({})
823            }
824            async fn execute(&self, _args: serde_json::Value) -> Result<String, ToolCallError> {
825                Ok("ok".into())
826            }
827        }
828
829        #[test]
830        fn builder_defaults_tools_empty() {
831            let b = AgentRuntimeBuilder::new();
832            assert!(b.tools.is_empty());
833        }
834
835        #[test]
836        fn builder_with_tool_instance() {
837            let b = AgentRuntimeBuilder::new().with_tool_instance(Box::new(MockTool));
838            assert_eq!(b.tools.len(), 1);
839        }
840
841        #[tokio::test]
842        async fn build_with_tool_adds_to_registry() {
843            let rt = AgentRuntimeBuilder::new()
844                .api_key("sk-test")
845                .model("x")
846                .with_tool::<MockTool>()
847                .build()
848                .unwrap();
849            let registry = rt.tool_registry();
850            let guard = registry.read().await;
851            let tools = guard.get_all_tools();
852            assert!(tools.contains_key("mock_tool"));
853        }
854
855        #[tokio::test]
856        async fn tool_registry_accessor() {
857            let rt = AgentRuntimeBuilder::new()
858                .api_key("sk-test")
859                .model("x")
860                .build()
861                .unwrap();
862            let reg = rt.tool_registry();
863            let guard = reg.read().await;
864            let tools = guard.get_all_tools();
865            assert!(tools.is_empty());
866        }
867    }
868
869    #[cfg(feature = "skill")]
870    mod skill_tests {
871        use super::*;
872
873        #[test]
874        fn builder_with_skill_inline() {
875            let b = AgentRuntimeBuilder::new().with_skill("s1", "desc", "content");
876            assert_eq!(b.skills.len(), 1);
877            assert_eq!(b.skills[0].name, "s1");
878            assert_eq!(b.skills[0].description, "desc");
879            assert_eq!(b.skills[0].content, "content");
880        }
881
882        #[test]
883        fn builder_with_skill_active_adds_to_list() {
884            let b = AgentRuntimeBuilder::new()
885                .with_skill_active("s1")
886                .with_skill_active("s2");
887            assert_eq!(b.skill_names_to_activate, vec!["s1", "s2"]);
888        }
889
890        #[test]
891        fn builder_with_skills_default_path_sets_flag() {
892            let b = AgentRuntimeBuilder::new().with_skills_default_path();
893            assert!(b.load_default_skills);
894        }
895
896        #[test]
897        fn builder_skills_combined() {
898            let b = AgentRuntimeBuilder::new()
899                .with_skill("a", "", "aaa")
900                .with_skill("b", "", "bbb")
901                .with_skill_active("a");
902            assert_eq!(b.skills.len(), 2);
903            assert_eq!(b.skill_names_to_activate, vec!["a"]);
904        }
905    }
906
907    #[test]
908    fn builder_set_max_iterations() {
909        let b = AgentRuntimeBuilder::new().max_iterations(20);
910        assert_eq!(b.max_iterations, 20);
911    }
912
913    #[test]
914    fn builder_set_channel_buffer() {
915        let b = AgentRuntimeBuilder::new().channel_buffer(64);
916        assert_eq!(b.channel_buffer, 64);
917    }
918
919    #[test]
920    fn builder_set_model() {
921        let b = AgentRuntimeBuilder::new().model("test-model");
922        assert_eq!(b.model, Some("test-model".into()));
923    }
924
925    #[test]
926    fn builder_set_api_key() {
927        let b = AgentRuntimeBuilder::new().api_key("sk-test");
928        assert_eq!(b.api_key, Some("sk-test".into()));
929    }
930
931    #[test]
932    fn builder_set_base_url() {
933        let b = AgentRuntimeBuilder::new().base_url(Some("https://example.com".into()));
934        assert_eq!(b.base_url, Some("https://example.com".into()));
935    }
936
937    #[test]
938    fn builder_set_base_url_none_noop() {
939        let b = AgentRuntimeBuilder::new().base_url(None);
940        assert!(b.base_url.is_none());
941    }
942
943    #[test]
944    fn builder_set_client() {
945        let cfg = async_openai::config::OpenAIConfig::default();
946        let client = async_openai::Client::with_config(cfg);
947        let b = AgentRuntimeBuilder::new().client(client);
948        assert!(b.client.is_some());
949    }
950
951    // ── build ──────────────────────────────────────────────────────
952
953    #[tokio::test]
954    async fn build_with_explicit_key() {
955        let rt = AgentRuntimeBuilder::new()
956            .api_key("sk-test")
957            .model("test-model")
958            .build()
959            .expect("build should succeed with api_key");
960        assert_eq!(rt.model(), "test-model");
961        assert_eq!(rt.max_iterations(), 10);
962        assert_eq!(rt.channel_buffer(), 32);
963    }
964
965    #[tokio::test]
966    async fn build_custom_params() {
967        let rt = AgentRuntimeBuilder::new()
968            .api_key("sk-test")
969            .model("my-model")
970            .max_iterations(15)
971            .channel_buffer(8)
972            .build()
973            .unwrap();
974        assert_eq!(rt.model(), "my-model");
975        assert_eq!(rt.max_iterations(), 15);
976        assert_eq!(rt.channel_buffer(), 8);
977    }
978
979    #[tokio::test]
980    async fn build_fails_without_key() {
981        let has_key = std::env::var("OPENAI_API_KEY").is_ok();
982        if has_key {
983            // Can't test failure when key is present in env
984            return;
985        }
986        let result = AgentRuntimeBuilder::new().model("x").build();
987        assert!(matches!(result, Err(OrchestrateError::Config(_))));
988    }
989
990    #[tokio::test]
991    async fn build_model_fallback_default() {
992        let has_model = std::env::var("OPENAI_MODEL").is_ok();
993        if has_model {
994            return;
995        }
996        let rt = AgentRuntimeBuilder::new()
997            .api_key("sk-test")
998            .build()
999            .unwrap();
1000        assert_eq!(rt.model(), "gpt-4o");
1001    }
1002
1003    // ── session management ─────────────────────────────────────────
1004
1005    #[tokio::test]
1006    async fn session_actor_is_alive() {
1007        let rt = AgentRuntimeBuilder::new()
1008            .api_key("sk-test")
1009            .model("x")
1010            .build()
1011            .unwrap();
1012        let tx = rt.session_tx();
1013        assert!(tx.send(SessionCmd::Clear).is_ok());
1014    }
1015
1016    #[tokio::test]
1017    async fn session_context_works_immediately() {
1018        let rt = AgentRuntimeBuilder::new()
1019            .api_key("sk-test")
1020            .model("x")
1021            .build()
1022            .unwrap();
1023        let ctx = FuneraSession::new(rt.session_tx()).session_context().await;
1024        assert!(ctx.is_empty());
1025    }
1026
1027    #[tokio::test]
1028    async fn reset_clears_messages() {
1029        let rt = AgentRuntimeBuilder::new()
1030            .api_key("sk-test")
1031            .model("x")
1032            .build()
1033            .unwrap();
1034        let session = FuneraSession::new(rt.session_tx());
1035        session.push_message(FuneraMessage::new(
1036            Role::User,
1037            MsgVariant::Text(TextMessage {
1038                text: "hi".into(),
1039                reasoning_content: None,
1040            }),
1041        ));
1042        let ctx_before = session.session_context().await;
1043        assert_eq!(ctx_before.len(), 1);
1044
1045        rt.reset();
1046
1047        let ctx_after = session.session_context().await;
1048        assert_eq!(ctx_after.len(), 0);
1049    }
1050
1051    #[tokio::test]
1052    async fn subscribe_env_state_works() {
1053        let rt = AgentRuntimeBuilder::new()
1054            .api_key("sk-test")
1055            .model("x")
1056            .build()
1057            .unwrap();
1058        let mut rx = rt.subscribe_env_state();
1059        // Send an event after subscribing to verify the channel works
1060        rt.env_state_tx
1061            .send(EnvStateEvent::LlmChanged("new-model".into()))
1062            .unwrap();
1063        let got = tokio::time::timeout(std::time::Duration::from_secs(1), rx.recv()).await;
1064        assert!(matches!(
1065            got,
1066            Ok(Ok(EnvStateEvent::LlmChanged(m))) if m == "new-model"
1067        ));
1068    }
1069
1070    // ── sandbox integration tests ───────────────────────────────────
1071
1072    #[cfg(feature = "sandbox")]
1073    #[tokio::test]
1074    async fn builder_sandbox_policy_flows_to_env() {
1075        use funera_core::security::sandbox::SandboxPolicy;
1076
1077        let custom_policy = SandboxPolicy {
1078            read_write_paths: vec!["/project".into()],
1079            block_network: true,
1080            ..Default::default()
1081        };
1082
1083        let rt = AgentRuntimeBuilder::new()
1084            .api_key("sk-test")
1085            .model("x")
1086            .with_sandbox_policy(custom_policy.clone())
1087            .build()
1088            .unwrap();
1089
1090        let stored = rt.sandbox_policy();
1091        assert_eq!(stored.read_write_paths, custom_policy.read_write_paths);
1092        assert_eq!(stored.block_network, custom_policy.block_network);
1093        assert!(stored.enabled);
1094    }
1095
1096    #[cfg(feature = "sandbox")]
1097    #[tokio::test]
1098    async fn builder_no_sandbox_uses_default() {
1099        let rt = AgentRuntimeBuilder::new()
1100            .api_key("sk-test")
1101            .model("x")
1102            .build()
1103            .unwrap();
1104        let stored = rt.sandbox_policy();
1105        // Default policy is enabled with network blocked and empty paths
1106        assert!(stored.enabled);
1107        assert!(stored.block_network);
1108        assert!(stored.read_paths.is_empty());
1109        assert!(stored.read_write_paths.is_empty());
1110        assert!(stored.execute_paths.is_empty());
1111    }
1112
1113    #[cfg(feature = "sandbox")]
1114    #[tokio::test]
1115    async fn builder_sandbox_with_custom_environments() {
1116        use funera_core::security::sandbox::SandboxPolicy;
1117
1118        // Test that a disabled sandbox policy flows correctly
1119        let rt = AgentRuntimeBuilder::new()
1120            .api_key("sk-test")
1121            .model("x")
1122            .with_sandbox_policy(SandboxPolicy::disabled())
1123            .build()
1124            .unwrap();
1125
1126        let stored = rt.sandbox_policy();
1127        assert!(!stored.enabled, "disabled policy should stay disabled");
1128    }
1129}
funera_orchestrate/runtime.rs

funera_orchestrate/
runtime.rs
