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