runifold_agent/agent/
execution.rs

1//! Canonical Agent execution engine and its private runtime helpers.
2
3use super::checkpointing::{
4    AgentProgress, save_checkpoint, validate_exact_usage, validate_usage_floor,
5};
6use super::observability::{consume_budget, emit_usage, record_domain, terminal_event};
7use super::{
8    Agent, AgentCheckpoint, AgentCheckpointPhase, AgentCheckpointState, AgentError,
9    AgentEventStream, AgentFuture, AgentObserver, AgentOutcome, AgentStreamEvent, Arc,
10    BufferedObserver, CheckpointCursor, ContentPart, DurableConversationCheckpoint, Either,
11    EventId, Instant, LifecycleEvent, Message, ModelCallContext, ModelError, ModelErrorKind,
12    ModelRequest, ModelResponse, ModelStreamAccumulator, NoopObserver, ResumePolicy, Role,
13    RunContext, RunEventKind, StreamExt, ToolCall, Usage, emit_agent_event, select,
14};
15use crate::conversation::{
16    AgentConversationError, AgentConversationOutcome, AutomaticConversationSummary,
17    ConversationAppend, ConversationContextPolicy, ConversationId, ConversationStore,
18    ConversationSummaryCommit, ConversationSummaryRequest, DurableConversationCommit,
19    DurableConversationRequest, DurableConversationStore, MemoryNamespace, SemanticMemoryQuery,
20    is_transient_context, semantic_memory_message, summary_message,
21};
22use runifold_core::{CheckpointId, CheckpointStore};
23use runifold_retrieval::RetrievalContext;
24
25impl Agent {
26    /// Runs a user text turn with a default root context.
27    ///
28    /// This is the ergonomic surface for one-off prompts. It grants only
29    /// callables registered on this Agent and applies no hard budget limit.
30    /// Use [`Self::run`] when the caller must provide explicit authority,
31    /// budget, deadline, observability, or run-tree identity.
32    pub fn prompt<'a>(
33        &'a self,
34        input: impl Into<String> + Send + 'a,
35    ) -> AgentFuture<'a, Result<AgentOutcome, AgentError>> {
36        let input = input.into();
37        Box::pin(async move {
38            let run = self.default_run_context();
39            self.run(input, &run).await
40        })
41    }
42
43    /// Runs an ergonomic prompt and returns only model-visible text.
44    ///
45    /// Rich content, usage, warnings, the canonical transcript, and provider
46    /// events are intentionally discarded. Use [`Self::prompt`] when that
47    /// information matters.
48    pub fn prompt_text<'a>(
49        &'a self,
50        input: impl Into<String> + Send + 'a,
51    ) -> AgentFuture<'a, Result<String, AgentError>> {
52        let input = input.into();
53        Box::pin(async move { self.prompt(input).await.map(AgentOutcome::into_text) })
54    }
55
56    /// Runs a user text turn inside an existing runtime context.
57    pub fn run<'a>(
58        &'a self,
59        input: impl Into<String> + Send + 'a,
60        run: &'a RunContext,
61    ) -> AgentFuture<'a, Result<AgentOutcome, AgentError>> {
62        let input = input.into();
63        let state = self.initial_state(input, run.root_run_id().to_string());
64        Box::pin(async move {
65            self.execute_state(state, run, None, Arc::new(NoopObserver), true, true)
66                .await
67        })
68    }
69
70    /// Runs and atomically commits one bounded multi-turn conversation.
71    ///
72    /// Transcript messages remain append-only. Execution-journal events stay
73    /// in [`runifold_core::Journal`], summaries remain lossy derived views,
74    /// and semantic memory is injected only as explicitly untrusted context.
75    pub fn run_conversation<'a>(
76        &'a self,
77        input: impl Into<String> + Send + 'a,
78        run: &'a RunContext,
79        store: &'a dyn ConversationStore,
80        conversation_id: ConversationId,
81        namespace: MemoryNamespace,
82        policy: ConversationContextPolicy,
83    ) -> AgentFuture<'a, Result<AgentConversationOutcome, AgentConversationError>> {
84        let input = input.into();
85        Box::pin(async move {
86            store.create(conversation_id, namespace.clone()).await?;
87            let view = store
88                .load_view(
89                    conversation_id,
90                    namespace.clone(),
91                    policy.window,
92                    policy.summary_batch,
93                )
94                .await?;
95            if view.requires_summary() {
96                return Err(AgentConversationError::SummaryRequired {
97                    conversation_id,
98                    buffered_entries: u64::try_from(view.summary_buffer.len())
99                        .unwrap_or(u64::MAX)
100                        .saturating_add(view.summary_backlog),
101                });
102            }
103            let mut transcript = self.instructions.clone();
104            if let Some(summary) = &view.summary {
105                transcript.push(summary_message(summary));
106            }
107            if let Some(limit) = policy.semantic_memory_limit {
108                let query =
109                    SemanticMemoryQuery::new(namespace.clone(), input.clone(), limit.get())?;
110                let search = store
111                    .search_memory_scoped(query, RetrievalContext::for_run(run))
112                    .await?;
113                if search.usage != Usage::default() {
114                    consume_budget(run, search.usage, None).map_err(AgentConversationError::Run)?;
115                }
116                if let Some(message) = semantic_memory_message(&search.memories) {
117                    transcript.push(message);
118                }
119            }
120            transcript.extend(view.window.iter().map(|entry| entry.message.clone()));
121            let persisted_prefix_len = transcript.len();
122            transcript.push(Message::user(input));
123            let state =
124                self.initial_state_from_transcript(transcript, run.root_run_id().to_string());
125            let outcome = self
126                .execute_state(state, run, None, Arc::new(NoopObserver), true, true)
127                .await
128                .map_err(AgentConversationError::Run)?;
129            let messages = outcome
130                .transcript
131                .iter()
132                .skip(persisted_prefix_len)
133                .filter(|message| !is_transient_context(message))
134                .cloned()
135                .collect();
136            let append = ConversationAppend {
137                conversation_id,
138                expected_version: view.version,
139                messages,
140            };
141            match store.append(namespace, append).await {
142                Ok(conversation_version) => Ok(AgentConversationOutcome {
143                    outcome,
144                    conversation_version,
145                }),
146                Err(source) => Err(AgentConversationError::Commit {
147                    source,
148                    outcome: Box::new(outcome),
149                }),
150            }
151        })
152    }
153
154    /// Summarizes an overflowing prefix before running a conversational turn.
155    ///
156    /// Summary generation uses the supplied [`AutomaticConversationSummary`]
157    /// and the same [`RunContext`], preserving cancellation, deadline, budget,
158    /// and journal behavior. The immutable transcript is never rewritten.
159    pub fn run_conversation_with_summary<'a>(
160        &'a self,
161        input: impl Into<String> + Send + 'a,
162        run: &'a RunContext,
163        store: &'a dyn ConversationStore,
164        conversation_id: ConversationId,
165        namespace: MemoryNamespace,
166        automatic_summary: AutomaticConversationSummary<'a>,
167    ) -> AgentFuture<'a, Result<AgentConversationOutcome, AgentConversationError>> {
168        let input = input.into();
169        Box::pin(async move {
170            let policy = automatic_summary.context;
171            store.create(conversation_id, namespace.clone()).await?;
172            for pass in 0..automatic_summary.max_passes.get() {
173                let view = store
174                    .load_view(
175                        conversation_id,
176                        namespace.clone(),
177                        policy.window,
178                        policy.summary_batch,
179                    )
180                    .await?;
181                let Some(through_sequence) = view.summary_buffer.last().map(|entry| entry.sequence)
182                else {
183                    break;
184                };
185                let summary_backlog = view.summary_backlog;
186                let summary = automatic_summary
187                    .summarizer
188                    .summarize(
189                        ConversationSummaryRequest {
190                            transcript_version: view.version,
191                            previous_summary: view.summary,
192                            entries: view.summary_buffer,
193                        },
194                        run,
195                    )
196                    .await?;
197                store
198                    .commit_summary(
199                        namespace.clone(),
200                        ConversationSummaryCommit {
201                            conversation_id,
202                            expected_version: view.version,
203                            through_sequence,
204                            content: summary,
205                        },
206                    )
207                    .await?;
208                if summary_backlog == 0 {
209                    break;
210                }
211                if pass + 1 == automatic_summary.max_passes.get() {
212                    return Err(AgentConversationError::SummaryPassLimitExceeded {
213                        conversation_id,
214                        remaining_entries: summary_backlog,
215                    });
216                }
217            }
218            self.run_conversation(input, run, store, conversation_id, namespace, policy)
219                .await
220        })
221    }
222
223    /// Streams real-time events while driving the canonical Agent loop.
224    pub fn stream<'a>(
225        &'a self,
226        input: impl Into<String> + Send + 'a,
227        run: &'a RunContext,
228    ) -> AgentEventStream<'a> {
229        let state = self.initial_state(input.into(), run.root_run_id().to_string());
230        let observer = BufferedObserver::default();
231        let events = observer.events();
232        let execution =
233            Box::pin(self.execute_state(state, run, None, Arc::new(observer), true, true));
234        AgentEventStream::new(execution, events)
235    }
236
237    /// Runs one conversational turn with atomic transcript and checkpoint commit.
238    ///
239    /// Intermediate checkpoints are written ahead of model and callable work.
240    /// The terminal checkpoint and transcript append are committed together by
241    /// [`DurableConversationStore`].
242    pub fn run_durable_conversation<'a>(
243        &'a self,
244        input: impl Into<String> + Send + 'a,
245        run: &'a RunContext,
246        store: Arc<dyn DurableConversationStore>,
247        request: DurableConversationRequest,
248    ) -> AgentFuture<'a, Result<AgentConversationOutcome, AgentConversationError>> {
249        let input = input.into();
250        Box::pin(async move {
251            let DurableConversationRequest {
252                checkpoint_id,
253                conversation_id,
254                namespace,
255                policy,
256            } = request;
257            store.create(conversation_id, namespace.clone()).await?;
258            let view = store
259                .load_view(
260                    conversation_id,
261                    namespace.clone(),
262                    policy.window,
263                    policy.summary_batch,
264                )
265                .await?;
266            if view.requires_summary() {
267                return Err(AgentConversationError::SummaryRequired {
268                    conversation_id,
269                    buffered_entries: u64::try_from(view.summary_buffer.len())
270                        .unwrap_or(u64::MAX)
271                        .saturating_add(view.summary_backlog),
272                });
273            }
274            let mut transcript = self.instructions.clone();
275            if let Some(summary) = &view.summary {
276                transcript.push(summary_message(summary));
277            }
278            if let Some(limit) = policy.semantic_memory_limit {
279                let query =
280                    SemanticMemoryQuery::new(namespace.clone(), input.clone(), limit.get())?;
281                let search = store
282                    .search_memory_scoped(query, RetrievalContext::for_run(run))
283                    .await?;
284                if search.usage != Usage::default() {
285                    consume_budget(run, search.usage, None).map_err(AgentConversationError::Run)?;
286                }
287                if let Some(message) = semantic_memory_message(&search.memories) {
288                    transcript.push(message);
289                }
290            }
291            transcript.extend(view.window.iter().map(|entry| entry.message.clone()));
292            let persisted_prefix_len = u64::try_from(transcript.len()).map_err(|_| {
293                AgentConversationError::Run(checkpoint_payload_error(
294                    "conversation context length exceeds durable checkpoint range",
295                ))
296            })?;
297            transcript.push(Message::user(input));
298            let durable = DurableConversationCheckpoint {
299                conversation_id,
300                namespace,
301                expected_version: view.version,
302                persisted_prefix_len,
303            };
304            let mut state =
305                self.initial_state_from_transcript(transcript, checkpoint_id.to_string());
306            state.durable_conversation = Some(durable.clone());
307            state.usage = run.budget().usage();
308            let checkpoint_store: Arc<dyn CheckpointStore> = store.clone();
309            let checkpoint = AgentCheckpoint::existing(checkpoint_id, checkpoint_store);
310            let mut cursor = CheckpointCursor::create(&checkpoint, run, &state)
311                .map_err(AgentConversationError::Run)?;
312            let outcome = self
313                .execute_state(
314                    state,
315                    run,
316                    Some(&mut cursor),
317                    Arc::new(NoopObserver),
318                    true,
319                    false,
320                )
321                .await
322                .map_err(AgentConversationError::Run)?;
323            self.commit_durable_outcome(store.as_ref(), run, &cursor, durable, outcome)
324                .await
325        })
326    }
327
328    /// Resumes a durable conversational turn from its write-ahead checkpoint.
329    pub fn resume_durable_conversation<'a>(
330        &'a self,
331        store: Arc<dyn DurableConversationStore>,
332        checkpoint_id: CheckpointId,
333        run: &'a RunContext,
334        policy: ResumePolicy,
335    ) -> AgentFuture<'a, Result<AgentConversationOutcome, AgentConversationError>> {
336        Box::pin(async move {
337            let checkpoint_store: Arc<dyn CheckpointStore> = store.clone();
338            let checkpoint = AgentCheckpoint::existing(checkpoint_id, checkpoint_store);
339            let (envelope, mut state) = checkpoint
340                .load()
341                .map_err(AgentError::from)
342                .map_err(AgentConversationError::Run)?;
343            self.validate_checkpoint_identity(&state)
344                .map_err(AgentConversationError::Run)?;
345            let durable = state.durable_conversation.clone().ok_or_else(|| {
346                AgentConversationError::Run(checkpoint_payload_error(
347                    "checkpoint is not a durable conversation turn",
348                ))
349            })?;
350            if let Some(outcome) = state.outcome() {
351                let conversation_version = durable
352                    .expected_version
353                    .get()
354                    .checked_add(1)
355                    .map(crate::ConversationVersion::new)
356                    .ok_or_else(|| {
357                        AgentConversationError::Run(checkpoint_payload_error(
358                            "durable conversation version overflow",
359                        ))
360                    })?;
361                return Ok(AgentConversationOutcome {
362                    outcome,
363                    conversation_version,
364                });
365            }
366            if let AgentCheckpointPhase::TurnInFlight { turn } = state.phase {
367                if policy == ResumePolicy::RejectAmbiguous {
368                    return Err(AgentConversationError::Run(
369                        AgentError::AmbiguousCheckpoint { turn },
370                    ));
371                }
372                validate_usage_floor(state.usage, run.budget().usage())
373                    .map_err(AgentConversationError::Run)?;
374                state.usage = run.budget().usage();
375                state.phase = AgentCheckpointPhase::ReadyForTurn;
376            } else {
377                validate_exact_usage(state.usage, run.budget().usage())
378                    .map_err(AgentConversationError::Run)?;
379            }
380            let mut cursor = CheckpointCursor::loaded(&checkpoint, envelope);
381            let outcome = self
382                .execute_state(
383                    state,
384                    run,
385                    Some(&mut cursor),
386                    Arc::new(NoopObserver),
387                    false,
388                    false,
389                )
390                .await
391                .map_err(AgentConversationError::Run)?;
392            self.commit_durable_outcome(store.as_ref(), run, &cursor, durable, outcome)
393                .await
394        })
395    }
396
397    /// Runs with write-ahead checkpoint persistence.
398    pub fn run_checkpointed<'a>(
399        &'a self,
400        input: impl Into<String> + Send + 'a,
401        run: &'a RunContext,
402        checkpoint: &'a AgentCheckpoint,
403    ) -> AgentFuture<'a, Result<AgentOutcome, AgentError>> {
404        let input = input.into();
405        Box::pin(async move {
406            let mut state = self.initial_state(input, checkpoint.id().to_string());
407            state.usage = run.budget().usage();
408            let mut cursor = CheckpointCursor::create(checkpoint, run, &state)?;
409            self.execute_state(
410                state,
411                run,
412                Some(&mut cursor),
413                Arc::new(NoopObserver),
414                true,
415                true,
416            )
417            .await
418        })
419    }
420
421    /// Resumes a persisted Agent execution.
422    pub fn resume<'a>(
423        &'a self,
424        checkpoint: &'a AgentCheckpoint,
425        run: &'a RunContext,
426        policy: ResumePolicy,
427    ) -> AgentFuture<'a, Result<AgentOutcome, AgentError>> {
428        Box::pin(async move {
429            let (envelope, mut state) = checkpoint.load()?;
430            self.validate_checkpoint_identity(&state)?;
431            if let Some(outcome) = state.outcome() {
432                validate_exact_usage(state.usage, run.budget().usage())?;
433                return Ok(outcome);
434            }
435            if let AgentCheckpointPhase::TurnInFlight { turn } = state.phase {
436                if policy == ResumePolicy::RejectAmbiguous {
437                    return Err(AgentError::AmbiguousCheckpoint { turn });
438                }
439                validate_usage_floor(state.usage, run.budget().usage())?;
440                state.usage = run.budget().usage();
441                state.phase = AgentCheckpointPhase::ReadyForTurn;
442            } else {
443                validate_exact_usage(state.usage, run.budget().usage())?;
444            }
445            let mut cursor = CheckpointCursor::loaded(checkpoint, envelope);
446            self.execute_state(
447                state,
448                run,
449                Some(&mut cursor),
450                Arc::new(NoopObserver),
451                false,
452                true,
453            )
454            .await
455        })
456    }
457
458    fn initial_state(&self, input: String, execution_id: String) -> AgentCheckpointState {
459        let mut transcript = self.instructions.clone();
460        transcript.push(Message::user(input));
461        self.initial_state_from_transcript(transcript, execution_id)
462    }
463
464    fn initial_state_from_transcript(
465        &self,
466        transcript: Vec<Message>,
467        execution_id: String,
468    ) -> AgentCheckpointState {
469        AgentCheckpointState {
470            execution_id,
471            agent: self.name.clone(),
472            model: self.model_ref.clone(),
473            transcript,
474            turns: 0,
475            tool_calls: 0,
476            delegations: 0,
477            usage: Usage::default(),
478            phase: AgentCheckpointPhase::ReadyForTurn,
479            durable_conversation: None,
480        }
481    }
482
483    async fn execute_state(
484        &self,
485        state: AgentCheckpointState,
486        run: &RunContext,
487        mut checkpoint: Option<&mut CheckpointCursor>,
488        observer: Arc<dyn AgentObserver>,
489        retrieve_context: bool,
490        persist_terminal_checkpoint: bool,
491    ) -> Result<AgentOutcome, AgentError> {
492        let started = run
493            .record(
494                RunEventKind::Lifecycle(LifecycleEvent::Started),
495                run.caused_by(),
496            )?
497            .map(|event| event.meta.event_id);
498        emit_agent_event(
499            observer.as_ref(),
500            AgentStreamEvent::Started {
501                agent: self.name.clone(),
502            },
503        )
504        .await;
505        let result = async {
506            let has_context = !self.context.is_empty() || !self.dynamic_context.is_empty();
507            let state = if retrieve_context && has_context {
508                let mut prepared = self
509                    .prepare_context(state, run, started, observer.as_ref())
510                    .await?;
511                prepared.usage = run.budget().usage();
512                save_checkpoint(&mut checkpoint, &prepared)?;
513                prepared
514            } else {
515                state
516            };
517            self.run_loop(
518                state,
519                run,
520                started,
521                checkpoint,
522                observer.as_ref(),
523                persist_terminal_checkpoint,
524            )
525            .await
526        }
527        .await;
528        let terminal = terminal_event(&self.name, &result);
529        run.record(terminal, started)?;
530        if let Ok(outcome) = &result {
531            emit_agent_event(
532                observer.as_ref(),
533                AgentStreamEvent::Completed {
534                    outcome: outcome.clone(),
535                },
536            )
537            .await;
538        }
539        result
540    }
541
542    async fn run_loop(
543        &self,
544        state: AgentCheckpointState,
545        run: &RunContext,
546        caused_by: Option<EventId>,
547        mut checkpoint: Option<&mut CheckpointCursor>,
548        observer: &dyn AgentObserver,
549        persist_terminal_checkpoint: bool,
550    ) -> Result<AgentOutcome, AgentError> {
551        self.validate_config()?;
552        let mut progress = AgentProgress::from(state);
553
554        loop {
555            Self::check_lifecycle(run)?;
556            if progress.turns >= self.config.max_turns {
557                return Err(AgentError::MaxTurns {
558                    max_turns: self.config.max_turns,
559                });
560            }
561            save_checkpoint(
562                &mut checkpoint,
563                &self.checkpoint_state(
564                    &progress,
565                    run,
566                    AgentCheckpointPhase::TurnInFlight {
567                        turn: progress.turns + 1,
568                    },
569                ),
570            )?;
571            consume_budget(
572                run,
573                Usage {
574                    turns: 1,
575                    ..Usage::default()
576                },
577                caused_by,
578            )?;
579            progress.turns += 1;
580            emit_agent_event(
581                observer,
582                AgentStreamEvent::TurnStarted {
583                    turn: progress.turns,
584                },
585            )
586            .await;
587            emit_usage(observer, run).await;
588            record_domain(
589                run,
590                "turn.started",
591                serde_json::json!({"agent": self.name, "turn": progress.turns}),
592                caused_by,
593            )?;
594
595            let response = self
596                .invoke_model(
597                    &progress.transcript,
598                    run,
599                    progress.turns,
600                    caused_by,
601                    observer,
602                )
603                .await?;
604
605            let calls = tool_calls_from(&response.content);
606            let assistant = Message::new(Role::Assistant, response.content.clone())
607                .map_err(|error| AgentError::Protocol(error.to_string()))?;
608            progress.transcript.push(assistant);
609
610            if calls.is_empty() {
611                if matches!(
612                    response.finish_reason,
613                    runifold_model::FinishReason::ToolCalls
614                ) {
615                    return Err(AgentError::Protocol(
616                        "model stopped for tool calls without emitting a tool call".into(),
617                    ));
618                }
619                if persist_terminal_checkpoint {
620                    save_checkpoint(
621                        &mut checkpoint,
622                        &self.checkpoint_state(
623                            &progress,
624                            run,
625                            AgentCheckpointPhase::Completed {
626                                response: Box::new(response.clone()),
627                            },
628                        ),
629                    )?;
630                }
631                return Ok(progress.outcome(response, run.budget().usage()));
632            }
633
634            self.execute_calls(calls, run, caused_by, &mut progress, observer)
635                .await?;
636            save_checkpoint(
637                &mut checkpoint,
638                &self.checkpoint_state(&progress, run, AgentCheckpointPhase::ReadyForTurn),
639            )?;
640        }
641    }
642
643    async fn invoke_model(
644        &self,
645        transcript: &[Message],
646        run: &RunContext,
647        turn: u32,
648        caused_by: Option<EventId>,
649        observer: &dyn AgentObserver,
650    ) -> Result<ModelResponse, AgentError> {
651        record_domain(
652            run,
653            "model.started",
654            serde_json::json!({
655                "agent": self.name,
656                "turn": turn,
657                "provider": self.model_ref.provider,
658                "model": self.model_ref.name,
659            }),
660            caused_by,
661        )?;
662        let response = match self
663            .stream_model_response(self.request(transcript)?, run, turn, observer)
664            .await
665        {
666            Ok(response) => response,
667            Err(error) => {
668                record_domain(
669                    run,
670                    "model.failed",
671                    serde_json::json!({
672                        "agent": self.name,
673                        "turn": turn,
674                        "kind": format!("{:?}", error.kind),
675                    }),
676                    caused_by,
677                )?;
678                return Err(error.into());
679            }
680        };
681        record_domain(
682            run,
683            "model.completed",
684            serde_json::json!({
685                "agent": self.name,
686                "turn": turn,
687                "finish_reason": response.finish_reason,
688                "usage": response.usage,
689            }),
690            caused_by,
691        )?;
692        consume_budget(run, response.usage.into(), caused_by)?;
693        emit_usage(observer, run).await;
694        Ok(response)
695    }
696
697    async fn stream_model_response(
698        &self,
699        request: ModelRequest,
700        run: &RunContext,
701        turn: u32,
702        observer: &dyn AgentObserver,
703    ) -> Result<ModelResponse, ModelError> {
704        let context = ModelCallContext::for_run(run);
705        let cancellation = context.cancellation().clone();
706        let opening = self.model.stream(request, context);
707        let mut stream = match select(Box::pin(cancellation.cancelled()), Box::pin(opening)).await {
708            Either::Left(_) => return Err(cancelled_model_error()),
709            Either::Right((result, _)) => result?,
710        };
711        let mut accumulator = ModelStreamAccumulator::new();
712        loop {
713            let next = stream.next();
714            let event = match select(Box::pin(cancellation.cancelled()), Box::pin(next)).await {
715                Either::Left(_) => return Err(cancelled_model_error()),
716                Either::Right((Some(event), _)) => event?,
717                Either::Right((None, _)) => {
718                    return Err(ModelError::local(
719                        ModelErrorKind::Protocol,
720                        "model stream ended before a terminal response event",
721                    ));
722                }
723            };
724            let response = accumulator.push(event.clone())?;
725            emit_agent_event(observer, AgentStreamEvent::Model { turn, event }).await;
726            if let Some(response) = response {
727                return Ok(response);
728            }
729        }
730    }
731
732    fn validate_config(&self) -> Result<(), AgentError> {
733        if self.name.trim().is_empty() {
734            return Err(AgentError::InvalidConfig(
735                "agent name cannot be empty".into(),
736            ));
737        }
738        if self.config.max_turns == 0 {
739            return Err(AgentError::InvalidConfig(
740                "max_turns must be greater than zero".into(),
741            ));
742        }
743        if let Some(collision) = self
744            .agents
745            .model_specs()
746            .into_iter()
747            .find(|spec| self.tools.contains(&spec.name))
748        {
749            return Err(AgentError::InvalidConfig(format!(
750                "callable name `{}` is registered as both a tool and an agent",
751                collision.name
752            )));
753        }
754        Ok(())
755    }
756
757    fn validate_checkpoint_identity(&self, state: &AgentCheckpointState) -> Result<(), AgentError> {
758        if state.agent != self.name || state.model != self.model_ref {
759            return Err(runifold_core::CheckpointError::new(
760                runifold_core::CheckpointErrorKind::InvalidPayload,
761                "checkpoint Agent or model identity does not match",
762            )
763            .into());
764        }
765        Ok(())
766    }
767
768    fn checkpoint_state(
769        &self,
770        progress: &AgentProgress,
771        run: &RunContext,
772        phase: AgentCheckpointPhase,
773    ) -> AgentCheckpointState {
774        AgentCheckpointState {
775            execution_id: progress.execution_id.clone(),
776            agent: self.name.clone(),
777            model: self.model_ref.clone(),
778            transcript: progress.transcript.clone(),
779            turns: progress.turns,
780            tool_calls: progress.tool_calls,
781            delegations: progress.delegations,
782            usage: run.budget().usage(),
783            phase,
784            durable_conversation: progress.durable_conversation.clone(),
785        }
786    }
787
788    async fn commit_durable_outcome(
789        &self,
790        store: &dyn DurableConversationStore,
791        run: &RunContext,
792        cursor: &CheckpointCursor,
793        durable: DurableConversationCheckpoint,
794        outcome: AgentOutcome,
795    ) -> Result<AgentConversationOutcome, AgentConversationError> {
796        let persisted_prefix_len = usize::try_from(durable.persisted_prefix_len).map_err(|_| {
797            AgentConversationError::Run(checkpoint_payload_error(
798                "durable conversation prefix does not fit this platform",
799            ))
800        })?;
801        if persisted_prefix_len >= outcome.transcript.len() {
802            return Err(AgentConversationError::Run(checkpoint_payload_error(
803                "durable conversation checkpoint has an invalid transcript prefix",
804            )));
805        }
806        let messages = outcome
807            .transcript
808            .iter()
809            .skip(persisted_prefix_len)
810            .filter(|message| !is_transient_context(message))
811            .cloned()
812            .collect();
813        let state = AgentCheckpointState {
814            execution_id: cursor.id().to_string(),
815            agent: self.name.clone(),
816            model: self.model_ref.clone(),
817            transcript: outcome.transcript.clone(),
818            turns: outcome.turns,
819            tool_calls: outcome.tool_calls,
820            delegations: outcome.delegations,
821            usage: run.budget().usage(),
822            phase: AgentCheckpointPhase::Completed {
823                response: Box::new(outcome.response.clone()),
824            },
825            durable_conversation: Some(durable.clone()),
826        };
827        let checkpoint = cursor.next(&state).map_err(AgentConversationError::Run)?;
828        let command = DurableConversationCommit {
829            namespace: durable.namespace,
830            append: ConversationAppend {
831                conversation_id: durable.conversation_id,
832                expected_version: durable.expected_version,
833                messages,
834            },
835            checkpoint,
836            expected_checkpoint_revision: cursor.revision(),
837        };
838        match store.commit_durable_turn(command).await {
839            Ok(conversation_version) => Ok(AgentConversationOutcome {
840                outcome,
841                conversation_version,
842            }),
843            Err(source) => Err(AgentConversationError::Commit {
844                source,
845                outcome: Box::new(outcome),
846            }),
847        }
848    }
849
850    pub(super) fn check_lifecycle(run: &RunContext) -> Result<(), AgentError> {
851        let error = if run.cancellation().is_cancelled() {
852            Some((
853                runifold_model::ModelErrorKind::Cancelled,
854                "agent run was cancelled",
855            ))
856        } else if run
857            .deadline()
858            .is_some_and(|deadline| deadline <= Instant::now())
859        {
860            Some((
861                runifold_model::ModelErrorKind::DeadlineExceeded,
862                "agent run deadline elapsed",
863            ))
864        } else {
865            None
866        };
867        if let Some((kind, message)) = error {
868            return Err(runifold_model::ModelError::local(kind, message).into());
869        }
870        Ok(())
871    }
872
873    fn request(&self, transcript: &[Message]) -> Result<ModelRequest, AgentError> {
874        let (first, rest) = transcript
875            .split_first()
876            .ok_or_else(|| AgentError::Protocol("agent transcript is empty".into()))?;
877        let mut request = ModelRequest::new(self.model_ref.clone(), first.clone());
878        request.messages.extend_from_slice(rest);
879        request.tools = self.tools.model_specs();
880        request.tools.extend(self.agents.model_specs());
881        for tool in &self.provider_tools {
882            request = request.provider_tool(tool.clone());
883        }
884        request.generation.clone_from(&self.generation);
885        request = request.response_mode(self.response_mode);
886        request.provider_options.clone_from(&self.provider_options);
887        request.feature_policy = self.config.feature_policy;
888        request.output_format.clone_from(&self.output_format);
889        Ok(request)
890    }
891}
892
893fn checkpoint_payload_error(message: &str) -> AgentError {
894    runifold_core::CheckpointError::new(runifold_core::CheckpointErrorKind::InvalidPayload, message)
895        .into()
896}
897
898fn cancelled_model_error() -> ModelError {
899    ModelError::local(ModelErrorKind::Cancelled, "model invocation was cancelled")
900}
901
902fn tool_calls_from(content: &[ContentPart]) -> Vec<ToolCall> {
903    content
904        .iter()
905        .filter_map(|part| match part {
906            ContentPart::ToolCall(call) => Some(call.clone()),
907            _ => None,
908        })
909        .collect()
910}
runifold_agent/agent/execution.rs

runifold_agent/agent/
execution.rs
