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    /// Low-level construction for an explicitly supplied Model.
144    ///
145    /// For provider-backed applications prefer the facade `ProviderRuntime::agent`
146    /// entry point and share the runtime across requests. This constructor is
147    /// useful for custom model composition and deterministic offline tests.
148    /// It uses the same canonical execution engine.
149    pub fn new(name: impl Into<String>, model: Arc<dyn Model>, model_ref: ModelRef) -> Self {
150        Self {
151            name: name.into(),
152            model,
153            model_ref,
154            instructions: Vec::new(),
155            context: Vec::new(),
156            dynamic_context: Vec::new(),
157            tools: ToolRegistry::new(),
158            agents: AgentGateway::new(),
159            effects: EffectExecutor::new(Arc::new(InMemoryEffectStore::new())),
160            effect_recovery: EffectRecoveryPolicy::RejectAmbiguous,
161            config: AgentConfig::default(),
162            tool_concurrency: std::num::NonZeroUsize::MIN,
163            min_successful_tool_calls: 0,
164            output_format: OutputFormat::Text,
165            generation: GenerationOptions::default(),
166            response_mode: ResponseMode::Streaming,
167            provider_tools: Vec::new(),
168            provider_options: BTreeMap::new(),
169            completion_requirement: CompletionRequirement::default(),
170            completion_validator: completion::CompletionValidator::content(),
171            turn_review: None,
172            terminal_review: None,
173        }
174    }
175
176    /// Limits concurrent calls in each contiguous batch of read-only local tools.
177    /// Defaults to one. Writes, unknown tools, and child agents are serial barriers.
178    /// Results retain model order; started siblings are drained before an error returns.
179    #[must_use]
180    pub const fn tool_concurrency(mut self, limit: std::num::NonZeroUsize) -> Self {
181        self.tool_concurrency = limit;
182        self
183    }
184
185    /// Appends a system instruction.
186    #[must_use]
187    pub fn system(mut self, instruction: impl Into<String>) -> Self {
188        self.instructions.push(Message::system(instruction));
189        self
190    }
191
192    /// Installs the registry whose tools are exposed and executable.
193    #[must_use]
194    pub fn tools(mut self, tools: ToolRegistry) -> Self {
195        self.tools = tools;
196        self
197    }
198
199    /// Installs the gateway whose child agents are exposed and callable.
200    #[must_use]
201    pub fn agents(mut self, agents: AgentGateway) -> Self {
202        self.agents = agents;
203        self
204    }
205
206    /// Replaces the write-ahead effect coordinator shared by callables.
207    #[must_use]
208    pub fn effect_executor(mut self, effects: EffectExecutor) -> Self {
209        self.effects = effects;
210        self
211    }
212
213    /// Sets recovery behavior for ambiguous callable effects.
214    #[must_use]
215    pub const fn effect_recovery_policy(mut self, policy: EffectRecoveryPolicy) -> Self {
216        self.effect_recovery = policy;
217        self
218    }
219
220    /// Replaces local execution configuration.
221    #[must_use]
222    pub const fn with_config(mut self, config: AgentConfig) -> Self {
223        self.config = config;
224        self
225    }
226
227    /// Sets terminal validation and bounded repair behavior.
228    #[must_use]
229    pub const fn completion_requirement(mut self, requirement: CompletionRequirement) -> Self {
230        self.completion_requirement = requirement;
231        self
232    }
233
234    /// Installs semantic review after selected model responses and before any
235    /// tool call from those responses can execute.
236    #[must_use]
237    pub fn turn_reviewer<R>(
238        self,
239        reviewer: R,
240        policy: TurnReviewPolicy,
241        capabilities: CapabilitySet,
242    ) -> Self
243    where
244        R: TurnReviewer + 'static,
245    {
246        self.shared_turn_reviewer(Arc::new(reviewer), policy, capabilities)
247    }
248
249    /// Installs a shared, type-erased internal-turn reviewer.
250    #[must_use]
251    pub fn shared_turn_reviewer(
252        mut self,
253        reviewer: Arc<dyn TurnReviewer>,
254        policy: TurnReviewPolicy,
255        capabilities: CapabilitySet,
256    ) -> Self {
257        let descriptor = reviewer.turn_descriptor().clone();
258        self.turn_review = Some(TurnReviewConfig {
259            reviewer,
260            descriptor,
261            policy,
262            capabilities,
263        });
264        self
265    }
266
267    /// Installs semantic review before a locally valid terminal candidate is
268    /// committed as the Agent outcome.
269    ///
270    /// Reviewer capabilities are attenuated against the parent Run and review
271    /// repairs consume ordinary shared budgets.
272    #[must_use]
273    pub fn terminal_reviewer<R>(
274        self,
275        reviewer: R,
276        policy: TerminalReviewPolicy,
277        capabilities: CapabilitySet,
278    ) -> Self
279    where
280        R: TerminalReviewer + 'static,
281    {
282        self.shared_terminal_reviewer(Arc::new(reviewer), policy, capabilities)
283    }
284
285    /// Installs a shared, type-erased terminal reviewer.
286    #[must_use]
287    pub fn shared_terminal_reviewer(
288        mut self,
289        reviewer: Arc<dyn TerminalReviewer>,
290        policy: TerminalReviewPolicy,
291        capabilities: CapabilitySet,
292    ) -> Self {
293        let descriptor = reviewer.descriptor().clone();
294        self.terminal_review = Some(TerminalReviewConfig {
295            reviewer,
296            descriptor,
297            policy,
298            capabilities,
299        });
300        self
301    }
302
303    /// Requires this many successful local Tool calls before terminal output.
304    ///
305    /// Failed Tool results, child-Agent delegations, provider-hosted Tools,
306    /// and Tool results from earlier conversation turns do not satisfy this
307    /// execution-local completion contract. A value of zero disables it.
308    #[must_use]
309    pub const fn min_successful_tool_calls(mut self, minimum: u32) -> Self {
310        self.min_successful_tool_calls = minimum;
311        self
312    }
313
314    /// Sets the desired format for the terminal model response.
315    #[must_use]
316    pub fn output_format(mut self, output_format: OutputFormat) -> Self {
317        self.output_format = output_format;
318        self
319    }
320
321    /// Requests strict structured output described by the Rust type `T`.
322    #[must_use]
323    pub fn structured_output<T>(self, name: impl Into<String>) -> Self
324    where
325        T: JsonSchema,
326    {
327        self.structured_output_with_strictness::<T>(name, true)
328    }
329
330    /// Requests structured output described by `T` with explicit provider
331    /// strictness.
332    #[must_use]
333    pub fn structured_output_with_strictness<T>(self, name: impl Into<String>, strict: bool) -> Self
334    where
335        T: JsonSchema,
336    {
337        self.output_format(OutputFormat::typed_with_strictness::<T>(name, strict))
338    }
339
340    /// Binds provider schema generation and local decoding to the same type.
341    pub fn into_structured<T>(self, name: impl Into<String>) -> StructuredAgent<T>
342    where
343        T: JsonSchema + serde::de::DeserializeOwned + Send + 'static,
344    {
345        self.into_structured_with_strictness::<T>(name, true)
346    }
347
348    /// Binds provider schema generation and local decoding to `T` with
349    /// explicit provider strictness.
350    pub fn into_structured_with_strictness<T>(
351        mut self,
352        name: impl Into<String>,
353        strict: bool,
354    ) -> StructuredAgent<T>
355    where
356        T: JsonSchema + serde::de::DeserializeOwned + Send + 'static,
357    {
358        self = self.structured_output_with_strictness::<T>(name, strict);
359        self.completion_validator = completion::CompletionValidator::structured::<T>();
360        StructuredAgent::new(self)
361    }
362
363    /// Returns the stable local agent name.
364    pub fn name(&self) -> &str {
365        &self.name
366    }
367
368    /// Returns the provider and model identity used by this Agent.
369    pub const fn model_ref(&self) -> &ModelRef {
370        &self.model_ref
371    }
372
373    /// Returns capabilities for every Tool, child Agent, context source, and
374    /// terminal reviewer exposed by this Agent.
375    ///
376    /// The returned set is not granted automatically. Applications explicitly
377    /// decide whether to install it on a root or delegated Run.
378    pub fn callable_capabilities(&self) -> CapabilitySet {
379        let mut capabilities = CapabilitySet::new();
380        for spec in self.tools.model_specs() {
381            if let Some(descriptor) = self.tools.descriptor(&spec.name) {
382                capabilities.grant(descriptor.capability());
383            }
384        }
385        for spec in self.agents.model_specs() {
386            if let Some(descriptor) = self.agents.descriptor(&spec.name) {
387                capabilities.grant(descriptor.capability());
388            }
389        }
390        for source in &self.dynamic_context {
391            capabilities.grant(source.retriever.descriptor().capability());
392        }
393        if let Some(review) = &self.terminal_review {
394            for capability in review.capabilities.iter() {
395                capabilities.grant(capability.clone());
396            }
397        }
398        capabilities
399    }
400
401    /// Creates a root context for the ergonomic prompt surface.
402    ///
403    /// The context has no hard budget limits and grants only the Tool and child
404    /// Agent capabilities explicitly registered on this Agent. Applications
405    /// that need deadlines, tighter budgets, narrower authority, durable
406    /// journals, or shared run trees should construct a [`RunContext`] and use
407    /// [`Self::run`] instead.
408    #[must_use]
409    pub fn default_run_context(&self) -> RunContext {
410        RunContext::root(
411            BudgetTracker::new(Budget::default()),
412            self.callable_capabilities(),
413        )
414    }
415}
416
417impl std::fmt::Debug for Agent {
418    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
419        formatter
420            .debug_struct("Agent")
421            .field("name", &self.name)
422            .field("model_ref", &self.model_ref)
423            .field("instructions", &self.instructions)
424            .field("context", &self.context)
425            .field("dynamic_context", &self.dynamic_context)
426            .field("tools", &self.tools)
427            .field("agents", &self.agents)
428            .field("effects", &self.effects)
429            .field("effect_recovery", &self.effect_recovery)
430            .field("config", &self.config)
431            .field("output_format", &self.output_format)
432            .field("terminal_review", &self.terminal_review)
433            .finish_non_exhaustive()
434    }
435}
436
437#[cfg(test)]
438mod tests;
runifold_agent/agent.rs

runifold_agent/
agent.rs
