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

runifold_agent/agent/
execution.rs
