cairn_runtime/
agent.rs

1use std::collections::BTreeMap;
2use std::sync::Arc;
3use std::time::{Duration, Instant};
4
5use chrono::Utc;
6use serde_json::{Value, json};
7
8use crate::budget::Budget;
9use crate::event::{Event, EventKind};
10use crate::journal::{InMemoryJournal, Journal, JournalError};
11use crate::message::{Content, Message, content_text};
12use futures::StreamExt;
13
14use crate::model::{Model, ModelError, ModelRequest, ModelResponse, StreamEvent, Usage};
15use crate::policy::{Decision, EffectPolicy, Policy};
16use crate::state::{PendingCall, RunState, Status};
17use crate::tool::{Tool, ToolCtx};
18
19#[derive(Debug, thiserror::Error)]
20pub enum AgentError {
21    #[error(transparent)]
22    Journal(#[from] JournalError),
23    #[error("no run with id `{0}`")]
24    UnknownRun(String),
25    #[error("a run with id `{0}` already exists")]
26    RunExists(String),
27    #[error("call `{call_id}` in run `{run_id}` is not awaiting approval")]
28    NotAwaitingApproval { run_id: String, call_id: String },
29    #[error("invalid agent configuration: {0}")]
30    Config(String),
31    #[error("run `{run_id}` is being driven by `{owner}` (lease until {expires_at})")]
32    Leased { run_id: String, owner: String, expires_at: chrono::DateTime<chrono::Utc> },
33    #[error("lost the lease on run `{run_id}`; another worker may have taken over")]
34    LeaseLost { run_id: String },
35}
36
37/// Result of driving a run until it finishes or needs a human.
38#[derive(Debug, Clone)]
39pub struct RunOutcome {
40    pub run_id: String,
41    pub status: Status,
42    pub steps: u32,
43    pub usage: Usage,
44    pub cost_usd: f64,
45    pub awaiting_approval: Vec<PendingCall>,
46    /// The validated final answer, when the agent has a finish schema.
47    pub structured: Option<Value>,
48}
49
50impl RunOutcome {
51    pub fn output(&self) -> Option<&str> {
52        match &self.status {
53            Status::Completed { output } => Some(output),
54            _ => None,
55        }
56    }
57
58    fn from_state(state: &RunState) -> Self {
59        Self {
60            run_id: state.run_id.clone(),
61            status: state.status.clone(),
62            steps: state.steps,
63            usage: state.usage,
64            cost_usd: state.cost_usd,
65            awaiting_approval: state.awaiting_approval(),
66            structured: state.structured_output.clone(),
67        }
68    }
69
70    /// Deserialize the structured final answer into `T`.
71    pub fn output_as<T: serde::de::DeserializeOwned>(&self) -> Option<Result<T, serde_json::Error>> {
72        self.structured.clone().map(serde_json::from_value)
73    }
74}
75
76pub struct Agent {
77    name: String,
78    model: Arc<dyn Model>,
79    system: Option<String>,
80    tools: Vec<Arc<dyn Tool>>,
81    budget: Budget,
82    policy: Arc<dyn Policy>,
83    journal: Arc<dyn Journal>,
84    max_tokens: u32,
85    temperature: Option<f32>,
86    provider_options: BTreeMap<String, Value>,
87    on_delta: Option<DeltaFn>,
88    tool_timeout: Option<Duration>,
89    finish: Option<FinishSchema>,
90    worker_id: String,
91    lease_ttl: Duration,
92}
93
94/// Default lease length; renewed every third of it while a run is driven.
95pub const DEFAULT_LEASE_TTL: Duration = Duration::from_secs(60);
96
97/// Holds a run's lease while it is driven and renews it in the background.
98struct Lease {
99    lost: Arc<std::sync::atomic::AtomicBool>,
100    renewer: tokio::task::JoinHandle<()>,
101}
102
103impl Lease {
104    fn lost(&self) -> bool {
105        self.lost.load(std::sync::atomic::Ordering::SeqCst)
106    }
107}
108
109impl Drop for Lease {
110    fn drop(&mut self) {
111        self.renewer.abort();
112    }
113}
114
115/// A required shape for the run's final answer.
116struct FinishSchema {
117    schema: Value,
118    validator: jsonschema::Validator,
119    retries: u32,
120}
121
122impl FinishSchema {
123    /// The validated JSON, or why the answer does not satisfy the schema.
124    fn check(&self, text: &str) -> Result<Value, String> {
125        let value = crate::output::extract_json(text).ok_or_else(|| "the answer is not valid JSON".to_owned())?;
126        let errors: Vec<String> = self
127            .validator
128            .iter_errors(&value)
129            .take(5)
130            .map(|e| {
131                let path = e.instance_path().to_string();
132                if path.is_empty() { e.to_string() } else { format!("{path}: {e}") }
133            })
134            .collect();
135        if errors.is_empty() { Ok(value) } else { Err(errors.join("; ")) }
136    }
137}
138
139/// Default limit for a single tool call.
140pub const DEFAULT_TOOL_TIMEOUT: Duration = Duration::from_secs(120);
141
142type DeltaFn = Arc<dyn Fn(&StreamEvent) + Send + Sync>;
143
144impl Agent {
145    pub fn builder(name: impl Into<String>) -> AgentBuilder {
146        AgentBuilder {
147            name: name.into(),
148            model: None,
149            system: None,
150            tools: Vec::new(),
151            budget: Budget::default(),
152            policy: Arc::new(EffectPolicy::default()),
153            journal: Arc::new(InMemoryJournal::new()),
154            max_tokens: 4096,
155            temperature: None,
156            provider_options: BTreeMap::new(),
157            on_delta: None,
158            tool_timeout: Some(DEFAULT_TOOL_TIMEOUT),
159            finish_schema: None,
160            finish_retries: 1,
161            worker_id: None,
162            lease_ttl: DEFAULT_LEASE_TTL,
163        }
164    }
165
166    pub fn name(&self) -> &str {
167        &self.name
168    }
169
170    /// A builder pre-filled with this agent's configuration, for deriving
171    /// variants (another model, journal or tool set).
172    pub fn to_builder(&self) -> AgentBuilder {
173        AgentBuilder {
174            name: self.name.clone(),
175            model: Some(self.model.clone()),
176            system: self.system.clone(),
177            tools: self.tools.clone(),
178            budget: self.budget.clone(),
179            policy: self.policy.clone(),
180            journal: self.journal.clone(),
181            max_tokens: self.max_tokens,
182            temperature: self.temperature,
183            provider_options: self.provider_options.clone(),
184            on_delta: self.on_delta.clone(),
185            tool_timeout: self.tool_timeout,
186            finish_schema: self.finish.as_ref().map(|f| f.schema.clone()),
187            finish_retries: self.finish.as_ref().map_or(1, |f| f.retries),
188            worker_id: Some(self.worker_id.clone()),
189            lease_ttl: self.lease_ttl,
190        }
191    }
192
193    pub fn worker_id(&self) -> &str {
194        &self.worker_id
195    }
196
197    pub fn journal(&self) -> &Arc<dyn Journal> {
198        &self.journal
199    }
200
201    /// Start a new run with a user message and drive it.
202    pub async fn run(&self, input: impl Into<String>) -> Result<RunOutcome, AgentError> {
203        self.run_message(Message::user(input)).await
204    }
205
206    pub async fn run_message(&self, input: Message) -> Result<RunOutcome, AgentError> {
207        self.start(uuid::Uuid::now_v7().to_string(), input).await
208    }
209
210    /// Start a new run under an id the caller chooses, and drive it.
211    ///
212    /// For a worker serving an outside job queue: derive the id from the job
213    /// (e.g. `"task-42:1"`), and after a crash the run can be found and
214    /// [`resume`](Self::resume)d without having stored a generated id
215    /// anywhere first. Fails with [`AgentError::RunExists`] if the id is
216    /// already in the journal.
217    pub async fn run_with_id(&self, run_id: &str, input: impl Into<String>) -> Result<RunOutcome, AgentError> {
218        // The id names a file in the JSONL journal, so it must be a plain name:
219        // nothing that could climb out of the directory or trip up a filesystem.
220        let plain = !run_id.is_empty()
221            && run_id.len() <= 128
222            && !run_id.starts_with('.')
223            && run_id.chars().all(|c| c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.'));
224        if !plain {
225            return Err(AgentError::Config(format!(
226                "run id `{run_id}` must be 1–128 characters of A–Z, a–z, 0–9, '-', '_' or '.', not starting with '.'"
227            )));
228        }
229        if !self.journal.load(run_id).await?.is_empty() {
230            return Err(AgentError::RunExists(run_id.to_owned()));
231        }
232        self.start(run_id.to_owned(), Message::user(input)).await
233    }
234
235    async fn start(&self, run_id: String, input: Message) -> Result<RunOutcome, AgentError> {
236        let first = Event {
237            run_id: run_id.clone(),
238            seq: 0,
239            at: Utc::now(),
240            kind: EventKind::RunStarted { agent: self.name.clone(), input },
241        };
242        // Two workers racing to start the same chosen id: the journal's
243        // sequence check lets exactly one in.
244        match self.journal.append(&first).await {
245            Ok(()) => {}
246            Err(JournalError::Conflict { .. }) => return Err(AgentError::RunExists(run_id)),
247            Err(e) => return Err(e.into()),
248        }
249        let state = RunState::from_events(std::slice::from_ref(&first)).expect("one event");
250        tracing::info!(run_id = %run_id, agent = %self.name, "run started");
251        self.drive_leased(state).await
252    }
253
254    /// Continue a run from its journal, e.g. after a crash or an approval.
255    /// Completed tool calls are never re-executed; a call that was in flight
256    /// during a crash runs again with the same idempotency key.
257    ///
258    /// Takes the run's lease first, so two workers never drive the same run;
259    /// the loser gets [`AgentError::Leased`].
260    pub async fn resume(&self, run_id: &str) -> Result<RunOutcome, AgentError> {
261        let lease = self.take_lease(run_id).await?;
262        // Load after taking the lease so we see everything the previous holder wrote.
263        let result = match self.state(run_id).await {
264            Ok(state) => self.drive(state, &lease).await,
265            Err(e) => Err(e),
266        };
267        self.end_lease(run_id, lease).await;
268        result
269    }
270
271    async fn drive_leased(&self, state: RunState) -> Result<RunOutcome, AgentError> {
272        let run_id = state.run_id.clone();
273        let lease = self.take_lease(&run_id).await?;
274        let result = self.drive(state, &lease).await;
275        self.end_lease(&run_id, lease).await;
276        result
277    }
278
279    async fn take_lease(&self, run_id: &str) -> Result<Lease, AgentError> {
280        if let Err(holder) = self.journal.acquire_lease(run_id, &self.worker_id, self.lease_ttl).await? {
281            return Err(AgentError::Leased {
282                run_id: run_id.to_owned(),
283                owner: holder.owner,
284                expires_at: holder.expires_at,
285            });
286        }
287        let lost = Arc::new(std::sync::atomic::AtomicBool::new(false));
288        let (journal, owner, ttl, flag, id) =
289            (self.journal.clone(), self.worker_id.clone(), self.lease_ttl, lost.clone(), run_id.to_owned());
290        let renewer = tokio::spawn(async move {
291            let mut tick = tokio::time::interval((ttl / 3).max(Duration::from_millis(10)));
292            tick.tick().await;
293            loop {
294                tick.tick().await;
295                if !matches!(journal.renew_lease(&id, &owner, ttl).await, Ok(true)) {
296                    tracing::warn!(run_id = %id, %owner, "lost run lease");
297                    flag.store(true, std::sync::atomic::Ordering::SeqCst);
298                    return;
299                }
300            }
301        });
302        Ok(Lease { lost, renewer })
303    }
304
305    async fn end_lease(&self, run_id: &str, lease: Lease) {
306        drop(lease);
307        if let Err(e) = self.journal.release_lease(run_id, &self.worker_id).await {
308            tracing::warn!(run_id, error = %e, "could not release run lease; it will expire");
309        }
310    }
311
312    /// Current state of a run, rebuilt from its journal.
313    pub async fn state(&self, run_id: &str) -> Result<RunState, AgentError> {
314        let events = self.journal.load(run_id).await?;
315        RunState::from_events(&events).ok_or_else(|| AgentError::UnknownRun(run_id.to_owned()))
316    }
317
318    /// Record a human decision for a call that is awaiting approval.
319    /// Call [`Agent::resume`] afterwards to continue the run.
320    pub async fn decide(
321        &self,
322        run_id: &str,
323        call_id: &str,
324        approved: bool,
325        by: Option<&str>,
326        note: Option<&str>,
327    ) -> Result<(), AgentError> {
328        let mut state = self.state(run_id).await?;
329        if !state.pending.iter().any(|p| p.id == call_id && p.awaiting_approval) {
330            return Err(AgentError::NotAwaitingApproval { run_id: run_id.to_owned(), call_id: call_id.to_owned() });
331        }
332        self.emit(
333            &mut state,
334            EventKind::ApprovalDecided {
335                call_id: call_id.to_owned(),
336                approved,
337                by: by.map(str::to_owned),
338                note: note.map(str::to_owned),
339            },
340        )
341        .await
342    }
343
344    async fn emit(&self, state: &mut RunState, kind: EventKind) -> Result<(), AgentError> {
345        let event = Event { run_id: state.run_id.clone(), seq: state.next_seq, at: Utc::now(), kind };
346        tracing::debug!(run_id = %event.run_id, seq = event.seq, event = event.kind.name(), "append");
347        self.journal.append(&event).await?;
348        state.apply(&event);
349        Ok(())
350    }
351
352    fn find_tool(&self, name: &str) -> Option<&Arc<dyn Tool>> {
353        self.tools.iter().find(|t| t.spec().name == name)
354    }
355
356    async fn drive(&self, mut state: RunState, lease: &Lease) -> Result<RunOutcome, AgentError> {
357        loop {
358            if state.status.is_terminal() {
359                return Ok(RunOutcome::from_state(&state));
360            }
361            if lease.lost() {
362                return Err(AgentError::LeaseLost { run_id: state.run_id.clone() });
363            }
364
365            if !state.pending.is_empty() {
366                self.handle_tool_calls(&mut state).await?;
367                if state.status == Status::AwaitingApproval {
368                    return Ok(RunOutcome::from_state(&state));
369                }
370                continue;
371            }
372
373            if let Some(reason) = self.budget.exceeded(state.steps, state.usage.total(), state.cost_usd) {
374                self.emit(&mut state, EventKind::BudgetExceeded { reason }).await?;
375                continue;
376            }
377
378            self.model_turn(&mut state).await?;
379        }
380    }
381
382    async fn model_turn(&self, state: &mut RunState) -> Result<(), AgentError> {
383        let mut request = ModelRequest {
384            system: self.system_prompt(),
385            messages: state.messages.clone(),
386            tools: self.tools.iter().map(|t| t.spec().clone()).collect(),
387            max_tokens: self.max_tokens,
388            temperature: self.temperature,
389            provider_options: self.provider_options.clone(),
390        };
391        // Shrink this step's answer limit to what the remaining budget can
392        // pay for, so one call cannot overshoot it.
393        let est_input = estimate_input_tokens(&request);
394        match self.budget.output_cap(self.max_tokens, state.usage.total(), state.cost_usd, est_input, |i, o| {
395            self.model.max_call_cost(i, o)
396        }) {
397            Ok(cap) => request.max_tokens = cap,
398            Err(reason) => return self.emit(state, EventKind::BudgetExceeded { reason }).await,
399        }
400        let step = state.steps + 1;
401        let started = Instant::now();
402
403        let result = match &self.on_delta {
404            Some(on_delta) => self.stream_turn(&request, on_delta).await,
405            None => self.model.complete(&request).await,
406        };
407        match result {
408            Ok(response) => {
409                let cost_usd = self.model.cost_usd(&response);
410                let wants_tools = response.content.iter().any(|c| matches!(c, Content::ToolUse { .. }));
411                let text = content_text(&response.content);
412                let duration_ms = started.elapsed().as_millis() as u64;
413                self.emit(state, EventKind::ModelCalled { step, response, cost_usd, duration_ms }).await?;
414                if !wants_tools {
415                    self.finish(state, step, text).await?;
416                }
417            }
418            Err(e) => {
419                let error = e.to_string();
420                let (usage, spent) = e.spent();
421                // Unpriced spend can only happen without a USD budget.
422                let cost_usd = spent.or_else(|| self.model.pricing().map(|p| p.cost(&usage)));
423                self.emit(state, EventKind::ModelFailed { step, error: error.clone(), usage, cost_usd }).await?;
424                self.emit(state, EventKind::RunFailed { error }).await?;
425            }
426        }
427        Ok(())
428    }
429
430    /// The configured system prompt, plus — when a finish schema is set —
431    /// the instruction stating the required final-answer format. Both are
432    /// part of the request and visible in traces.
433    fn system_prompt(&self) -> Option<String> {
434        let Some(finish) = &self.finish else { return self.system.clone() };
435        let rule = format!(
436            "When you have finished, reply with only a JSON object (no other text) that matches this JSON Schema:\n{}",
437            finish.schema
438        );
439        Some(match &self.system {
440            Some(s) => format!("{s}\n\n{rule}"),
441            None => rule,
442        })
443    }
444
445    /// Complete the run, enforcing the finish schema if there is one.
446    async fn finish(&self, state: &mut RunState, step: u32, text: String) -> Result<(), AgentError> {
447        let Some(finish) = &self.finish else {
448            return self.emit(state, EventKind::RunCompleted { output: text, structured: None }).await;
449        };
450        match finish.check(&text) {
451            Ok(value) => self.emit(state, EventKind::RunCompleted { output: text, structured: Some(value) }).await,
452            Err(reason) if state.finish_rejections < finish.retries => {
453                self.emit(state, EventKind::FinishRejected { step, reason }).await
454            }
455            Err(reason) => {
456                let error = format!(
457                    "final answer did not match the finish schema after {} retr{}: {reason}",
458                    finish.retries,
459                    if finish.retries == 1 { "y" } else { "ies" }
460                );
461                self.emit(state, EventKind::RunFailed { error }).await
462            }
463        }
464    }
465
466    /// Streams deltas to the observer. Only the completed response is
467    /// journaled, so replay and resume are identical with or without streaming.
468    async fn stream_turn(&self, request: &ModelRequest, on_delta: &DeltaFn) -> Result<ModelResponse, ModelError> {
469        let mut stream = self.model.stream(request).await?;
470        while let Some(event) = stream.next().await {
471            match event? {
472                StreamEvent::Completed(response) => return Ok(response),
473                delta => on_delta(&delta),
474            }
475        }
476        Err(ModelError::Decode("stream ended before the response completed".into()))
477    }
478
479    async fn handle_tool_calls(&self, state: &mut RunState) -> Result<(), AgentError> {
480        for call in state.pending.clone() {
481            if call.awaiting_approval {
482                continue;
483            }
484            if let Some(&approved) = state.approvals.get(&call.id) {
485                if approved {
486                    self.execute(state, &call).await?;
487                } else {
488                    let kind = EventKind::ToolDenied {
489                        call_id: call.id.clone(),
490                        tool: call.name.clone(),
491                        reason: "rejected by approver".into(),
492                    };
493                    self.emit(state, kind).await?;
494                }
495                continue;
496            }
497            let Some(tool) = self.find_tool(&call.name) else {
498                let kind = EventKind::ToolCompleted {
499                    call_id: call.id.clone(),
500                    tool: call.name.clone(),
501                    output: json!({ "error": format!("unknown tool `{}`", call.name) }),
502                    is_error: true,
503                    duration_ms: 0,
504                };
505                self.emit(state, kind).await?;
506                continue;
507            };
508            let kind = match self.policy.decide(tool.spec(), &call.input) {
509                Decision::Allow => {
510                    self.execute(state, &call).await?;
511                    continue;
512                }
513                Decision::Deny { reason } => {
514                    EventKind::ToolDenied { call_id: call.id.clone(), tool: call.name.clone(), reason }
515                }
516                Decision::RequireApproval { reason } => EventKind::ApprovalRequested {
517                    call_id: call.id.clone(),
518                    tool: call.name.clone(),
519                    input: call.input.clone(),
520                    reason,
521                },
522            };
523            self.emit(state, kind).await?;
524        }
525        Ok(())
526    }
527
528    async fn execute(&self, state: &mut RunState, call: &PendingCall) -> Result<(), AgentError> {
529        let tool = self.find_tool(&call.name).cloned();
530        let ctx = ToolCtx::new(&state.run_id, &call.id);
531        let started = Instant::now();
532        let (output, is_error) = match tool {
533            Some(tool) => {
534                let limit = tool.timeout().or(self.tool_timeout);
535                let fut = tool.call(call.input.clone(), &ctx);
536                let result = match limit {
537                    Some(limit) => tokio::time::timeout(limit, fut).await.unwrap_or_else(|_| {
538                        tracing::warn!(tool = %call.name, ?limit, "tool timed out");
539                        Err(crate::tool::ToolError(format!(
540                            "tool `{}` timed out after {:.1}s and was cancelled",
541                            call.name,
542                            limit.as_secs_f64()
543                        )))
544                    }),
545                    None => fut.await,
546                };
547                match result {
548                    Ok(v) => (v, false),
549                    Err(e) => (json!({ "error": e.0 }), true),
550                }
551            }
552            None => (json!({ "error": format!("unknown tool `{}`", call.name) }), true),
553        };
554        let kind = EventKind::ToolCompleted {
555            call_id: call.id.clone(),
556            tool: call.name.clone(),
557            output,
558            is_error,
559            duration_ms: started.elapsed().as_millis() as u64,
560        };
561        self.emit(state, kind).await
562    }
563}
564
565/// Conservative input-size estimate: about 3 characters per token over the
566/// serialized system prompt, messages and tool definitions.
567fn estimate_input_tokens(request: &ModelRequest) -> u64 {
568    let chars = request.system.as_deref().map_or(0, str::len)
569        + serde_json::to_string(&request.messages).map_or(0, |s| s.len())
570        + serde_json::to_string(&request.tools).map_or(0, |s| s.len());
571    (chars as u64).div_ceil(3)
572}
573
574pub struct AgentBuilder {
575    name: String,
576    model: Option<Arc<dyn Model>>,
577    system: Option<String>,
578    tools: Vec<Arc<dyn Tool>>,
579    budget: Budget,
580    policy: Arc<dyn Policy>,
581    journal: Arc<dyn Journal>,
582    max_tokens: u32,
583    temperature: Option<f32>,
584    provider_options: BTreeMap<String, Value>,
585    on_delta: Option<DeltaFn>,
586    tool_timeout: Option<Duration>,
587    finish_schema: Option<Value>,
588    finish_retries: u32,
589    worker_id: Option<String>,
590    lease_ttl: Duration,
591}
592
593impl AgentBuilder {
594    pub fn model(mut self, model: impl Model + 'static) -> Self {
595        self.model = Some(Arc::new(model));
596        self
597    }
598
599    /// Replace the tool set.
600    pub fn replace_tools(mut self, tools: Vec<Arc<dyn Tool>>) -> Self {
601        self.tools = tools;
602        self
603    }
604
605    pub(crate) fn tool_list(&self) -> &[Arc<dyn Tool>] {
606        &self.tools
607    }
608
609    /// Stop streaming deltas (e.g. for background replays).
610    pub fn without_on_delta(mut self) -> Self {
611        self.on_delta = None;
612        self
613    }
614
615    pub fn system(mut self, prompt: impl Into<String>) -> Self {
616        self.system = Some(prompt.into());
617        self
618    }
619
620    pub fn tool(mut self, tool: impl Tool + 'static) -> Self {
621        self.tools.push(Arc::new(tool));
622        self
623    }
624
625    /// Add several tools at once (e.g. all tools from an MCP server).
626    pub fn tools<T: Tool + 'static>(mut self, tools: impl IntoIterator<Item = T>) -> Self {
627        self.tools.extend(tools.into_iter().map(|t| Arc::new(t) as Arc<dyn Tool>));
628        self
629    }
630
631    pub fn budget(mut self, budget: Budget) -> Self {
632        self.budget = budget;
633        self
634    }
635
636    pub fn policy(mut self, policy: impl Policy + 'static) -> Self {
637        self.policy = Arc::new(policy);
638        self
639    }
640
641    pub fn journal(mut self, journal: impl Journal + 'static) -> Self {
642        self.journal = Arc::new(journal);
643        self
644    }
645
646    pub fn shared_journal(mut self, journal: Arc<dyn Journal>) -> Self {
647        self.journal = journal;
648        self
649    }
650
651    pub fn max_tokens(mut self, n: u32) -> Self {
652        self.max_tokens = n;
653        self
654    }
655
656    pub fn temperature(mut self, t: f32) -> Self {
657        self.temperature = Some(t);
658        self
659    }
660
661    /// Provider-specific request options, e.g.
662    /// `.provider_option("anthropic", json!({"thinking": {"type": "enabled", "budget_tokens": 4096}}))`.
663    pub fn provider_option(mut self, provider: &str, options: Value) -> Self {
664        self.provider_options.insert(provider.to_owned(), options);
665        self
666    }
667
668    /// Stream model output as it is generated (text, thinking, tool-call
669    /// arguments). Enables streaming calls to the model.
670    pub fn on_delta(mut self, f: impl Fn(&StreamEvent) + Send + Sync + 'static) -> Self {
671        self.on_delta = Some(Arc::new(f));
672        self
673    }
674
675    /// Default limit for each tool call (120 s unless changed). Tools can
676    /// override it with `with_timeout`. A timed-out call is cancelled and
677    /// reported to the model as an error; the run continues.
678    pub fn tool_timeout(mut self, limit: Duration) -> Self {
679        self.tool_timeout = Some(limit);
680        self
681    }
682
683    /// Require the run's final answer to be JSON matching `schema` (JSON
684    /// Schema). The schema is added to the system prompt; an answer that does
685    /// not match gets [`finish_retries`](Self::finish_retries) corrective
686    /// retries (default 1), after which the run fails. The validated value
687    /// is available as [`RunOutcome::structured`].
688    pub fn finish_schema(mut self, schema: Value) -> Self {
689        self.finish_schema = Some(schema);
690        self
691    }
692
693    /// [`finish_schema`](Self::finish_schema) derived from a Rust type.
694    pub fn finish_as<T: schemars::JsonSchema>(self) -> Self {
695        let mut schema = serde_json::to_value(schemars::schema_for!(T)).unwrap_or(Value::Null);
696        if let Value::Object(map) = &mut schema {
697            map.remove("$schema");
698        }
699        self.finish_schema(schema)
700    }
701
702    /// Name this worker in run leases (default: a random `worker-<uuid>`).
703    pub fn worker_id(mut self, id: impl Into<String>) -> Self {
704        self.worker_id = Some(id.into());
705        self
706    }
707
708    /// How long a run lease lasts without renewal (default 60 s). A crashed
709    /// worker's runs can be taken over after this long.
710    pub fn lease_ttl(mut self, ttl: Duration) -> Self {
711        self.lease_ttl = ttl;
712        self
713    }
714
715    /// Corrective retries allowed for a final answer that fails the schema.
716    pub fn finish_retries(mut self, n: u32) -> Self {
717        self.finish_retries = n;
718        self
719    }
720
721    /// Remove the default tool timeout (per-tool timeouts still apply).
722    pub fn without_tool_timeout(mut self) -> Self {
723        self.tool_timeout = None;
724        self
725    }
726
727    pub fn build(self) -> Result<Agent, AgentError> {
728        let model = self.model.ok_or_else(|| AgentError::Config("a model is required".into()))?;
729        if self.budget.max_usd.is_some() {
730            let unpriced = model.unpriced_models();
731            if !unpriced.is_empty() {
732                return Err(AgentError::Config(format!(
733                    "max_usd is set but these models have no pricing, so spend could not be counted: {}. \
734                     Set pricing (e.g. Priced::new(model, Pricing::new(input, output))); \
735                     use Pricing::new(0.0, 0.0) for a model that is genuinely free",
736                    unpriced.join(", ")
737                )));
738            }
739        }
740        let finish = match self.finish_schema {
741            Some(schema) => {
742                let validator = jsonschema::validator_for(&schema)
743                    .map_err(|e| AgentError::Config(format!("invalid finish schema: {e}")))?;
744                Some(FinishSchema { schema, validator, retries: self.finish_retries })
745            }
746            None => None,
747        };
748        let mut seen = std::collections::HashSet::new();
749        for t in &self.tools {
750            if !seen.insert(t.spec().name.clone()) {
751                return Err(AgentError::Config(format!("duplicate tool name `{}`", t.spec().name)));
752            }
753        }
754        Ok(Agent {
755            name: self.name,
756            model,
757            system: self.system,
758            tools: self.tools,
759            budget: self.budget,
760            policy: self.policy,
761            journal: self.journal,
762            max_tokens: self.max_tokens,
763            temperature: self.temperature,
764            provider_options: self.provider_options,
765            on_delta: self.on_delta,
766            tool_timeout: self.tool_timeout,
767            finish,
768            worker_id: self.worker_id.unwrap_or_else(|| format!("worker-{}", uuid::Uuid::now_v7())),
769            lease_ttl: self.lease_ttl,
770        })
771    }
772}
cairn_runtime/agent.rs

cairn_runtime/
agent.rs
