1use serde_json::Value;
2use std::collections::BTreeMap;
3
4use crate::llm::{LlmClient, LlmError};
5use crate::memory::CompactionExhaustedError;
6use crate::tools::ToolSpecKind;
7use crate::types::{AgentResult, AgentTask, CompletionReason, ToolDirective, ToolExecutionResult};
8
9use crate::runtime::cancellation::CancellationToken;
10use crate::runtime::cycle_runner::{is_prompt_too_long_error, MAX_PROMPT_TOO_LONG_RETRIES};
11use crate::runtime::model_calls::ModelCallDispatchRequest;
12use crate::runtime::results::assistant_message_from_response;
13use crate::runtime::tool_call_runner::{
14 apply_tool_use_behavior, needs_tool_call_id, skipped_tool_result,
15};
16
17use super::approval::{
18 approval_error_result, approval_provider_result, PendingToolApprovalCapture,
19};
20use super::budget::{
21 budget_snapshot, enforce_cycle_start, finalize_run_budget, lock_budget,
22 observe_tool_batch_completion, preflight_tool_batch, project_model_call_completion,
23 PreparedRunBudget,
24};
25use super::checkpoint::{CheckpointModelCompletion, CheckpointToolPlan};
26use super::controls::{CheckpointRuntimeControl, RuntimeRunControls};
27use super::helpers::{
28 cancelled_agent_result, collect_interruption_messages, controls_cancelled,
29 drain_steering_queue, failed_agent_result, finalize_terminal_projection,
30 image_notification_from_tool_result, project_cycle_cancellation, task_token_usage,
31};
32use super::lifecycle::{
33 finalize_no_tool_cycle, finalize_tool_cycle, NoToolCycleFinalization, ToolCycleFinalization,
34};
35use super::memory::{
36 compact_cycle_memory, memory_compact_completed_event, memory_compact_event_payload,
37 memory_compact_started_event, memory_inference_failure_result, notify_memory_after_compact,
38 notify_memory_before_compact, CycleMemoryCompaction, MemoryCompactCompletion,
39};
40use super::model_request::{
41 build_model_request, cycle_stream_callback, effective_model_call_target,
42};
43use super::planning::block_on_engine_tool_run;
44use super::run_setup::{
45 prepare_approval_broker, prepare_run_setup, prepare_runtime_accounting, PreparedRun,
46 PreparedRuntimeAccounting,
47};
48use super::state::AgentRuntime;
49use super::tool_batch::{PreparedToolBatch, ToolBatchSetup};
50
51impl<C: LlmClient + Clone + 'static> AgentRuntime<C> {
52 pub fn run_with_controls(
53 &self,
54 mut task: AgentTask,
55 mut controls: RuntimeRunControls,
56 ) -> Result<AgentResult, LlmError> {
57 if let Some(policy) = self.tool_policy.as_ref() {
58 crate::runtime::tool_planner::project_tool_policy(&mut task, policy);
59 }
60 prepare_approval_broker(&mut controls);
61 let PreparedRun {
62 task,
63 messages,
64 cycles,
65 shared_state,
66 workspace_path,
67 workspace_backend,
68 sub_task_manager,
69 memory_manager,
70 memory_model_provider,
71 } = prepare_run_setup(self, task, &controls)?;
72 let cycle_index_start = controls.cycle_index_start.unwrap_or(1);
73 let backend_manages_checkpoint_cycles = self.execution_backend.manages_checkpoint_cycles();
74 self.emit_run_started(&controls, &task, &workspace_path);
75 let PreparedRunBudget {
76 limits: effective_budget_limits,
77 controller: budget_controller,
78 early_result,
79 } = self.prepare_run_budget(&controls, &messages, &cycles, &shared_state);
80 let configured_budget = effective_budget_limits.is_some();
81 let child_budget_limits = effective_budget_limits.clone();
82
83 let effective_cancellation_token = controls.effective_cancellation_token();
84 let PreparedRuntimeAccounting {
85 model_call_ledger,
86 model_call_coordinator,
87 checkpoint,
88 mut memory_manager,
89 } = prepare_runtime_accounting(
90 self,
91 &task,
92 &mut controls,
93 memory_manager,
94 memory_model_provider,
95 &budget_controller,
96 )?;
97 if !backend_manages_checkpoint_cycles {
98 if let Some(result) = checkpoint.begin_run_cycle(cycle_index_start)? {
99 return Ok(result);
100 }
101 }
102 if let Some(mut result) = early_result {
103 result.token_usage = model_call_ledger.usage();
104 return Ok(result);
105 }
106 self.emit_log(
107 &controls,
108 "agent_started",
109 BTreeMap::from([("model".to_string(), Value::String(task.model.clone()))]),
110 );
111 if !configured_budget && controls_cancelled(&controls) {
112 self.emit_log(
113 &controls,
114 "run_cancelled",
115 BTreeMap::from([(
116 "error".to_string(),
117 Value::String("Operation was cancelled".to_string()),
118 )]),
119 );
120 return Ok(cancelled_agent_result(
121 messages,
122 cycles,
123 shared_state,
124 task_token_usage(&controls),
125 ));
126 }
127 let effective_event_handler = controls.effective_event_handler();
128 let mut pending_error = None;
129 let cycle_count = controls.cycle_count.unwrap_or(task.max_cycles);
130 let mut result = self.execution_backend.execute_with_state(
131 &task,
132 messages,
133 cycles,
134 shared_state,
135 |cycle_index, messages, cycles, shared_state, cancellation_token| {
136 let _cancellation_scope = CancellationToken::enter_scope(cancellation_token);
137 if !backend_manages_checkpoint_cycles {
138 if let Some(result) =
139 checkpoint.begin_cycle(cycle_index, messages, cycles, shared_state)
140 {
141 return Some(result);
142 }
143 }
144 if let Some(result) = project_cycle_cancellation(
145 self,
146 &controls,
147 cycle_index,
148 cancellation_token,
149 messages,
150 cycles,
151 shared_state,
152 ) {
153 return Some(result);
154 }
155 let active_after_cycle_denials = match self.read_after_cycle_denials(
156 &controls,
157 cycle_index,
158 messages,
159 cycles,
160 shared_state,
161 ) {
162 Ok(denials) => denials,
163 Err(result) => return Some(*result),
164 };
165 self.apply_cycle_inputs(&controls, cycle_index, messages, shared_state);
166 if let Some(result) = project_cycle_cancellation(
167 self,
168 &controls,
169 cycle_index,
170 cancellation_token,
171 messages,
172 cycles,
173 shared_state,
174 ) {
175 return Some(result);
176 }
177 if let Some(result) = enforce_cycle_start(
178 &budget_controller,
179 &controls,
180 cycle_index,
181 messages,
182 cycles,
183 shared_state,
184 ) {
185 return Some(result);
186 }
187 if let Some(result) = checkpoint.update_budget_usage(
188 || budget_snapshot(&budget_controller),
189 messages,
190 cycles,
191 shared_state,
192 ) {
193 return Some(result);
194 }
195 self.emit_log(
196 &controls,
197 "cycle_started",
198 BTreeMap::from([
199 ("cycle".to_string(), Value::from(cycle_index)),
200 ("max_cycles".to_string(), Value::from(task.max_cycles)),
201 ("message_count".to_string(), Value::from(messages.len())),
202 ]),
203 );
204 let hook_manager = self.hook_manager();
205 let CycleMemoryCompaction {
206 messages: mut compacted_messages,
207 changed: memory_compacted,
208 recent_tool_call_ids,
209 } = match compact_cycle_memory(
210 self,
211 &controls,
212 &task,
213 &hook_manager,
214 &mut memory_manager,
215 cycle_index,
216 messages,
217 cycles,
218 shared_state,
219 &model_call_ledger,
220 ) {
221 Ok(outcome) => outcome,
222 Err(error) => {
223 return Some(memory_inference_failure_result(
224 error,
225 &checkpoint,
226 &budget_controller,
227 &controls,
228 messages,
229 cycles,
230 shared_state,
231 ));
232 }
233 };
234 *messages = compacted_messages.clone();
235 let tool_schemas = self.planned_tool_schemas_with_after_cycle_denials(
236 &task,
237 &active_after_cycle_denials,
238 );
239 let llm_messages = compacted_messages.clone();
240 let (request_messages, request_tool_schemas) = hook_manager.apply_before_llm(
241 &task,
242 cycle_index,
243 llm_messages,
244 tool_schemas,
245 shared_state,
246 );
247 if let Err(error) = memory_manager
248 .validate_model_recovery_surface(&request_messages, &request_tool_schemas)
249 {
250 return Some(failed_agent_result(
251 messages.clone(),
252 cycles.clone(),
253 shared_state.clone(),
254 error,
255 task_token_usage(&controls),
256 ));
257 }
258 let mut request_messages = request_messages;
259 let mut request_tool_schemas = request_tool_schemas;
260 let mut memory_compacted = memory_compacted;
261 let mut prompt_too_long_retries = 0;
262 let mut stream_metadata = task.metadata.clone();
263 if let Some(execution_context) = controls.execution_context.as_ref() {
264 stream_metadata.extend(execution_context.metadata.clone());
265 }
266 let cycle_stream_callback = cycle_stream_callback(
267 effective_event_handler.as_ref(),
268 &stream_metadata,
269 cycle_index,
270 );
271 let response = loop {
272 let request = build_model_request(
273 &task,
274 &controls,
275 &request_messages,
276 &request_tool_schemas,
277 );
278 let (effective_backend, effective_model) = effective_model_call_target(
279 &task,
280 &controls,
281 self.default_backend.as_deref(),
282 );
283 let operation_slot = if prompt_too_long_retries == 0 {
284 "main".to_string()
285 } else {
286 format!("prompt_too_long_{prompt_too_long_retries}")
287 };
288 let completion = checkpoint.complete_model(
289 ModelCallDispatchRequest {
290 cycle_index,
291 operation_slot: &operation_slot,
292 operation: crate::types::ModelCallOperation::AgentCycle,
293 backend: &effective_backend,
294 model: &effective_model,
295 request: &request,
296 accounting: &model_call_coordinator,
297 },
298 || budget_snapshot(&budget_controller),
299 |request| {
300 self.llm_client
301 .complete_with_stream(request, cycle_stream_callback.clone())
302 },
303 (messages, cycles, shared_state),
304 );
305 let completion = match completion {
306 CheckpointModelCompletion::Continue(completion) => *completion,
307 CheckpointModelCompletion::Stop(result) => return Some(*result),
308 };
309 match completion {
310 Ok(dispatch) => break dispatch,
311 Err(error) if is_prompt_too_long_error(&error) => {
312 prompt_too_long_retries += 1;
313 if prompt_too_long_retries > MAX_PROMPT_TOO_LONG_RETRIES {
314 let error =
315 LlmError::CompactionExhausted(CompactionExhaustedError::new(
316 prompt_too_long_retries,
317 Some(error.to_string()),
318 ));
319 let message = error.to_string();
320 pending_error = Some(error);
321 return Some(failed_agent_result(
322 messages.clone(),
323 cycles.clone(),
324 shared_state.clone(),
325 message,
326 task_token_usage(&controls),
327 ));
328 }
329 memory_compacted = true;
330 let before_retry_compact = compacted_messages.clone();
331 let started = memory_compact_started_event(
332 controls.execution_context.as_ref(),
333 &memory_manager,
334 &task,
335 cycle_index,
336 &before_retry_compact,
337 None,
338 recent_tool_call_ids.as_ref(),
339 true,
340 0,
341 0,
342 )
343 .expect("forced memory compaction always starts a lifecycle");
344 let started = notify_memory_before_compact(
345 controls.execution_context.as_ref(),
346 started,
347 &before_retry_compact,
348 );
349 self.emit_log(
350 &controls,
351 "memory_compact_started",
352 memory_compact_event_payload(&started),
353 );
354 let compaction_mode;
355 compacted_messages = if prompt_too_long_retries == 1 {
356 let outcome = match memory_manager
357 .compact_for_cycle_with_usage_observed(
358 &compacted_messages,
359 cycle_index,
360 true,
361 None,
362 recent_tool_call_ids.as_ref(),
363 None,
364 ) {
365 Ok(outcome) => outcome,
366 Err(error) => {
367 return Some(memory_inference_failure_result(
368 error,
369 &checkpoint,
370 &budget_controller,
371 &controls,
372 messages,
373 cycles,
374 shared_state,
375 ));
376 }
377 };
378 compaction_mode = outcome.mode;
379 outcome.messages
380 } else {
381 let emergency = memory_manager.emergency_compact(
382 &compacted_messages,
383 (0.2 * prompt_too_long_retries as f64).min(0.95),
384 );
385 compaction_mode = if emergency == compacted_messages {
386 crate::events::MemoryCompactMode::None
387 } else {
388 crate::events::MemoryCompactMode::Emergency
389 };
390 emergency
391 };
392 let completed = memory_compact_completed_event(
393 &started,
394 MemoryCompactCompletion::new(
395 cycle_index,
396 &before_retry_compact,
397 &compacted_messages,
398 &memory_manager.config.model,
399 compaction_mode,
400 ),
401 );
402 let completed = notify_memory_after_compact(
403 controls.execution_context.as_ref(),
404 completed,
405 );
406 self.emit_log(
407 &controls,
408 "memory_compact_completed",
409 memory_compact_event_payload(&completed),
410 );
411 let retry_tool_schemas = self
412 .planned_tool_schemas_with_after_cycle_denials(
413 &task,
414 &active_after_cycle_denials,
415 );
416 let llm_messages = compacted_messages.clone();
417 (request_messages, request_tool_schemas) = hook_manager
418 .apply_before_llm(
419 &task,
420 cycle_index,
421 llm_messages,
422 retry_tool_schemas,
423 shared_state,
424 );
425 if let Err(error) = memory_manager.validate_model_recovery_surface(
426 &request_messages,
427 &request_tool_schemas,
428 ) {
429 return Some(failed_agent_result(
430 messages.clone(),
431 cycles.clone(),
432 shared_state.clone(),
433 error,
434 task_token_usage(&controls),
435 ));
436 }
437 }
438 Err(error) => {
439 let message = error.to_string();
440 return Some(failed_agent_result(
441 messages.clone(),
442 cycles.clone(),
443 shared_state.clone(),
444 message,
445 task_token_usage(&controls),
446 ));
447 }
448 }
449 };
450 let model_budget_exhaustion = response.budget_exhaustion.clone();
451 let model_usage = response.usage.clone();
452 let response = hook_manager.apply_after_llm(
453 &task,
454 cycle_index,
455 &request_messages,
456 &request_tool_schemas,
457 response.response,
458 shared_state,
459 );
460 *messages = request_messages;
461 messages.push(assistant_message_from_response(&response));
462 let mut cycle = crate::types::CycleRecord::from_response(
463 cycle_index,
464 &response,
465 Vec::<ToolExecutionResult>::new(),
466 );
467 cycle.memory_compacted = memory_compacted;
468
469 let model_boundary_result = project_model_call_completion(
470 &budget_controller,
471 &controls,
472 model_budget_exhaustion,
473 cancellation_token,
474 &cycle,
475 messages,
476 cycles,
477 shared_state,
478 );
479 if let Some(result) = checkpoint.update_budget_usage(
480 || budget_snapshot(&budget_controller),
481 messages,
482 cycles,
483 shared_state,
484 ) {
485 return Some(result);
486 }
487 if let Some(result) = model_boundary_result {
488 return Some(result);
489 }
490 self.emit_cycle_llm_response(&controls, &cycle, &model_usage);
491
492 if response.tool_calls.is_empty() {
493 return finalize_no_tool_cycle(NoToolCycleFinalization {
494 runtime: self,
495 controls: &controls,
496 task: &task,
497 cycle_index,
498 response: &response,
499 cycle,
500 messages,
501 cycles,
502 shared_state,
503 checkpoint: &checkpoint,
504 budget_controller: &budget_controller,
505 persisted_denials: &active_after_cycle_denials,
506 });
507 }
508
509 if let Some(result) = preflight_tool_batch(
510 &budget_controller,
511 &controls,
512 cycle_index,
513 &response.tool_calls,
514 &cycle,
515 messages,
516 cycles,
517 shared_state,
518 ) {
519 return Some(result);
520 }
521
522 let PreparedToolBatch {
523 mut context,
524 orchestrator: tool_orchestrator,
525 options: tool_run_options,
526 } = self.prepare_tool_batch(ToolBatchSetup {
527 task: &task,
528 controls: &controls,
529 workspace_path: &workspace_path,
530 workspace_backend: &workspace_backend,
531 shared_state,
532 sub_task_manager: &sub_task_manager,
533 cycle_index,
534 cancellation_token,
535 child_budget_limits: &child_budget_limits,
536 request_tool_schemas: &request_tool_schemas,
537 after_cycle_disallowed_tools: &active_after_cycle_denials,
538 });
539
540 let mut directive_result = None;
541 let mut directive_completion_reason = None;
542 let mut directive_completion_tool_name = None;
543 let mut image_notifications = Vec::new();
544 for (call_index, call) in response.tool_calls.iter().enumerate() {
545 if cancellation_token.is_some_and(CancellationToken::is_cancelled)
546 || controls_cancelled(&controls)
547 {
548 *shared_state = context.shared_state.clone();
549 if let Some(controller) = &budget_controller {
550 lock_budget(controller).tool_batch_complete(
551 &controls,
552 cycle_index,
553 false,
554 true,
555 );
556 }
557 cycles.push(cycle);
558 self.emit_log(
559 &controls,
560 "run_cancelled",
561 BTreeMap::from([
562 ("cycle".to_string(), Value::from(cycle_index)),
563 (
564 "error".to_string(),
565 Value::String("Operation was cancelled".to_string()),
566 ),
567 ]),
568 );
569 return Some(cancelled_agent_result(
570 messages.clone(),
571 cycles.clone(),
572 shared_state.clone(),
573 task_token_usage(&controls),
574 ));
575 }
576 let (patched_call, short_circuit_result) = hook_manager.apply_before_tool_call(
577 &task,
578 cycle_index,
579 call.clone(),
580 &context,
581 );
582 let checkpoint_plan = checkpoint.plan_tool(
583 cycle_index,
584 &patched_call,
585 || {
586 let idempotency = crate::runtime::run_definition::tool_idempotency_for(
587 &self.tool_registry,
588 &patched_call.name,
589 );
590 let budget_usage = budget_snapshot(&budget_controller);
591 (idempotency, budget_usage)
592 },
593 messages,
594 cycles,
595 shared_state,
596 );
597 let checkpoint_plan = match checkpoint_plan {
598 CheckpointToolPlan::Continue(plan) => plan,
599 CheckpointToolPlan::Stop(result) => return Some(*result),
600 };
601 let tool_kind = self
602 .tool_registry
603 .get(&patched_call.name)
604 .map(|spec| spec.kind)
605 .ok();
606 let mut approval_failure = None;
607 let mut execution = if let Some(mut result) = short_circuit_result {
608 if needs_tool_call_id(&result.tool_call_id) {
609 result.tool_call_id = call.id.clone();
610 }
611 tool_orchestrator.observe_result_without_execution(
612 patched_call.clone(),
613 result,
614 &tool_run_options,
615 )
616 } else if let Some(result) = checkpoint_plan
617 .as_ref()
618 .and_then(|plan| plan.replay_result.clone())
619 {
620 context.idempotency_key = checkpoint_plan
621 .as_ref()
622 .and_then(|plan| plan.idempotency_key.clone());
623 tool_orchestrator.observe_result_without_execution(
624 patched_call.clone(),
625 result,
626 &tool_run_options,
627 )
628 } else {
629 let effective_tool_run_options = checkpoint.before_tool_dispatch(
630 tool_run_options.clone().idempotency_key(
631 checkpoint_plan
632 .as_ref()
633 .and_then(|plan| plan.idempotency_key.clone()),
634 ),
635 cycle_index,
636 );
637 let execution = match block_on_engine_tool_run(
638 tool_orchestrator.run_one_with_approval_and_metadata_deferred(
639 patched_call.clone(),
640 &mut context,
641 effective_tool_run_options.clone(),
642 |call, effective_requirement, approval_context, tool_metadata| {
643 let result = match approval_provider_result(
644 self,
645 &controls,
646 &task,
647 cycle_index,
648 call,
649 effective_requirement,
650 tool_metadata,
651 ) {
652 Ok(result) => result,
653 Err(error) => {
654 approval_failure = Some(error);
655 return Some(approval_error_result(
656 call,
657 "approval_provider_error",
658 "Approval provider failed.",
659 ));
660 }
661 };
662 if result.as_ref().is_some_and(|result| {
663 result.error_code.as_deref()
664 == Some("tool_approval_required")
665 }) {
666 self.capture_pending_tool_approval(
667 PendingToolApprovalCapture {
668 task: &task,
669 hook_manager: &hook_manager,
670 cycle_index,
671 call,
672 context: approval_context,
673 options: &effective_tool_run_options,
674 orchestrator: &tool_orchestrator,
675 result: result
676 .as_ref()
677 .expect("checked approval result"),
678 },
679 );
680 }
681 result
682 },
683 ),
684 ) {
685 Ok(execution) => execution,
686 Err(error) => crate::tools::orchestrator::DeferredToolExecution::without_lifecycle(
687 approval_error_result(
688 &patched_call,
689 "tool_orchestrator_error",
690 error.to_string(),
691 ),
692 ),
693 };
694 if let Some(result) =
695 checkpoint.pending_failure(messages, cycles, shared_state)
696 {
697 return Some(result);
698 }
699 execution
700 };
701 let execution_started = execution.execution_started();
702 let mut result = execution.result().clone();
703 if needs_tool_call_id(&result.tool_call_id) {
704 result.tool_call_id = patched_call.id.clone();
705 }
706 result = hook_manager.apply_after_tool_call(
707 &task,
708 cycle_index,
709 &patched_call,
710 &context,
711 result,
712 );
713 if needs_tool_call_id(&result.tool_call_id) {
714 result.tool_call_id = patched_call.id.clone();
715 }
716 let behavior_reason =
717 apply_tool_use_behavior(&task, &patched_call, &mut result);
718 execution.replace_result(result);
719 let result = execution.complete();
720 if let Some(error) = approval_failure {
721 *shared_state = context.shared_state.clone();
722 if let Some(controller) = &budget_controller {
723 lock_budget(controller).tool_batch_complete(
724 &controls,
725 cycle_index,
726 true,
727 false,
728 );
729 }
730 cycles.push(cycle);
731 self.emit_log(
732 &controls,
733 "cycle_failed",
734 BTreeMap::from([
735 ("cycle".to_string(), Value::from(cycle_index)),
736 ("error".to_string(), Value::String(error.to_string())),
737 ]),
738 );
739 return Some(failed_agent_result(
740 messages.clone(),
741 cycles.clone(),
742 shared_state.clone(),
743 error.to_string(),
744 task_token_usage(&controls),
745 ));
746 }
747 if let Some(result) = checkpoint.finish_tool(
748 cycle_index,
749 &patched_call,
750 &result,
751 || budget_snapshot(&budget_controller),
752 (messages, cycles, shared_state),
753 ) {
754 return Some(result);
755 }
756 if matches!(
757 tool_kind,
758 Some(ToolSpecKind::Agent | ToolSpecKind::BackgroundAgent)
759 ) && execution_started
760 {
761 self.emit_log(
762 &controls,
763 "sub_run_completed",
764 BTreeMap::from([
765 ("task_id".to_string(), Value::String(task.task_id.clone())),
766 (
767 "agent_name".to_string(),
768 Value::String(
769 task.metadata
770 .get("agent_name")
771 .and_then(Value::as_str)
772 .unwrap_or(&task.task_id)
773 .to_string(),
774 ),
775 ),
776 ("cycle".to_string(), Value::from(cycle_index)),
777 (
778 "parent_run_id".to_string(),
779 Value::String(task.task_id.clone()),
780 ),
781 (
782 "parent_tool_call_id".to_string(),
783 Value::String(patched_call.id.clone()),
784 ),
785 (
786 "status".to_string(),
787 super::logging::tool_result_status_value(result.status),
788 ),
789 (
790 "final_output".to_string(),
791 Value::String(result.content.clone()),
792 ),
793 ]),
794 );
795 }
796 self.emit_tool_result(&controls, cycle_index, &patched_call, &result);
797
798 let interruption_messages = collect_interruption_messages(&controls);
799 let steering_prompts = drain_steering_queue(&controls);
800 let steering_count = interruption_messages.len() + steering_prompts.len();
801 if steering_count == 0 && result.directive != ToolDirective::Continue {
802 directive_completion_reason =
803 behavior_reason.or(Some(match result.directive {
804 ToolDirective::WaitUser => CompletionReason::WaitUser,
805 ToolDirective::Finish => CompletionReason::ToolFinish,
806 ToolDirective::Continue => unreachable!(),
807 }));
808 directive_completion_tool_name = Some(patched_call.name.clone());
809 directive_result = Some(result.clone());
810 }
811 messages.push(result.to_message());
812 if let Some(image_notification) =
813 image_notification_from_tool_result(&result, task.native_multimodal)
814 {
815 image_notifications.push(image_notification);
816 }
817 cycle.tool_results.push(result);
818 if steering_count > 0 {
819 for prompt in &steering_prompts {
820 self.emit_log(
821 &controls,
822 "session_steer_interrupt",
823 BTreeMap::from([
824 ("cycle".to_string(), Value::from(cycle_index)),
825 (
826 "after_tool_call_id".to_string(),
827 Value::String(patched_call.id.clone()),
828 ),
829 (
830 "after_tool_name".to_string(),
831 Value::String(patched_call.name.clone()),
832 ),
833 ("prompt".to_string(), Value::String(prompt.clone())),
834 ]),
835 );
836 }
837 for skipped_call in response.tool_calls.iter().skip(call_index + 1) {
838 let skipped = skipped_tool_result(
839 skipped_call,
840 "skipped_due_to_steering",
841 "Tool skipped due to queued steering message.",
842 );
843 self.emit_skipped_tool_result(
844 &controls,
845 cycle_index,
846 skipped_call,
847 &skipped,
848 );
849 messages.push(skipped.to_message());
850 cycle.tool_results.push(skipped);
851 }
852 for prompt in &steering_prompts {
853 messages.push(crate::types::Message::user(prompt.clone()));
854 }
855 messages.extend(interruption_messages);
856 self.emit_log(
857 &controls,
858 "run_steered",
859 BTreeMap::from([
860 ("cycle".to_string(), Value::from(cycle_index)),
861 (
862 "after_tool_call_id".to_string(),
863 Value::String(patched_call.id.clone()),
864 ),
865 (
866 "after_tool_name".to_string(),
867 Value::String(patched_call.name.clone()),
868 ),
869 (
870 "prompt_count".to_string(),
871 Value::from(steering_count as u64),
872 ),
873 (
874 "steering_count".to_string(),
875 Value::from(steering_count as u64),
876 ),
877 ]),
878 );
879 break;
880 }
881 if directive_result.is_some() {
882 let (error_code, message) = match directive_result
883 .as_ref()
884 .map(|result| result.directive)
885 .unwrap_or(ToolDirective::Continue)
886 {
887 ToolDirective::WaitUser => (
888 "skipped_due_to_wait_user",
889 "Tool skipped because a previous tool requested user input.",
890 ),
891 ToolDirective::Finish => (
892 "skipped_due_to_finish",
893 "Tool skipped because a previous tool finished the task.",
894 ),
895 ToolDirective::Continue => {
896 ("skipped_due_to_directive", "Tool skipped.")
897 }
898 };
899 for skipped_call in response.tool_calls.iter().skip(call_index + 1) {
900 let skipped = skipped_tool_result(skipped_call, error_code, message);
901 self.emit_skipped_tool_result(
902 &controls,
903 cycle_index,
904 skipped_call,
905 &skipped,
906 );
907 messages.push(skipped.to_message());
908 cycle.tool_results.push(skipped);
909 }
910 break;
911 }
912 }
913 messages.extend(image_notifications);
914 *shared_state = context.shared_state.clone();
915
916 cycles.push(cycle);
917 let tool_boundary_result = observe_tool_batch_completion(
918 &budget_controller,
919 &controls,
920 cycle_index,
921 cancellation_token,
922 messages,
923 cycles,
924 shared_state,
925 );
926 if let Some(result) = checkpoint.update_budget_usage(
927 || budget_snapshot(&budget_controller),
928 messages,
929 cycles,
930 shared_state,
931 ) {
932 return Some(result);
933 }
934 if let Some(result) = tool_boundary_result {
935 return Some(result);
936 }
937 finalize_tool_cycle(ToolCycleFinalization {
938 runtime: self,
939 controls: &controls,
940 task: &task,
941 cycle_index,
942 directive_result: directive_result.as_ref(),
943 completion_reason: directive_completion_reason,
944 completion_tool_name: directive_completion_tool_name.as_deref(),
945 messages,
946 cycles,
947 shared_state,
948 checkpoint: &checkpoint,
949 budget_controller: &budget_controller,
950 persisted_denials: &active_after_cycle_denials,
951 })
952 },
953 effective_cancellation_token.as_ref(),
954 cycle_index_start,
955 cycle_count,
956 effective_budget_limits,
957 controls.initial_budget_usage.clone(),
958 controls
959 .checkpoint_controller
960 .clone()
961 .map(CheckpointRuntimeControl::into_controller),
962 );
963 if let Some(error) = checkpoint.take_llm_error() {
964 return Err(error);
965 }
966 if let Some(error) = pending_error {
967 return Err(error);
968 }
969 if backend_manages_checkpoint_cycles && !checkpoint.refresh_model_call_ledger()? {
970 model_call_ledger
971 .replace(result.token_usage.model_calls.clone())
972 .map_err(LlmError::Request)?;
973 }
974 result.token_usage = model_call_ledger.usage();
975 result = finalize_run_budget(
976 &budget_controller,
977 &controls,
978 effective_cancellation_token.as_ref(),
979 result,
980 );
981 result = finalize_terminal_projection(
982 self,
983 &controls,
984 effective_cancellation_token.as_ref(),
985 result,
986 );
987 Ok(result)
988 }
989}