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            record_domain(
597                run,
598                "turn.started",
599                serde_json::json!({"agent": self.name, "turn": progress.turns}),
600                caused_by,
601            )?;
602
603            let response = self
604                .invoke_model(
605                    &progress.transcript,
606                    run,
607                    progress.turns,
608                    tool_choice,
609                    caused_by,
610                    observer,
611                )
612                .await?;
613
614            let calls = tool_calls_from(&response.content);
615            if calls.is_empty() {
616                if matches!(
617                    response.finish_reason,
618                    runifold_model::FinishReason::ToolCalls
619                ) && !response.content.is_empty()
620                {
621                    return Err(AgentError::Protocol(
622                        "model stopped for tool calls without emitting a tool call".into(),
623                    ));
624                }
625                if requires_tool {
626                    return Err(AgentError::ToolRequirementUnsatisfied {
627                        required: self.min_successful_tool_calls,
628                        successful: self.successful_local_tool_calls(&progress)?,
629                    });
630                }
631                if let Some(outcome) = self
632                    .complete_terminal_candidate(
633                        response,
634                        run,
635                        &mut progress,
636                        &mut checkpoint,
637                        TerminalCompletionContext {
638                            caused_by,
639                            observer,
640                            persist_terminal_checkpoint,
641                        },
642                    )
643                    .await?
644                {
645                    return Ok(outcome);
646                }
647                continue;
648            }
649
650            let assistant = Message::new(Role::Assistant, response.content.clone())
651                .map_err(|error| AgentError::Protocol(error.to_string()))?;
652            progress.transcript.push(assistant);
653
654            self.execute_calls(calls, run, caused_by, &mut progress, observer)
655                .await?;
656            save_checkpoint(
657                &mut checkpoint,
658                &self.checkpoint_state(&progress, run, AgentCheckpointPhase::ReadyForTurn),
659            )?;
660        }
661    }
662
663    async fn invoke_model(
664        &self,
665        transcript: &[Message],
666        run: &RunContext,
667        turn: u32,
668        tool_choice: ToolChoice,
669        caused_by: Option<EventId>,
670        observer: &dyn AgentObserver,
671    ) -> Result<ModelResponse, AgentError> {
672        record_domain(
673            run,
674            "model.started",
675            serde_json::json!({
676                "agent": self.name,
677                "turn": turn,
678                "provider": self.model_ref.provider,
679                "model": self.model_ref.name,
680            }),
681            caused_by,
682        )?;
683        let response = match self
684            .stream_model_response(self.request(transcript, tool_choice)?, run, turn, observer)
685            .await
686        {
687            Ok(response) => response,
688            Err(error) => {
689                record_domain(
690                    run,
691                    "model.failed",
692                    serde_json::json!({
693                        "agent": self.name,
694                        "turn": turn,
695                        "kind": format!("{:?}", error.kind),
696                    }),
697                    caused_by,
698                )?;
699                return Err(error.into());
700            }
701        };
702        record_domain(
703            run,
704            "model.completed",
705            serde_json::json!({
706                "agent": self.name,
707                "turn": turn,
708                "finish_reason": response.finish_reason,
709                "usage": response.usage,
710            }),
711            caused_by,
712        )?;
713        consume_budget(run, response.usage.into(), caused_by)?;
714        emit_usage(observer, run).await;
715        Ok(response)
716    }
717
718    async fn stream_model_response(
719        &self,
720        request: ModelRequest,
721        run: &RunContext,
722        turn: u32,
723        observer: &dyn AgentObserver,
724    ) -> Result<ModelResponse, ModelError> {
725        let context = ModelCallContext::for_run(run);
726        let cancellation = context.cancellation().clone();
727        let opening = self.model.stream(request, context);
728        let mut stream = match select(Box::pin(cancellation.cancelled()), Box::pin(opening)).await {
729            Either::Left(_) => return Err(cancelled_model_error()),
730            Either::Right((result, _)) => result?,
731        };
732        let mut accumulator = ModelStreamAccumulator::new();
733        loop {
734            let next = stream.next();
735            let event = match select(Box::pin(cancellation.cancelled()), Box::pin(next)).await {
736                Either::Left(_) => return Err(cancelled_model_error()),
737                Either::Right((Some(event), _)) => event?,
738                Either::Right((None, _)) => {
739                    return Err(ModelError::local(
740                        ModelErrorKind::Protocol,
741                        "model stream ended before a terminal response event",
742                    ));
743                }
744            };
745            let response = accumulator.push(event.clone())?;
746            emit_agent_event(observer, AgentStreamEvent::Model { turn, event }).await;
747            if let Some(response) = response {
748                return Ok(response);
749            }
750        }
751    }
752
753    fn validate_config(&self) -> Result<(), AgentError> {
754        if self.name.trim().is_empty() {
755            return Err(AgentError::InvalidConfig(
756                "agent name cannot be empty".into(),
757            ));
758        }
759        if self.config.max_turns == 0 {
760            return Err(AgentError::InvalidConfig(
761                "max_turns must be greater than zero".into(),
762            ));
763        }
764        if self.min_successful_tool_calls > 0 && self.tools.is_empty() {
765            return Err(AgentError::InvalidConfig(format!(
766                "min_successful_tool_calls={} requires at least one registered local Tool",
767                self.min_successful_tool_calls
768            )));
769        }
770        if let Some(collision) = self
771            .agents
772            .model_specs()
773            .into_iter()
774            .find(|spec| self.tools.contains(&spec.name))
775        {
776            return Err(AgentError::InvalidConfig(format!(
777                "callable name `{}` is registered as both a tool and an agent",
778                collision.name
779            )));
780        }
781        Ok(())
782    }
783
784    fn validate_checkpoint_identity(&self, state: &AgentCheckpointState) -> Result<(), AgentError> {
785        if state.agent != self.name || state.model != self.model_ref {
786            return Err(runifold_core::CheckpointError::new(
787                runifold_core::CheckpointErrorKind::InvalidPayload,
788                "checkpoint Agent or model identity does not match",
789            )
790            .into());
791        }
792        Ok(())
793    }
794
795    pub(super) fn checkpoint_state(
796        &self,
797        progress: &AgentProgress,
798        run: &RunContext,
799        phase: AgentCheckpointPhase,
800    ) -> AgentCheckpointState {
801        AgentCheckpointState {
802            execution_id: progress.execution_id.clone(),
803            agent: self.name.clone(),
804            model: self.model_ref.clone(),
805            transcript: progress.transcript.clone(),
806            turns: progress.turns,
807            tool_calls: progress.tool_calls,
808            delegations: progress.delegations,
809            usage: run.budget().usage(),
810            phase,
811            durable_conversation: progress.durable_conversation.clone(),
812        }
813    }
814
815    async fn commit_durable_outcome(
816        &self,
817        store: &dyn DurableConversationStore,
818        run: &RunContext,
819        cursor: &CheckpointCursor,
820        durable: DurableConversationCheckpoint,
821        outcome: AgentOutcome,
822    ) -> Result<AgentConversationOutcome, AgentConversationError> {
823        let persisted_prefix_len = usize::try_from(durable.persisted_prefix_len).map_err(|_| {
824            AgentConversationError::Run(checkpoint_payload_error(
825                "durable conversation prefix does not fit this platform",
826            ))
827        })?;
828        if persisted_prefix_len >= outcome.transcript.len() {
829            return Err(AgentConversationError::Run(checkpoint_payload_error(
830                "durable conversation checkpoint has an invalid transcript prefix",
831            )));
832        }
833        let messages = outcome
834            .transcript
835            .iter()
836            .skip(persisted_prefix_len)
837            .filter(|message| !is_transient_context(message))
838            .cloned()
839            .collect();
840        let state = AgentCheckpointState {
841            execution_id: cursor.id().to_string(),
842            agent: self.name.clone(),
843            model: self.model_ref.clone(),
844            transcript: outcome.transcript.clone(),
845            turns: outcome.turns,
846            tool_calls: outcome.tool_calls,
847            delegations: outcome.delegations,
848            usage: run.budget().usage(),
849            phase: AgentCheckpointPhase::Completed {
850                response: Box::new(outcome.response.clone()),
851            },
852            durable_conversation: Some(durable.clone()),
853        };
854        let checkpoint = cursor.next(&state).map_err(AgentConversationError::Run)?;
855        let command = DurableConversationCommit {
856            namespace: durable.namespace,
857            append: ConversationAppend {
858                conversation_id: durable.conversation_id,
859                expected_version: durable.expected_version,
860                messages,
861            },
862            checkpoint,
863            expected_checkpoint_revision: cursor.revision(),
864        };
865        match store.commit_durable_turn(command).await {
866            Ok(conversation_version) => Ok(AgentConversationOutcome {
867                outcome,
868                conversation_version,
869            }),
870            Err(source) => Err(AgentConversationError::Commit {
871                source,
872                outcome: Box::new(outcome),
873            }),
874        }
875    }
876
877    pub(super) fn check_lifecycle(run: &RunContext) -> Result<(), AgentError> {
878        let error = if run.cancellation().is_cancelled() {
879            Some((
880                runifold_model::ModelErrorKind::Cancelled,
881                "agent run was cancelled",
882            ))
883        } else if run
884            .deadline()
885            .is_some_and(|deadline| deadline <= Instant::now())
886        {
887            Some((
888                runifold_model::ModelErrorKind::DeadlineExceeded,
889                "agent run deadline elapsed",
890            ))
891        } else {
892            None
893        };
894        if let Some((kind, message)) = error {
895            return Err(runifold_model::ModelError::local(kind, message).into());
896        }
897        Ok(())
898    }
899
900    fn request(
901        &self,
902        transcript: &[Message],
903        tool_choice: ToolChoice,
904    ) -> Result<ModelRequest, AgentError> {
905        let (first, rest) = transcript
906            .split_first()
907            .ok_or_else(|| AgentError::Protocol("agent transcript is empty".into()))?;
908        let mut request = ModelRequest::new(self.model_ref.clone(), first.clone());
909        request.messages.extend_from_slice(rest);
910        request.tools = self.tools.model_specs();
911        request.tools.extend(self.agents.model_specs());
912        request.tool_choice = tool_choice;
913        for tool in &self.provider_tools {
914            request = request.provider_tool(tool.clone());
915        }
916        request.generation.clone_from(&self.generation);
917        request = request.response_mode(self.response_mode);
918        request.provider_options.clone_from(&self.provider_options);
919        request.feature_policy = self.config.feature_policy;
920        request.output_format.clone_from(&self.output_format);
921        Ok(request)
922    }
923
924    fn successful_local_tool_calls(&self, progress: &AgentProgress) -> Result<u32, AgentError> {
925        let count = progress
926            .transcript
927            .iter()
928            .filter(|message| {
929                message
930                    .metadata
931                    .get(TOOL_RESULT_EXECUTION_ID_METADATA)
932                    .and_then(serde_json::Value::as_str)
933                    == Some(progress.execution_id.as_str())
934            })
935            .flat_map(|message| &message.content)
936            .filter(|part| {
937                matches!(
938                    part,
939                    ContentPart::ToolResult(result)
940                        if !result.is_error
941                            && result
942                                .name
943                                .as_deref()
944                                .is_some_and(|name| self.tools.contains(name))
945                )
946            })
947            .count();
948        u32::try_from(count)
949            .map_err(|_| AgentError::Protocol("successful Tool-call counter overflow".into()))
950    }
951
952    fn next_tool_choice(
953        &self,
954        progress: &AgentProgress,
955        run: &RunContext,
956    ) -> Result<ToolChoice, AgentError> {
957        let successful = self.successful_local_tool_calls(progress)?;
958        let remaining_required = self.min_successful_tool_calls.saturating_sub(successful);
959        Self::validate_tool_requirement_budget(remaining_required, run)?;
960        if progress.turns >= self.config.max_turns {
961            if remaining_required > 0 {
962                return Err(AgentError::ToolRequirementUnsatisfied {
963                    required: self.min_successful_tool_calls,
964                    successful,
965                });
966            }
967            return Err(AgentError::MaxTurns {
968                max_turns: self.config.max_turns,
969            });
970        }
971        Ok(if remaining_required > 0 {
972            ToolChoice::Required
973        } else {
974            ToolChoice::Auto
975        })
976    }
977
978    fn validate_tool_requirement_budget(
979        remaining_required: u32,
980        run: &RunContext,
981    ) -> Result<(), AgentError> {
982        let Some(limit) = run.budget().limit().tool_calls else {
983            return Ok(());
984        };
985        let remaining = limit.saturating_sub(run.budget().usage().tool_calls);
986        if u64::from(remaining_required) > remaining {
987            return Err(AgentError::ToolRequirementExceedsBudget {
988                required: remaining_required,
989                remaining,
990            });
991        }
992        Ok(())
993    }
994}
995
996fn checkpoint_payload_error(message: &str) -> AgentError {
997    runifold_core::CheckpointError::new(runifold_core::CheckpointErrorKind::InvalidPayload, message)
998        .into()
999}
1000
1001fn cancelled_model_error() -> ModelError {
1002    ModelError::local(ModelErrorKind::Cancelled, "model invocation was cancelled")
1003}
1004
1005fn tool_calls_from(content: &[ContentPart]) -> Vec<ToolCall> {
1006    content
1007        .iter()
1008        .filter_map(|part| match part {
1009            ContentPart::ToolCall(call) => Some(call.clone()),
1010            _ => None,
1011        })
1012        .collect()
1013}
runifold_agent/agent/execution.rs

runifold_agent/agent/
execution.rs
