runifold_agent/
agent.rs

1use std::{collections::BTreeMap, future::Future, pin::Pin, sync::Arc};
2
3use futures_util::{
4    StreamExt,
5    future::{Either, select},
6};
7use runifold_core::{
8    Budget, BudgetEvent, BudgetTracker, CapabilitySet, DomainEvent, EffectId, EffectKind,
9    EffectRequest, EventId, Instant, InvocationId, LifecycleEvent, RetrySafety, RunContext,
10    RunError, RunErrorKind, RunEventKind, Usage,
11};
12use runifold_effect::{
13    EffectExecutionContext, EffectExecutor, EffectFuture, EffectHandler, EffectRecoveryPolicy,
14    InMemoryEffectStore,
15};
16use runifold_model::{
17    ContentPart, FeaturePolicy, GenerationOptions, Message, Model, ModelCallContext, ModelError,
18    ModelErrorKind, ModelRef, ModelRequest, ModelResponse, ModelStreamAccumulator, OutputFormat,
19    ProviderToolSpec, ResponseMode, Role, ToolCall, ToolChoice, ToolResult,
20};
21use runifold_retrieval::{Document, Retriever};
22use runifold_tool::{ToolError, ToolErrorKind, ToolOutput, ToolRegistry};
23use schemars::JsonSchema;
24use serde::{Deserialize, Serialize};
25
26use crate::checkpoint::CheckpointCursor;
27use crate::stream::{AgentObserver, BufferedObserver, NoopObserver, emit_agent_event};
28use crate::{
29    AgentCheckpoint, AgentCheckpointPhase, AgentCheckpointState, DurableConversationCheckpoint,
30    ResumePolicy,
31};
32
33const TOOL_RESULT_EXECUTION_ID_METADATA: &str = "runifold.agent.execution_id";
34use crate::terminal_review::{TerminalReviewConfig, TurnReviewConfig};
35use crate::{
36    AgentError, AgentEventStream, AgentGateway, AgentOutcome, AgentStreamEvent, CallableKind,
37    CompletionRequirement, GatewayError, GatewayErrorKind, StructuredAgent,
38};
39use crate::{TerminalReviewPolicy, TerminalReviewer, TurnReviewPolicy, TurnReviewer};
40
41mod callable;
42mod checkpointing;
43pub(crate) mod completion;
44mod execution;
45mod observability;
46mod retrieval;
47mod review;
48
49/// A boxed, sendable future returned by an agent.
50#[cfg(not(target_arch = "wasm32"))]
51pub type AgentFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;
52
53/// A boxed future returned by an agent on single-threaded WASM.
54#[cfg(target_arch = "wasm32")]
55pub type AgentFuture<'a, T> = Pin<Box<dyn Future<Output = T> + 'a>>;
56
57/// One dynamic, capability-gated context source.
58#[derive(Clone)]
59pub(crate) struct DynamicContext {
60    pub(crate) limit: usize,
61    pub(crate) retriever: Arc<dyn Retriever>,
62}
63
64impl std::fmt::Debug for DynamicContext {
65    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
66        formatter
67            .debug_struct("DynamicContext")
68            .field("limit", &self.limit)
69            .field("retriever", self.retriever.descriptor())
70            .finish()
71    }
72}
73
74/// How the agent handles tool failures that are safe for model recovery.
75#[derive(Clone, Copy, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
76#[non_exhaustive]
77pub enum ToolErrorPolicy {
78    /// Return safe execution failures to the model as failed tool results.
79    #[default]
80    ReturnToModel,
81    /// Stop the agent immediately on every tool error.
82    FailFast,
83}
84
85/// Local bounds and recovery behavior for an agent.
86#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
87pub struct AgentConfig {
88    /// Local turn bound in addition to the shared run budget.
89    pub max_turns: u32,
90    /// Tool failure behavior.
91    pub tool_error_policy: ToolErrorPolicy,
92    /// Model capability-degradation policy.
93    pub feature_policy: FeaturePolicy,
94}
95
96impl Default for AgentConfig {
97    fn default() -> Self {
98        Self {
99            max_turns: 16,
100            tool_error_policy: ToolErrorPolicy::ReturnToModel,
101            feature_policy: FeaturePolicy::Strict,
102        }
103    }
104}
105
106/// One configured model-tool agent.
107#[derive(Clone)]
108pub struct Agent {
109    pub(crate) name: String,
110    pub(crate) model: Arc<dyn Model>,
111    pub(crate) model_ref: ModelRef,
112    pub(crate) instructions: Vec<Message>,
113    pub(crate) context: Vec<Document>,
114    pub(crate) dynamic_context: Vec<DynamicContext>,
115    pub(crate) tools: ToolRegistry,
116    pub(crate) agents: AgentGateway,
117    pub(crate) effects: EffectExecutor,
118    pub(crate) effect_recovery: EffectRecoveryPolicy,
119    pub(crate) config: AgentConfig,
120    pub(crate) tool_concurrency: std::num::NonZeroUsize,
121    pub(crate) min_successful_tool_calls: u32,
122    pub(crate) output_format: OutputFormat,
123    pub(crate) generation: GenerationOptions,
124    pub(crate) response_mode: ResponseMode,
125    pub(crate) provider_tools: Vec<ProviderToolSpec>,
126    pub(crate) provider_options: BTreeMap<String, serde_json::Value>,
127    pub(crate) completion_requirement: CompletionRequirement,
128    pub(crate) completion_validator: completion::CompletionValidator,
129    pub(crate) turn_review: Option<TurnReviewConfig>,
130    pub(crate) terminal_review: Option<TerminalReviewConfig>,
131}
132
133impl Agent {
134    /// Starts a fluent builder for an Agent.
135    pub fn builder(
136        name: impl Into<String>,
137        model: Arc<dyn Model>,
138        model_ref: ModelRef,
139    ) -> crate::AgentBuilder {
140        crate::AgentBuilder::new(name, model, model_ref)
141    }
142
143    /// Creates an agent without instructions or tools.
144    pub fn new(name: impl Into<String>, model: Arc<dyn Model>, model_ref: ModelRef) -> Self {
145        Self {
146            name: name.into(),
147            model,
148            model_ref,
149            instructions: Vec::new(),
150            context: Vec::new(),
151            dynamic_context: Vec::new(),
152            tools: ToolRegistry::new(),
153            agents: AgentGateway::new(),
154            effects: EffectExecutor::new(Arc::new(InMemoryEffectStore::new())),
155            effect_recovery: EffectRecoveryPolicy::RejectAmbiguous,
156            config: AgentConfig::default(),
157            tool_concurrency: std::num::NonZeroUsize::MIN,
158            min_successful_tool_calls: 0,
159            output_format: OutputFormat::Text,
160            generation: GenerationOptions::default(),
161            response_mode: ResponseMode::Streaming,
162            provider_tools: Vec::new(),
163            provider_options: BTreeMap::new(),
164            completion_requirement: CompletionRequirement::default(),
165            completion_validator: completion::CompletionValidator::content(),
166            turn_review: None,
167            terminal_review: None,
168        }
169    }
170
171    /// Limits concurrent calls in each contiguous batch of read-only local tools.
172    /// Defaults to one. Writes, unknown tools, and child agents are serial barriers.
173    /// Results retain model order; started siblings are drained before an error returns.
174    #[must_use]
175    pub const fn tool_concurrency(mut self, limit: std::num::NonZeroUsize) -> Self {
176        self.tool_concurrency = limit;
177        self
178    }
179
180    /// Appends a system instruction.
181    #[must_use]
182    pub fn system(mut self, instruction: impl Into<String>) -> Self {
183        self.instructions.push(Message::system(instruction));
184        self
185    }
186
187    /// Installs the registry whose tools are exposed and executable.
188    #[must_use]
189    pub fn tools(mut self, tools: ToolRegistry) -> Self {
190        self.tools = tools;
191        self
192    }
193
194    /// Installs the gateway whose child agents are exposed and callable.
195    #[must_use]
196    pub fn agents(mut self, agents: AgentGateway) -> Self {
197        self.agents = agents;
198        self
199    }
200
201    /// Replaces the write-ahead effect coordinator shared by callables.
202    #[must_use]
203    pub fn effect_executor(mut self, effects: EffectExecutor) -> Self {
204        self.effects = effects;
205        self
206    }
207
208    /// Sets recovery behavior for ambiguous callable effects.
209    #[must_use]
210    pub const fn effect_recovery_policy(mut self, policy: EffectRecoveryPolicy) -> Self {
211        self.effect_recovery = policy;
212        self
213    }
214
215    /// Replaces local execution configuration.
216    #[must_use]
217    pub const fn with_config(mut self, config: AgentConfig) -> Self {
218        self.config = config;
219        self
220    }
221
222    /// Sets terminal validation and bounded repair behavior.
223    #[must_use]
224    pub const fn completion_requirement(mut self, requirement: CompletionRequirement) -> Self {
225        self.completion_requirement = requirement;
226        self
227    }
228
229    /// Installs semantic review after selected model responses and before any
230    /// tool call from those responses can execute.
231    #[must_use]
232    pub fn turn_reviewer<R>(
233        self,
234        reviewer: R,
235        policy: TurnReviewPolicy,
236        capabilities: CapabilitySet,
237    ) -> Self
238    where
239        R: TurnReviewer + 'static,
240    {
241        self.shared_turn_reviewer(Arc::new(reviewer), policy, capabilities)
242    }
243
244    /// Installs a shared, type-erased internal-turn reviewer.
245    #[must_use]
246    pub fn shared_turn_reviewer(
247        mut self,
248        reviewer: Arc<dyn TurnReviewer>,
249        policy: TurnReviewPolicy,
250        capabilities: CapabilitySet,
251    ) -> Self {
252        let descriptor = reviewer.turn_descriptor().clone();
253        self.turn_review = Some(TurnReviewConfig {
254            reviewer,
255            descriptor,
256            policy,
257            capabilities,
258        });
259        self
260    }
261
262    /// Installs semantic review before a locally valid terminal candidate is
263    /// committed as the Agent outcome.
264    ///
265    /// Reviewer capabilities are attenuated against the parent Run and review
266    /// repairs consume ordinary shared budgets.
267    #[must_use]
268    pub fn terminal_reviewer<R>(
269        self,
270        reviewer: R,
271        policy: TerminalReviewPolicy,
272        capabilities: CapabilitySet,
273    ) -> Self
274    where
275        R: TerminalReviewer + 'static,
276    {
277        self.shared_terminal_reviewer(Arc::new(reviewer), policy, capabilities)
278    }
279
280    /// Installs a shared, type-erased terminal reviewer.
281    #[must_use]
282    pub fn shared_terminal_reviewer(
283        mut self,
284        reviewer: Arc<dyn TerminalReviewer>,
285        policy: TerminalReviewPolicy,
286        capabilities: CapabilitySet,
287    ) -> Self {
288        let descriptor = reviewer.descriptor().clone();
289        self.terminal_review = Some(TerminalReviewConfig {
290            reviewer,
291            descriptor,
292            policy,
293            capabilities,
294        });
295        self
296    }
297
298    /// Requires this many successful local Tool calls before terminal output.
299    ///
300    /// Failed Tool results, child-Agent delegations, provider-hosted Tools,
301    /// and Tool results from earlier conversation turns do not satisfy this
302    /// execution-local completion contract. A value of zero disables it.
303    #[must_use]
304    pub const fn min_successful_tool_calls(mut self, minimum: u32) -> Self {
305        self.min_successful_tool_calls = minimum;
306        self
307    }
308
309    /// Sets the desired format for the terminal model response.
310    #[must_use]
311    pub fn output_format(mut self, output_format: OutputFormat) -> Self {
312        self.output_format = output_format;
313        self
314    }
315
316    /// Requests strict structured output described by the Rust type `T`.
317    #[must_use]
318    pub fn structured_output<T>(self, name: impl Into<String>) -> Self
319    where
320        T: JsonSchema,
321    {
322        self.structured_output_with_strictness::<T>(name, true)
323    }
324
325    /// Requests structured output described by `T` with explicit provider
326    /// strictness.
327    #[must_use]
328    pub fn structured_output_with_strictness<T>(self, name: impl Into<String>, strict: bool) -> Self
329    where
330        T: JsonSchema,
331    {
332        self.output_format(OutputFormat::typed_with_strictness::<T>(name, strict))
333    }
334
335    /// Binds provider schema generation and local decoding to the same type.
336    pub fn into_structured<T>(self, name: impl Into<String>) -> StructuredAgent<T>
337    where
338        T: JsonSchema + serde::de::DeserializeOwned + Send + 'static,
339    {
340        self.into_structured_with_strictness::<T>(name, true)
341    }
342
343    /// Binds provider schema generation and local decoding to `T` with
344    /// explicit provider strictness.
345    pub fn into_structured_with_strictness<T>(
346        mut self,
347        name: impl Into<String>,
348        strict: bool,
349    ) -> StructuredAgent<T>
350    where
351        T: JsonSchema + serde::de::DeserializeOwned + Send + 'static,
352    {
353        self = self.structured_output_with_strictness::<T>(name, strict);
354        self.completion_validator = completion::CompletionValidator::structured::<T>();
355        StructuredAgent::new(self)
356    }
357
358    /// Returns the stable local agent name.
359    pub fn name(&self) -> &str {
360        &self.name
361    }
362
363    /// Returns the provider and model identity used by this Agent.
364    pub const fn model_ref(&self) -> &ModelRef {
365        &self.model_ref
366    }
367
368    /// Returns capabilities for every Tool, child Agent, context source, and
369    /// terminal reviewer exposed by this Agent.
370    ///
371    /// The returned set is not granted automatically. Applications explicitly
372    /// decide whether to install it on a root or delegated Run.
373    pub fn callable_capabilities(&self) -> CapabilitySet {
374        let mut capabilities = CapabilitySet::new();
375        for spec in self.tools.model_specs() {
376            if let Some(descriptor) = self.tools.descriptor(&spec.name) {
377                capabilities.grant(descriptor.capability());
378            }
379        }
380        for spec in self.agents.model_specs() {
381            if let Some(descriptor) = self.agents.descriptor(&spec.name) {
382                capabilities.grant(descriptor.capability());
383            }
384        }
385        for source in &self.dynamic_context {
386            capabilities.grant(source.retriever.descriptor().capability());
387        }
388        if let Some(review) = &self.terminal_review {
389            for capability in review.capabilities.iter() {
390                capabilities.grant(capability.clone());
391            }
392        }
393        capabilities
394    }
395
396    /// Creates a root context for the ergonomic prompt surface.
397    ///
398    /// The context has no hard budget limits and grants only the Tool and child
399    /// Agent capabilities explicitly registered on this Agent. Applications
400    /// that need deadlines, tighter budgets, narrower authority, durable
401    /// journals, or shared run trees should construct a [`RunContext`] and use
402    /// [`Self::run`] instead.
403    #[must_use]
404    pub fn default_run_context(&self) -> RunContext {
405        RunContext::root(
406            BudgetTracker::new(Budget::default()),
407            self.callable_capabilities(),
408        )
409    }
410}
411
412impl std::fmt::Debug for Agent {
413    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
414        formatter
415            .debug_struct("Agent")
416            .field("name", &self.name)
417            .field("model_ref", &self.model_ref)
418            .field("instructions", &self.instructions)
419            .field("context", &self.context)
420            .field("dynamic_context", &self.dynamic_context)
421            .field("tools", &self.tools)
422            .field("agents", &self.agents)
423            .field("effects", &self.effects)
424            .field("effect_recovery", &self.effect_recovery)
425            .field("config", &self.config)
426            .field("output_format", &self.output_format)
427            .field("terminal_review", &self.terminal_review)
428            .finish_non_exhaustive()
429    }
430}
431
432#[cfg(test)]
433mod tests;
runifold_agent/agent.rs

runifold_agent/
agent.rs
