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

runifold_agent/agent/
execution.rs
