agents_kit/builder.rs
1use std::collections::BTreeSet;
2use std::sync::{Arc, Mutex};
3use std::time::Duration;
4use tools_kit::ToolsetBackend;
5
6use providers_kit::ProviderBackend;
7use tokio::sync::oneshot;
8use tokio::task::JoinHandle;
9
10use crate::agent_storage::AgentStorageClaim;
11use crate::backends::{codex, custom, opencode};
12use crate::{
13 Agent, AgentBackend, AgentConfig, AgentId, AgentInfo, AgentStorage, BackendConfig, KitError,
14 KnowledgeOwnerId, ManagedKnowledge, ToolSelection,
15};
16use storage_kit::{Database, DatabaseBackend, StateStore, StateStoreBackend};
17
18/// Configures creation or reopening of one durable [`Agent`].
19///
20/// The builder accepts already-built provider, tool, and storage values without
21/// exposing their implementation details through `agents-kit`.
22///
23/// The hierarchy preserves concrete `S`, `D`, `P` and `T` types, with no Clone
24/// requirement on caller backends or `T::Handler`. New identity creation and
25/// durable reopen use separate constructors; reopening resupplies integrations
26/// instead of reconstructing credentials or lower configuration from records.
27///
28/// Native Codex/OpenCode can select `P` as static composition context without
29/// owning a provider value. Custom requires a value through [`Self::with_provider`].
30/// The builder owns supplied values, while [`Self::build`] accepts construction
31/// and cleanup work. Await consuming [`AgentBackend::shutdown`] on the built Agent
32/// while its runtime remains driven.
33///
34/// The local example uses memory Lance and a passively constructed Provider. It
35/// selects an exact model without verifying a running service or generating output.
36///
37/// # Examples
38///
39/// ```
40/// use std::{num::NonZeroUsize, time::Duration};
41/// use agents_kit::{AgentBackend, AgentBuilder, AgentStorage};
42/// use agents_kit::config::{AgentConfig, CustomAgentConfig, CustomLoopLimits, CustomToolApproval};
43/// use agents_kit::{model::ModelSelector, tool::ToolSelection};
44/// use providers_kit::{Provider, ProviderConfig, config::{OllamaConfig, ProviderKind, TransportConfig}};
45/// use storage_kit::{StateStore, state::config::{LanceStateStoreConfig, StateStoreConfig}};
46/// # let runtime = tokio::runtime::Builder::new_current_thread().enable_all().build()?;
47/// # runtime.block_on(async {
48/// let store = StateStore::connect(StateStoreConfig::Lance(LanceStateStoreConfig::memory())).await?;
49/// let storage = AgentStorage::new(store);
50/// let custom = CustomAgentConfig::new(ModelSelector::new("exact-model")?, ToolSelection::None,
51/// CustomLoopLimits::new(4, 0, 1)?, CustomToolApproval::Deny)?;
52/// let config = AgentConfig::new(custom.into(), Duration::from_secs(10), NonZeroUsize::MIN)?;
53/// let provider_config = ProviderConfig::new(ProviderKind::Ollama(OllamaConfig::local()), TransportConfig::default());
54/// let provider = Provider::from_config(provider_config.clone())?;
55/// // This constructs local owners and selects a model; it makes no generation request.
56/// let agent = AgentBuilder::create(config, storage).with_provider(provider).build().await?;
57/// assert!(agent.list_workspaces().await?.is_empty());
58/// agent.shutdown().await?;
59/// # Ok::<(), Box<dyn std::error::Error>>(())
60/// # })?;
61/// # Ok::<(), Box<dyn std::error::Error>>(())
62/// ```
63pub struct AgentBuilder<
64 S: StateStoreBackend = StateStore,
65 D: DatabaseBackend = Database,
66 P: ProviderBackend = providers_kit::Provider,
67 T: ToolsetBackend = tools_kit::Toolset,
68> {
69 config: AgentConfig,
70 storage: AgentStorage<S>,
71 mode: AgentBuildMode,
72 provider: Option<P>,
73 toolsets: Vec<Arc<T>>,
74 knowledge_database: Option<D>,
75}
76
77enum AgentBuildMode {
78 Create,
79 Reopen(AgentId),
80}
81
82impl<S: StateStoreBackend> AgentBuilder<S, Database, providers_kit::Provider> {
83 /// Configures a new durable agent over the required control-plane store.
84 ///
85 /// This records creation intent without opening a backend. [`Self::build`] creates
86 /// a fresh durable identity. Storage is required up front and retains its exact
87 /// associated transaction type. For an existing identity use [`Self::reopen`].
88 ///
89 /// # Examples
90 ///
91 /// ```
92 /// use std::{num::NonZeroUsize, time::Duration};
93 /// use agents_kit::{AgentBackend, AgentBuilder, AgentStorage};
94 /// use agents_kit::config::{AgentConfig, CustomAgentConfig, CustomLoopLimits, CustomToolApproval};
95 /// use agents_kit::{model::ModelSelector, tool::ToolSelection};
96 /// use providers_kit::{Provider, ProviderConfig, config::{OllamaConfig, ProviderKind, TransportConfig}};
97 /// use storage_kit::{StateStore, state::config::{LanceStateStoreConfig, StateStoreConfig}};
98 /// # let runtime = tokio::runtime::Builder::new_current_thread().enable_all().build()?;
99 /// # runtime.block_on(async {
100 /// let store = StateStore::connect(StateStoreConfig::Lance(LanceStateStoreConfig::memory())).await?;
101 /// let storage = AgentStorage::new(store);
102 /// let custom = CustomAgentConfig::new(ModelSelector::new("exact-model")?, ToolSelection::None,
103 /// CustomLoopLimits::new(4, 0, 1)?, CustomToolApproval::Deny)?;
104 /// let config = AgentConfig::new(custom.into(), Duration::from_secs(10), NonZeroUsize::MIN)?;
105 /// let provider_config = ProviderConfig::new(ProviderKind::Ollama(OllamaConfig::local()), TransportConfig::default());
106 /// let provider = Provider::from_config(provider_config.clone())?;
107 /// let builder = AgentBuilder::create(config, storage);
108 /// let agent = builder.with_provider(provider).build().await?;
109 /// assert!(agent.list_workspaces().await?.is_empty());
110 /// agent.shutdown().await?;
111 /// # Ok::<(), Box<dyn std::error::Error>>(())
112 /// # })?;
113 /// # Ok::<(), Box<dyn std::error::Error>>(())
114 /// ```
115 pub fn create(config: AgentConfig, storage: AgentStorage<S>) -> Self {
116 Self::create_for_backends::<providers_kit::Provider, tools_kit::Toolset>(config, storage)
117 }
118
119 /// Configures a new agent with fixed Provider and Toolset implementation types.
120 ///
121 /// Codex and OpenCode retain their native providers, so this selects only
122 /// the static provider type shared with other agents in a caller-owned
123 /// composition. Custom agents receive their provider value through
124 /// [`Self::with_provider`]. Repeated [`Self::with_toolset`] calls append T2
125 /// values to one homogeneous inventory without normalization or a Clone bound.
126 ///
127 ///
128 ///
129 /// # Examples
130 ///
131 /// ```
132 /// use std::{num::NonZeroUsize, time::Duration};
133 /// use agents_kit::{AgentBackend, AgentBuilder, AgentStorage};
134 /// use agents_kit::config::{AgentConfig, CustomAgentConfig, CustomLoopLimits, CustomToolApproval};
135 /// use agents_kit::{model::ModelSelector, tool::ToolSelection};
136 /// use providers_kit::{Provider, ProviderConfig, config::{OllamaConfig, ProviderKind, TransportConfig}};
137 /// use storage_kit::{StateStore, state::config::{LanceStateStoreConfig, StateStoreConfig}};
138 /// # let runtime = tokio::runtime::Builder::new_current_thread().enable_all().build()?;
139 /// # runtime.block_on(async {
140 /// let store = StateStore::connect(StateStoreConfig::Lance(LanceStateStoreConfig::memory())).await?;
141 /// let storage = AgentStorage::new(store);
142 /// let custom = CustomAgentConfig::new(ModelSelector::new("exact-model")?, ToolSelection::None,
143 /// CustomLoopLimits::new(4, 0, 1)?, CustomToolApproval::Deny)?;
144 /// let config = AgentConfig::new(custom.into(), Duration::from_secs(10), NonZeroUsize::MIN)?;
145 /// let provider_config = ProviderConfig::new(ProviderKind::Ollama(OllamaConfig::local()), TransportConfig::default());
146 /// let provider = Provider::from_config(provider_config.clone())?;
147 /// let builder = AgentBuilder::create_for_backends::<Provider, tools_kit::Toolset>(config, storage);
148 /// // Selecting P alone does not supply the provider required by Custom.
149 /// let agent = builder.with_provider(provider).build().await?;
150 /// assert!(agent.list_workspaces().await?.is_empty());
151 /// agent.shutdown().await?;
152 /// # Ok::<(), Box<dyn std::error::Error>>(())
153 /// # })?;
154 /// # Ok::<(), Box<dyn std::error::Error>>(())
155 /// ```
156 pub fn create_for_backends<P2: ProviderBackend, T2: ToolsetBackend>(
157 config: AgentConfig,
158 storage: AgentStorage<S>,
159 ) -> AgentBuilder<S, Database, P2, T2> {
160 AgentBuilder {
161 config,
162 storage,
163 mode: AgentBuildMode::Create,
164 provider: None,
165 toolsets: Vec::new(),
166 knowledge_database: None,
167 }
168 }
169
170 /// Configures reopening of an existing durable agent.
171 ///
172 /// Selects an existing identity; it never falls back to creating one. Build checks
173 /// stored compatibility and the storage clone family's live-ownership claim.
174 /// Resupply fresh provider/tool/database values required by that identity.
175 /// An AgentStorage clone shares coordination without requiring `S: Clone`.
176 ///
177 /// # Examples
178 ///
179 /// ```
180 /// use std::{num::NonZeroUsize, time::Duration};
181 /// use agents_kit::{AgentBackend, AgentBuilder, AgentStorage};
182 /// use agents_kit::config::{AgentConfig, CustomAgentConfig, CustomLoopLimits, CustomToolApproval};
183 /// use agents_kit::{model::ModelSelector, tool::ToolSelection};
184 /// use providers_kit::{Provider, ProviderConfig, config::{OllamaConfig, ProviderKind, TransportConfig}};
185 /// use storage_kit::{StateStore, state::config::{LanceStateStoreConfig, StateStoreConfig}};
186 /// # let runtime = tokio::runtime::Builder::new_current_thread().enable_all().build()?;
187 /// # runtime.block_on(async {
188 /// let store = StateStore::connect(StateStoreConfig::Lance(LanceStateStoreConfig::memory())).await?;
189 /// let storage = AgentStorage::new(store);
190 /// let custom = CustomAgentConfig::new(ModelSelector::new("exact-model")?, ToolSelection::None,
191 /// CustomLoopLimits::new(4, 0, 1)?, CustomToolApproval::Deny)?;
192 /// let config = AgentConfig::new(custom.into(), Duration::from_secs(10), NonZeroUsize::MIN)?;
193 /// let provider_config = ProviderConfig::new(ProviderKind::Ollama(OllamaConfig::local()), TransportConfig::default());
194 /// let provider = Provider::from_config(provider_config.clone())?;
195 /// let agent = AgentBuilder::create(config.clone(), storage.clone())
196 /// .with_provider(provider).build().await?;
197 /// let id = agent.id();
198 /// agent.shutdown().await?;
199 /// // Reuse the same storage family, and supply a fresh provider after shutdown.
200 /// let reopened = AgentBuilder::reopen(config, storage, id.clone())
201 /// .with_provider(Provider::from_config(provider_config)?).build().await?;
202 /// assert_eq!(reopened.id(), id);
203 /// reopened.shutdown().await?;
204 /// # Ok::<(), Box<dyn std::error::Error>>(())
205 /// # })?;
206 /// # Ok::<(), Box<dyn std::error::Error>>(())
207 /// ```
208 pub fn reopen(config: AgentConfig, storage: AgentStorage<S>, id: AgentId) -> Self {
209 Self::reopen_for_backends::<providers_kit::Provider, tools_kit::Toolset>(
210 config, storage, id,
211 )
212 }
213
214 /// Configures reopening with fixed Provider and Toolset implementation types.
215 ///
216 /// This is the reopening counterpart to [`Self::create_for_backends`]. It
217 /// selects no provider value and performs no provider operation.
218 ///
219 ///
220 ///
221 /// # Examples
222 ///
223 /// ```
224 /// use std::{num::NonZeroUsize, time::Duration};
225 /// use agents_kit::{AgentBackend, AgentBuilder, AgentStorage};
226 /// use agents_kit::config::{AgentConfig, CustomAgentConfig, CustomLoopLimits, CustomToolApproval};
227 /// use agents_kit::{model::ModelSelector, tool::ToolSelection};
228 /// use providers_kit::{Provider, ProviderConfig, config::{OllamaConfig, ProviderKind, TransportConfig}};
229 /// use storage_kit::{StateStore, state::config::{LanceStateStoreConfig, StateStoreConfig}};
230 /// # let runtime = tokio::runtime::Builder::new_current_thread().enable_all().build()?;
231 /// # runtime.block_on(async {
232 /// let store = StateStore::connect(StateStoreConfig::Lance(LanceStateStoreConfig::memory())).await?;
233 /// let storage = AgentStorage::new(store);
234 /// let custom = CustomAgentConfig::new(ModelSelector::new("exact-model")?, ToolSelection::None,
235 /// CustomLoopLimits::new(4, 0, 1)?, CustomToolApproval::Deny)?;
236 /// let config = AgentConfig::new(custom.into(), Duration::from_secs(10), NonZeroUsize::MIN)?;
237 /// let provider_config = ProviderConfig::new(ProviderKind::Ollama(OllamaConfig::local()), TransportConfig::default());
238 /// let provider = Provider::from_config(provider_config.clone())?;
239 /// let agent = AgentBuilder::create(config.clone(), storage.clone())
240 /// .with_provider(provider).build().await?;
241 /// let id = agent.id();
242 /// agent.shutdown().await?;
243 /// let builder = AgentBuilder::reopen_for_backends::<Provider, tools_kit::Toolset>(
244 /// config, storage, id.clone(),
245 /// );
246 /// let reopened = builder.with_provider(Provider::from_config(provider_config)?).build().await?;
247 /// assert_eq!(reopened.id(), id);
248 /// reopened.shutdown().await?;
249 /// # Ok::<(), Box<dyn std::error::Error>>(())
250 /// # })?;
251 /// # Ok::<(), Box<dyn std::error::Error>>(())
252 /// ```
253 pub fn reopen_for_backends<P2: ProviderBackend, T2: ToolsetBackend>(
254 config: AgentConfig,
255 storage: AgentStorage<S>,
256 id: AgentId,
257 ) -> AgentBuilder<S, Database, P2, T2> {
258 let mut builder = Self::create_for_backends::<P2, T2>(config, storage);
259 builder.mode = AgentBuildMode::Reopen(id);
260 builder
261 }
262}
263
264impl<S: StateStoreBackend, D: DatabaseBackend, P: ProviderBackend, T: ToolsetBackend>
265 AgentBuilder<S, D, P, T>
266{
267 /// Supplies the already-built provider required by the Custom backend.
268 ///
269 /// Consumes the provider and changes the builder's concrete provider parameter;
270 /// accepted Toolsets and the Database remain attached. Replacing an earlier value
271 /// drops that value, so complete any required lower lifecycle work before replacing
272 /// it. Codex/OpenCode reject a supplied provider value during build; use typed
273 /// construction to select their static provider parameter without a value.
274 ///
275 /// # Examples
276 ///
277 /// ```
278 /// use std::{num::NonZeroUsize, time::Duration};
279 /// use agents_kit::{AgentBackend, AgentBuilder, AgentStorage};
280 /// use agents_kit::config::{AgentConfig, CustomAgentConfig, CustomLoopLimits, CustomToolApproval};
281 /// use agents_kit::{model::ModelSelector, tool::ToolSelection};
282 /// use providers_kit::{Provider, ProviderConfig, config::{OllamaConfig, ProviderKind, TransportConfig}};
283 /// use storage_kit::{StateStore, state::config::{LanceStateStoreConfig, StateStoreConfig}};
284 /// # let runtime = tokio::runtime::Builder::new_current_thread().enable_all().build()?;
285 /// # runtime.block_on(async {
286 /// let store = StateStore::connect(StateStoreConfig::Lance(LanceStateStoreConfig::memory())).await?;
287 /// let storage = AgentStorage::new(store);
288 /// let custom = CustomAgentConfig::new(ModelSelector::new("exact-model")?, ToolSelection::None,
289 /// CustomLoopLimits::new(4, 0, 1)?, CustomToolApproval::Deny)?;
290 /// let config = AgentConfig::new(custom.into(), Duration::from_secs(10), NonZeroUsize::MIN)?;
291 /// let provider_config = ProviderConfig::new(ProviderKind::Ollama(OllamaConfig::local()), TransportConfig::default());
292 /// let provider = Provider::from_config(provider_config.clone())?;
293 /// let agent = AgentBuilder::create(config, storage)
294 /// .with_provider(provider).build().await?;
295 /// assert!(agent.list_workspaces().await?.is_empty());
296 /// // Shutdown includes the transferred provider; do not reuse it afterward.
297 /// agent.shutdown().await?;
298 /// # Ok::<(), Box<dyn std::error::Error>>(())
299 /// # })?;
300 /// # Ok::<(), Box<dyn std::error::Error>>(())
301 /// ```
302 /// # Direct lower implementations and non-Clone tools
303 ///
304 /// These caller-owned enums deliberately expose only Provider selection and
305 /// generation, and Local tool execution. Other Provider operations retain the
306 /// trait's Unsupported defaults. They delegate actual transport/handler
307 /// semantics, not fabricated success. Agent uses `LocalProvider::Model` and
308 /// `LocalTools::Handler`; neither tool enum implements `Clone`.
309 ///
310 /// This example executes a real Local handler and builds Custom without model
311 /// inference. A prompt would require the exact configured Ollama model. See
312 /// [`providers_kit::ProviderBackend`] and [`tools_kit::ToolsetBackend`] for
313 /// implementing those lower contracts rather than delegating them.
314 ///
315 /// ```
316 /// use std::{convert::Infallible, num::NonZeroUsize, time::Duration};
317 /// use agents_kit::{AgentBackend, AgentBuilder, AgentStorage};
318 /// use agents_kit::config::{AgentConfig, CustomAgentConfig, CustomLoopLimits, CustomToolApproval};
319 /// use agents_kit::{model::ModelSelector, tool::ToolSelection};
320 /// use providers_kit::{Model, ModelBackend, ModelId, Provider, ProviderBackend, ProviderConfig};
321 /// use providers_kit::config::{OllamaConfig, ProviderKind, TransportConfig};
322 /// use providers_kit::generation::{GenerateEventStream, GenerateRequest, GenerateResponse, GenerationContinuation};
323 /// use storage_kit::{StateStore, state::config::{LanceStateStoreConfig, StateStoreConfig}};
324 /// use tools_kit::{ToolError, ToolHandler, ToolHandlerBackend, Toolset, ToolsetBackend};
325 /// use tools_kit::declaration::{ToolInfo, ToolName};
326 /// use tools_kit::execution::{ToolHandlerInput, ToolHandlerOutput, ToolInvocation, ToolResult};
327 /// use tools_kit::{local::ToolRegistration, schema::JsonSchema};
328 ///
329 /// enum LocalProvider { Ollama(Provider) }
330 /// enum LocalModel { Ollama(Model) }
331 /// enum LocalTools { Functions(Toolset) }
332 /// enum LocalHandler { Function(ToolHandler) }
333 /// impl ProviderBackend for LocalProvider {
334 /// type Model = LocalModel;
335 /// fn model(&self, id: ModelId) -> Result<Self::Model, providers_kit::KitError> {
336 /// match self { Self::Ollama(value) => value.model(id).map(LocalModel::Ollama) }
337 /// }
338 /// async fn shutdown(&self) -> Result<(), providers_kit::KitError> {
339 /// match self { Self::Ollama(value) => value.shutdown().await }
340 /// }
341 /// }
342 /// impl ModelBackend for LocalModel {
343 /// type Continuation = GenerationContinuation;
344 /// type Session = Infallible;
345 /// fn model_id(&self) -> &ModelId {
346 /// match self { Self::Ollama(value) => value.model_id() }
347 /// }
348 /// async fn generate(&self, request: &GenerateRequest<Self::Continuation>)
349 /// -> Result<GenerateResponse<Self::Continuation>, providers_kit::KitError> {
350 /// match self { Self::Ollama(value) => value.generate(request).await }
351 /// }
352 /// async fn generate_stream(&self, request: &GenerateRequest<Self::Continuation>)
353 /// -> Result<GenerateEventStream<Self::Continuation>, providers_kit::KitError> {
354 /// match self { Self::Ollama(value) => value.generate_stream(request).await }
355 /// }
356 /// }
357 /// impl ToolsetBackend for LocalTools {
358 /// type Handler = LocalHandler;
359 /// async fn list_tools(&self) -> Result<Vec<ToolInfo>, ToolError> {
360 /// match self { Self::Functions(value) => value.list_tools().await }
361 /// }
362 /// async fn tool_handler(&self, name: ToolName) -> Result<Self::Handler, ToolError> {
363 /// match self { Self::Functions(value) => value.tool_handler(name).await.map(LocalHandler::Function) }
364 /// }
365 /// async fn shutdown(&self) -> Result<(), ToolError> {
366 /// match self { Self::Functions(value) => value.shutdown().await }
367 /// }
368 /// }
369 /// impl ToolHandlerBackend for LocalHandler {
370 /// async fn step(&mut self, input: ToolHandlerInput) -> Result<ToolHandlerOutput, ToolError> {
371 /// match self { Self::Function(value) => value.step(input).await }
372 /// }
373 /// }
374 /// # let runtime = tokio::runtime::Builder::new_current_thread().enable_all().build()?;
375 /// # runtime.block_on(async {
376 /// let tools = LocalTools::Functions(Toolset::local(vec![ToolRegistration::function(
377 /// ToolInfo::new(ToolName::new("ping")?, JsonSchema::strict(serde_json::json!({"type": "object"}))),
378 /// |_| async { Ok(ToolResult::json(serde_json::json!({"reply": "pong"}))) },
379 /// )])?);
380 /// let direct = async {
381 /// let mut handler = tools.tool_handler(ToolName::new("ping")?).await?;
382 /// let output = handler.step(ToolHandlerInput::Invoke(ToolInvocation {
383 /// call_id: Some("caller-ping".into()), arguments: serde_json::json!({}),
384 /// })).await?;
385 /// let ToolHandlerOutput::Completed(result) = output else {
386 /// return Err(ToolError::Unsupported { operation: "ping must complete without interaction".into() });
387 /// };
388 /// assert!(!result.is_error);
389 /// assert_eq!(result.structured_content, Some(serde_json::json!({"reply": "pong"})));
390 /// Ok::<(), ToolError>(())
391 /// }.await;
392 /// if let Err(error) = direct {
393 /// let primary = agents_kit::KitError::from(error);
394 /// let error = match tools.shutdown().await {
395 /// Ok(()) => primary,
396 /// Err(cleanup) => primary.with_supplementary(agents_kit::KitError::from(cleanup)),
397 /// };
398 /// return Err(Box::<dyn std::error::Error>::from(error));
399 /// }
400 /// let store = StateStore::connect(StateStoreConfig::Lance(LanceStateStoreConfig::memory())).await?;
401 /// let custom = CustomAgentConfig::new(ModelSelector::new("exact-model")?, ToolSelection::Auto,
402 /// CustomLoopLimits::new(4, 4, 1)?, CustomToolApproval::Allow)?;
403 /// let config = AgentConfig::new(custom.into(), Duration::from_secs(10), NonZeroUsize::MIN)?;
404 /// let provider = LocalProvider::Ollama(Provider::from_config(ProviderConfig::new(
405 /// ProviderKind::Ollama(OllamaConfig::local()), TransportConfig::default()))?);
406 /// let agent = AgentBuilder::create_for_backends::<LocalProvider, LocalTools>(config, AgentStorage::new(store))
407 /// .with_provider(provider).with_toolset(tools).build().await?;
408 /// let outcome = agent.list_workspaces().await;
409 /// let shutdown = agent.shutdown().await; // owns Provider and the non-Clone Toolset
410 /// assert!(outcome?.is_empty());
411 /// shutdown?;
412 /// # Ok::<(), Box<dyn std::error::Error>>(())
413 /// # })?;
414 /// # Ok::<(), Box<dyn std::error::Error>>(())
415 /// ```
416 pub fn with_provider<P2: ProviderBackend>(self, provider: P2) -> AgentBuilder<S, D, P2, T> {
417 AgentBuilder {
418 config: self.config,
419 storage: self.storage,
420 mode: self.mode,
421 provider: Some(provider),
422 toolsets: self.toolsets,
423 knowledge_database: self.knowledge_database,
424 }
425 }
426
427 /// Appends one already-built executable Toolset implementation.
428 ///
429 /// T remains fixed by construction. Changing the Provider or Database keeps
430 /// every accepted Toolset; its unique Handler remains associated as `T::Handler`.
431 ///
432 /// Each call appends a value of the construction-selected `T`; use caller-owned
433 /// exhaustive Toolset/Handler enums for heterogeneous sources. Custom executes the
434 /// associated Handler directly. Build validates unique effective names and selected
435 /// tools before publication. Native support is backend-specific: OpenCode rejects
436 /// supplied executable toolsets. Builder transfer alone never invokes a handler.
437 ///
438 /// # Examples
439 ///
440 /// ```
441 /// use std::{num::NonZeroUsize, time::Duration};
442 /// use agents_kit::{AgentBackend, AgentBuilder, AgentStorage};
443 /// use agents_kit::config::{AgentConfig, CustomAgentConfig, CustomLoopLimits, CustomToolApproval};
444 /// use agents_kit::{model::ModelSelector, tool::ToolSelection};
445 /// use providers_kit::{Provider, ProviderConfig, config::{OllamaConfig, ProviderKind, TransportConfig}};
446 /// use storage_kit::{StateStore, state::config::{LanceStateStoreConfig, StateStoreConfig}};
447 /// # let runtime = tokio::runtime::Builder::new_current_thread().enable_all().build()?;
448 /// # runtime.block_on(async {
449 /// let store = StateStore::connect(StateStoreConfig::Lance(LanceStateStoreConfig::memory())).await?;
450 /// let storage = AgentStorage::new(store);
451 /// let custom = CustomAgentConfig::new(ModelSelector::new("exact-model")?, ToolSelection::None,
452 /// CustomLoopLimits::new(4, 0, 1)?, CustomToolApproval::Deny)?;
453 /// let config = AgentConfig::new(custom.into(), Duration::from_secs(10), NonZeroUsize::MIN)?;
454 /// let provider_config = ProviderConfig::new(ProviderKind::Ollama(OllamaConfig::local()), TransportConfig::default());
455 /// let provider = Provider::from_config(provider_config.clone())?;
456 /// use tools_kit::{Toolset, declaration::{ToolInfo, ToolName}, execution::ToolResult,
457 /// local::ToolRegistration, schema::JsonSchema};
458 /// let tools = Toolset::local(vec![ToolRegistration::function(
459 /// ToolInfo::new(ToolName::new("ping")?, JsonSchema::strict(serde_json::json!({"type":"object"}))),
460 /// |_| async { Ok(ToolResult::text("pong")) },
461 /// )])?;
462 /// let agent = AgentBuilder::create(config, storage)
463 /// .with_toolset(tools).with_provider(provider).build().await?;
464 /// assert!(agent.list_workspaces().await?.is_empty());
465 /// // The accepted Toolset is retained across with_provider and closed by Agent shutdown.
466 /// agent.shutdown().await?;
467 /// # Ok::<(), Box<dyn std::error::Error>>(())
468 /// # })?;
469 /// # Ok::<(), Box<dyn std::error::Error>>(())
470 /// ```
471 pub fn with_toolset(mut self, toolset: T) -> Self {
472 self.toolsets.push(Arc::new(toolset));
473 self
474 }
475
476 /// Supplies the Database used by the private managed-knowledge owner.
477 ///
478 /// Consumes a direct Database implementation and changes only the builder's
479 /// concrete `D` parameter. Existing Toolsets and Provider remain attached. This
480 /// Database is separate from AgentStorage's control-plane records and exposes no
481 /// public knowledge runtime. Reopen must resupply a compatible Database when the
482 /// durable identity requires managed knowledge.
483 ///
484 /// # Examples
485 ///
486 /// ```
487 /// use std::{num::NonZeroUsize, time::Duration};
488 /// use agents_kit::{AgentBackend, AgentBuilder, AgentStorage};
489 /// use agents_kit::config::{AgentConfig, CustomAgentConfig, CustomLoopLimits, CustomToolApproval};
490 /// use agents_kit::{model::ModelSelector, tool::ToolSelection};
491 /// use providers_kit::{Provider, ProviderConfig, config::{OllamaConfig, ProviderKind, TransportConfig}};
492 /// use storage_kit::{StateStore, state::config::{LanceStateStoreConfig, StateStoreConfig}};
493 /// # let runtime = tokio::runtime::Builder::new_current_thread().enable_all().build()?;
494 /// # runtime.block_on(async {
495 /// let store = StateStore::connect(StateStoreConfig::Lance(LanceStateStoreConfig::memory())).await?;
496 /// let storage = AgentStorage::new(store);
497 /// let custom = CustomAgentConfig::new(ModelSelector::new("exact-model")?, ToolSelection::None,
498 /// CustomLoopLimits::new(4, 0, 1)?, CustomToolApproval::Deny)?;
499 /// let config = AgentConfig::new(custom.into(), Duration::from_secs(10), NonZeroUsize::MIN)?;
500 /// let provider_config = ProviderConfig::new(ProviderKind::Ollama(OllamaConfig::local()), TransportConfig::default());
501 /// let provider = Provider::from_config(provider_config.clone())?;
502 /// use storage_kit::{Database, database::config::{DatabaseConfig, LanceConfig}};
503 /// let database = Database::connect(DatabaseConfig::lance(LanceConfig::embedded_path("memory://"))).await?;
504 /// let agent = AgentBuilder::create(config, storage).with_provider(provider)
505 /// .with_knowledge_database(database).build().await?;
506 /// assert!(agent.list_workspaces().await?.is_empty());
507 /// agent.shutdown().await?;
508 /// # Ok::<(), Box<dyn std::error::Error>>(())
509 /// # })?;
510 /// # Ok::<(), Box<dyn std::error::Error>>(())
511 /// ```
512 pub fn with_knowledge_database<D2: DatabaseBackend>(
513 self,
514 database: D2,
515 ) -> AgentBuilder<S, D2, P, T> {
516 AgentBuilder {
517 config: self.config,
518 storage: self.storage,
519 mode: self.mode,
520 provider: self.provider,
521 toolsets: self.toolsets,
522 knowledge_database: Some(database),
523 }
524 }
525
526 /// Validates the complete configuration and creates or reopens the agent.
527 ///
528 /// Takes ownership of construction after acceptance. On failure, prepared resources
529 /// are cleaned in reverse ownership order within the lifecycle deadline; the first
530 /// failure remains primary and later failures remain supplementary. Losing the
531 /// caller observation does not detach accepted build/cleanup ownership.
532 ///
533 /// On success the unique Agent owns its runtime and transferred integrations.
534 /// Keep Tokio and lower runtimes driven through consuming shutdown. Custom model
535 /// selection alone does not verify remote model availability; later provider
536 /// operations own service validation.
537 ///
538 /// # Errors
539 ///
540 /// Returns InvalidConfig for invalid backend/integration combinations (including a
541 /// missing Custom provider), Unsupported for a supplied integration the selected
542 /// backend cannot represent, and Conflict/Busy for incompatible or already-owned
543 /// reopen state. Missing identities use NotFound; malformed durable evidence uses
544 /// InvalidState. Process, transport, protocol, storage and deadline failures retain
545 /// their owning portable classification. Inspect [`KitError::kind`] and
546 /// [`KitError::context`]; build and reopen carry distinct operation context.
547 ///
548 /// This example requires an installed, configured `codex` executable and its real
549 /// App Server process. It does not submit a model prompt.
550 ///
551 /// # Examples
552 ///
553 /// ```no_run
554 /// use std::{num::NonZeroUsize, time::Duration};
555 /// use agents_kit::{AgentBackend, AgentBuilder, AgentStorage};
556 /// use agents_kit::config::{AgentConfig, CodexConfig};
557 /// use storage_kit::{StateStore, state::config::{LanceStateStoreConfig, StateStoreConfig}};
558 /// # let runtime = tokio::runtime::Builder::new_current_thread().enable_all().build()?;
559 /// # runtime.block_on(async {
560 /// let config = AgentConfig::new(CodexConfig::spawn(Duration::from_secs(15),
561 /// Duration::from_secs(60))?.into(), Duration::from_secs(30), NonZeroUsize::MIN)?;
562 /// let store = StateStore::connect(StateStoreConfig::Lance(LanceStateStoreConfig::memory())).await?;
563 /// let agent = AgentBuilder::create(config, AgentStorage::new(store)).build().await?;
564 /// let observed = agent.info().await;
565 /// let cleanup = agent.shutdown().await;
566 /// let info = observed?;
567 /// assert!(!info.id().as_str().is_empty());
568 /// cleanup?;
569 /// # Ok::<(), Box<dyn std::error::Error>>(())
570 /// # })?;
571 /// # Ok::<(), Box<dyn std::error::Error>>(())
572 /// ```
573 pub async fn build(self) -> Result<Agent<S, D, P, T>, KitError> {
574 let operation = match self.mode {
575 AgentBuildMode::Create => crate::error::Operation::Build,
576 AgentBuildMode::Reopen(_) => crate::error::Operation::Reopen,
577 };
578 AgentBuildPublication::<Agent<S, D, P, T>>::run(AgentBuildOwner::new(self))
579 .await
580 .map_err(|error| {
581 error
582 .at_operation(operation)
583 .in_phase(crate::error::ErrorPhase::Admission)
584 })
585 }
586}
587
588struct AgentBuildOwner<
589 S: StateStoreBackend,
590 D: DatabaseBackend,
591 P: ProviderBackend,
592 T: ToolsetBackend,
593> {
594 config: Option<AgentConfig>,
595 mode: AgentBuildMode,
596 provider: Option<P>,
597 toolsets: Vec<Arc<T>>,
598 storage: Option<AgentStorage<S>>,
599 knowledge_database: Option<D>,
600 claim: Option<AgentStorageClaim>,
601 agent: Option<Agent<S, D, P, T>>,
602 cleanup_timeout: Duration,
603 tracing_enabled: bool,
604}
605
606impl<S: StateStoreBackend, D: DatabaseBackend, P: ProviderBackend, T: ToolsetBackend>
607 AgentBuildOwner<S, D, P, T>
608{
609 fn new(builder: AgentBuilder<S, D, P, T>) -> Self {
610 let cleanup_timeout = builder.config.lifecycle_timeout();
611 let tracing_enabled = builder.config.tracing_enabled();
612 let AgentBuilder {
613 config,
614 storage,
615 mode,
616 provider,
617 toolsets,
618 knowledge_database,
619 } = builder;
620 Self {
621 config: Some(config),
622 mode,
623 provider,
624 toolsets,
625 storage: Some(storage),
626 knowledge_database,
627 claim: None,
628 agent: None,
629 cleanup_timeout,
630 tracing_enabled,
631 }
632 }
633
634 async fn build(mut self) -> Result<Agent<S, D, P, T>, KitError> {
635 if let Err(error) = self.validate().await {
636 return Err(self.cleanup_after_error(error).await);
637 }
638 let lifecycle_deadline = match tokio::time::Instant::now().checked_add(self.cleanup_timeout)
639 {
640 Some(deadline) => deadline,
641 None => {
642 return Err(self
643 .cleanup_after_error(KitError::invalid_config(
644 "lifecycle timeout cannot be represented",
645 ))
646 .await);
647 }
648 };
649
650 let storage = self
651 .storage
652 .as_ref()
653 .expect("validated AgentStorage must be present")
654 .clone();
655 let reopen = matches!(&self.mode, AgentBuildMode::Reopen(_));
656 let info = match &self.mode {
657 AgentBuildMode::Reopen(id) => match storage.agent_info(id.clone()).await {
658 Ok(info)
659 if self
660 .config
661 .as_ref()
662 .expect("build configuration must remain owned")
663 .display_name()
664 .is_some_and(|name| Some(name) != info.display_name()) =>
665 {
666 return Err(self
667 .cleanup_after_error(KitError::conflict(
668 "configured agent display name differs from durable identity",
669 ))
670 .await);
671 }
672 Ok(info) => info,
673 Err(error) => return Err(self.cleanup_after_error(error).await),
674 },
675 AgentBuildMode::Create => match AgentInfo::new(
676 AgentId::new(),
677 self.config
678 .as_ref()
679 .expect("build configuration must remain owned")
680 .display_name()
681 .map(str::to_owned),
682 ) {
683 Ok(info) => info,
684 Err(error) => return Err(self.cleanup_after_error(error).await),
685 },
686 };
687
688 self.claim = match storage.claim_agent(info.id()) {
689 Ok(claim) => Some(claim),
690 Err(error) => return Err(self.cleanup_after_error(error).await),
691 };
692 let knowledge = self
693 .knowledge_database
694 .take()
695 .map(|database| ManagedKnowledge::new(storage.clone(), database));
696 if reopen {
697 if let Err(error) = storage
698 .settle_knowledge_forgets(knowledge.as_ref(), info.id(), lifecycle_deadline)
699 .await
700 {
701 return Err(self.cleanup_after_error(error).await);
702 }
703 if let Some(knowledge) = &knowledge {
704 let owner = KnowledgeOwnerId::Agent {
705 agent_id: info.id().clone(),
706 };
707 let verification = match tokio::time::timeout_at(
708 lifecycle_deadline,
709 knowledge.verify(owner),
710 )
711 .await
712 {
713 Ok(result) => result,
714 Err(_) => Err(KitError::timeout(
715 "knowledge reconciliation deadline elapsed",
716 )),
717 };
718 if let Err(error) = verification {
719 return Err(self.cleanup_after_error(error).await);
720 }
721 }
722 }
723
724 let config = self
725 .config
726 .take()
727 .expect("build configuration must transfer exactly once");
728 let backend = config.backend().clone();
729 let claim = self
730 .claim
731 .take()
732 .expect("validated live-agent claim must transfer exactly once");
733 let built = match backend {
734 BackendConfig::Codex(_) => {
735 let toolsets = std::mem::take(&mut self.toolsets);
736 codex::agent::CodexAgent::build(
737 info, config, storage, knowledge, claim, toolsets, reopen,
738 )
739 .await
740 .map(Agent::codex)
741 }
742 BackendConfig::OpenCode(_) => opencode::agent::OpenCodeAgent::build(
743 info, config, storage, knowledge, claim, reopen,
744 )
745 .await
746 .map(Agent::opencode),
747 BackendConfig::Custom(_) => {
748 let composition = custom::CustomAgent::<S, D, P, T>::compose(
749 self.provider
750 .take()
751 .expect("validated provider must transfer exactly once"),
752 std::mem::take(&mut self.toolsets),
753 );
754 custom::CustomAgent::build(
755 info,
756 config,
757 storage,
758 knowledge,
759 claim,
760 composition,
761 reopen,
762 )
763 .await
764 .map(Agent::custom)
765 }
766 };
767 let agent = match built {
768 Ok(agent) => agent,
769 Err(error) => return Err(self.cleanup_after_error(error).await),
770 };
771 self.agent = Some(agent);
772
773 Ok(self
774 .agent
775 .take()
776 .expect("constructed Agent must publish exactly once"))
777 }
778
779 async fn validate(&mut self) -> Result<(), KitError> {
780 let agent_config = self
781 .config
782 .as_ref()
783 .expect("build configuration must remain owned");
784 agent_config.validate()?;
785 if tokio::time::Instant::now()
786 .checked_add(agent_config.lifecycle_timeout())
787 .is_none()
788 {
789 return Err(KitError::invalid_config(
790 "lifecycle timeout cannot be represented",
791 ));
792 }
793 let has_provider = self.provider.is_some();
794 let has_tools = !self.toolsets.is_empty();
795 match agent_config.backend() {
796 BackendConfig::Codex(_) => {
797 if has_provider {
798 return Err(KitError::invalid_config(
799 "selected runtime does not accept a model runtime",
800 ));
801 }
802 codex::tools::CodexTools::<S, T>::validate_registrations(&self.toolsets).await?;
803 }
804 BackendConfig::OpenCode(_) => {
805 if has_provider {
806 return Err(KitError::invalid_config(
807 "selected runtime does not accept a model runtime",
808 ));
809 }
810 if has_tools {
811 return Err(KitError::unsupported(
812 "selected runtime does not accept executable tools",
813 ));
814 }
815 }
816 BackendConfig::Custom(config) => {
817 if !has_provider {
818 return Err(KitError::invalid_config(
819 "selected runtime requires a configured model runtime",
820 ));
821 }
822 let catalog = self.validate_catalog().await?;
823 Self::validate_custom_selection(config.default_tool_selection(), &catalog)?;
824 }
825 }
826 Ok(())
827 }
828
829 async fn validate_catalog(&self) -> Result<BTreeSet<String>, KitError> {
830 let mut executable_names = BTreeSet::new();
831 for toolset in &self.toolsets {
832 for tool in toolset.list_tools().await? {
833 let name = tool.name.as_str().to_owned();
834 if !executable_names.insert(name.clone()) {
835 return Err(KitError::invalid_config(
836 "configured tool names must be unique",
837 ));
838 }
839 }
840 }
841
842 Ok(executable_names)
843 }
844
845 fn validate_custom_selection(
846 selection: &ToolSelection,
847 executable_names: &BTreeSet<String>,
848 ) -> Result<(), KitError> {
849 selection.validate()?;
850 match selection {
851 ToolSelection::Auto | ToolSelection::None => Ok(()),
852 ToolSelection::Required if executable_names.is_empty() => {
853 Err(KitError::invalid_config(
854 "required tool selection needs at least one configured tool",
855 ))
856 }
857 ToolSelection::Required => Ok(()),
858 ToolSelection::Allowed(names) | ToolSelection::RequiredOneOf(names) => {
859 Self::require_known_names(names, executable_names, "default selected tool")
860 }
861 }
862 }
863
864 fn require_known_names(
865 requested: &[String],
866 available: &BTreeSet<String>,
867 label: &str,
868 ) -> Result<(), KitError> {
869 let missing = requested
870 .iter()
871 .filter(|name| !available.contains(*name))
872 .cloned()
873 .collect::<Vec<_>>();
874 if missing.is_empty() {
875 Ok(())
876 } else {
877 Err(KitError::invalid_config(format!(
878 "{label} names must exist in the supplied Toolsets"
879 )))
880 }
881 }
882
883 async fn cleanup_after_error(mut self, mut primary: KitError) -> KitError {
884 let cleanup_deadline = tokio::time::Instant::now()
885 .checked_add(self.cleanup_timeout)
886 .unwrap_or_else(tokio::time::Instant::now);
887 if let Some(agent) = self.agent.take()
888 && let Err(cleanup_error) = agent.shutdown().await
889 {
890 if self.tracing_enabled {
891 tracing::warn!(
892 cleanup_target = "agent",
893 "agent build cleanup did not complete successfully"
894 );
895 }
896 primary = primary
897 .with_supplementary(cleanup_error.at_phase(crate::error::ErrorPhase::Cleanup));
898 }
899
900 drop(self.claim.take());
901 while let Some(toolset) = self.toolsets.pop() {
902 let cleanup =
903 tokio::time::timeout_at(cleanup_deadline, async { toolset.shutdown().await }).await;
904 if !matches!(cleanup, Ok(Ok(()))) && self.tracing_enabled {
905 tracing::warn!(
906 cleanup_target = "toolset",
907 "agent build cleanup did not complete successfully"
908 );
909 }
910 let error = match cleanup {
911 Ok(Ok(())) => None,
912 Ok(Err(error)) => Some(KitError::from(error)),
913 Err(_) => Some(
914 KitError::timeout("tool cleanup deadline elapsed").with_detail(
915 crate::error::ErrorDetail::DeadlineElapsed {
916 kind: crate::error::DeadlineKind::Lifecycle,
917 timeout: Some(self.cleanup_timeout),
918 },
919 ),
920 ),
921 };
922 if let Some(error) = error {
923 primary =
924 primary.with_supplementary(error.at_phase(crate::error::ErrorPhase::Cleanup));
925 }
926 }
927 if let Some(provider) = self.provider.take() {
928 let cleanup =
929 tokio::time::timeout_at(cleanup_deadline, async { provider.shutdown().await })
930 .await;
931 if !matches!(cleanup, Ok(Ok(()))) && self.tracing_enabled {
932 tracing::warn!(
933 cleanup_target = "provider",
934 "agent build cleanup did not complete successfully"
935 );
936 }
937 let error = match cleanup {
938 Ok(Ok(())) => None,
939 Ok(Err(error)) => Some(KitError::from_provider(&error)),
940 Err(_) => Some(
941 KitError::timeout("provider cleanup deadline elapsed").with_detail(
942 crate::error::ErrorDetail::DeadlineElapsed {
943 kind: crate::error::DeadlineKind::Lifecycle,
944 timeout: Some(self.cleanup_timeout),
945 },
946 ),
947 ),
948 };
949 if let Some(error) = error {
950 primary =
951 primary.with_supplementary(error.at_phase(crate::error::ErrorPhase::Cleanup));
952 }
953 }
954 primary
955 }
956}
957
958struct AgentBuildPublication<A> {
959 result: Mutex<Option<Result<A, KitError>>>,
960 handle: Mutex<Option<JoinHandle<()>>>,
961}
962
963impl<S: StateStoreBackend, D: DatabaseBackend, P: ProviderBackend, T: ToolsetBackend>
964 AgentBuildPublication<Agent<S, D, P, T>>
965{
966 fn new() -> Self {
967 Self {
968 result: Mutex::new(None),
969 handle: Mutex::new(None),
970 }
971 }
972
973 async fn run(owner: AgentBuildOwner<S, D, P, T>) -> Result<Agent<S, D, P, T>, KitError> {
974 let publication = Arc::new(Self::new());
975 let task_publication = Arc::clone(&publication);
976 let (start_sender, start_receiver) = oneshot::channel();
977 let (ready_sender, ready_receiver) = oneshot::channel();
978 let (claimed_sender, claimed_receiver) = oneshot::channel();
979 let handle = tokio::spawn(async move {
980 if start_receiver.await.is_err() {
981 let _ = owner
982 .cleanup_after_error(KitError::invalid_state(
983 "agent build owner could not start",
984 ))
985 .await;
986 task_publication.finish();
987 return;
988 }
989
990 let result = owner.build().await;
991 let successful = result.is_ok();
992 task_publication.publish(result);
993 if ready_sender.send(()).is_err() {
994 if successful {
995 task_publication.cleanup_unclaimed().await;
996 }
997 task_publication.finish();
998 return;
999 }
1000 if successful && claimed_receiver.await.is_err() {
1001 task_publication.cleanup_unclaimed().await;
1002 }
1003 task_publication.finish();
1004 });
1005 publication.install(handle);
1006 start_sender
1007 .send(())
1008 .map_err(|_| KitError::invalid_state("agent build owner could not start"))?;
1009 ready_receiver
1010 .await
1011 .map_err(|_| KitError::invalid_state("agent build owner terminated unexpectedly"))?;
1012
1013 let mut agent = publication.take()?;
1014 agent.retain_storage_claim();
1015 if claimed_sender.send(()).is_err() {
1016 let primary =
1017 KitError::invalid_state("agent build publication terminated unexpectedly")
1018 .with_detail(crate::error::ErrorDetail::TaskTerminated)
1019 .in_phase(crate::error::ErrorPhase::Publication);
1020 return Err(match agent.shutdown().await {
1021 Ok(()) => primary,
1022 Err(cleanup) => {
1023 primary.with_supplementary(cleanup.at_phase(crate::error::ErrorPhase::Cleanup))
1024 }
1025 });
1026 }
1027 Ok(agent)
1028 }
1029
1030 fn install(&self, handle: JoinHandle<()>) {
1031 *self
1032 .handle
1033 .lock()
1034 .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(handle);
1035 }
1036
1037 fn publish(&self, result: Result<Agent<S, D, P, T>, KitError>) {
1038 *self
1039 .result
1040 .lock()
1041 .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(result);
1042 }
1043
1044 fn take(&self) -> Result<Agent<S, D, P, T>, KitError> {
1045 self.result
1046 .lock()
1047 .unwrap_or_else(|poisoned| poisoned.into_inner())
1048 .take()
1049 .ok_or_else(|| KitError::invalid_state("agent build result was not published"))?
1050 }
1051
1052 async fn cleanup_unclaimed(&self) {
1053 let result = self
1054 .result
1055 .lock()
1056 .unwrap_or_else(|poisoned| poisoned.into_inner())
1057 .take();
1058 if let Some(Ok(agent)) = result {
1059 let _ = agent.shutdown().await;
1060 }
1061 }
1062
1063 fn finish(&self) {
1064 self.handle
1065 .lock()
1066 .unwrap_or_else(|poisoned| poisoned.into_inner())
1067 .take();
1068 }
1069}