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 = compacted_messages.clone();
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 = compacted_messages.clone();
422 (request_messages, request_tool_schemas) = hook_manager
423 .apply_before_llm(
424 &task,
425 cycle_index,
426 llm_messages,
427 retry_tool_schemas,
428 shared_state,
429 );
430 }
431 Err(error) => {
432 let message = error.to_string();
433 return Some(failed_agent_result(
434 messages.clone(),
435 cycles.clone(),
436 shared_state.clone(),
437 message,
438 task_token_usage(&controls),
439 ));
440 }
441 }
442 };
443 let model_budget_exhaustion = response.budget_exhaustion.clone();
444 let model_usage = response.usage.clone();
445 let response = hook_manager.apply_after_llm(
446 &task,
447 cycle_index,
448 &request_messages,
449 &request_tool_schemas,
450 response.response,
451 shared_state,
452 );
453 *messages = request_messages;
454 messages.push(assistant_message_from_response(&response));
455 let mut cycle = crate::types::CycleRecord::from_response(
456 cycle_index,
457 &response,
458 Vec::<ToolExecutionResult>::new(),
459 );
460 cycle.memory_compacted = memory_compacted;
461
462 let model_boundary_result = project_model_call_completion(
463 &budget_controller,
464 &controls,
465 model_budget_exhaustion,
466 cancellation_token,
467 &cycle,
468 messages,
469 cycles,
470 shared_state,
471 );
472 if let Some(result) = checkpoint.update_budget_usage(
473 || budget_snapshot(&budget_controller),
474 messages,
475 cycles,
476 shared_state,
477 ) {
478 return Some(result);
479 }
480 if let Some(result) = model_boundary_result {
481 return Some(result);
482 }
483 self.emit_cycle_llm_response(&controls, &cycle, &model_usage);
484
485 if response.tool_calls.is_empty() {
486 return finalize_no_tool_cycle(NoToolCycleFinalization {
487 runtime: self,
488 controls: &controls,
489 task: &task,
490 cycle_index,
491 response: &response,
492 cycle,
493 messages,
494 cycles,
495 shared_state,
496 checkpoint: &checkpoint,
497 budget_controller: &budget_controller,
498 persisted_denials: &active_after_cycle_denials,
499 });
500 }
501
502 if let Some(result) = preflight_tool_batch(
503 &budget_controller,
504 &controls,
505 cycle_index,
506 &response.tool_calls,
507 &cycle,
508 messages,
509 cycles,
510 shared_state,
511 ) {
512 return Some(result);
513 }
514
515 let PreparedToolBatch {
516 mut context,
517 orchestrator: tool_orchestrator,
518 options: tool_run_options,
519 } = self.prepare_tool_batch(ToolBatchSetup {
520 task: &task,
521 controls: &controls,
522 workspace_path: &workspace_path,
523 workspace_backend: &workspace_backend,
524 shared_state,
525 sub_task_manager: &sub_task_manager,
526 cycle_index,
527 cancellation_token,
528 child_budget_limits: &child_budget_limits,
529 request_tool_schemas: &request_tool_schemas,
530 after_cycle_disallowed_tools: &active_after_cycle_denials,
531 });
532
533 let mut directive_result = None;
534 let mut directive_completion_reason = None;
535 let mut directive_completion_tool_name = None;
536 let mut image_notifications = Vec::new();
537 for (call_index, call) in response.tool_calls.iter().enumerate() {
538 if cancellation_token.is_some_and(CancellationToken::is_cancelled)
539 || controls_cancelled(&controls)
540 {
541 *shared_state = context.shared_state.clone();
542 if let Some(controller) = &budget_controller {
543 lock_budget(controller).tool_batch_complete(
544 &controls,
545 cycle_index,
546 false,
547 true,
548 );
549 }
550 cycles.push(cycle);
551 self.emit_log(
552 &controls,
553 "run_cancelled",
554 BTreeMap::from([
555 ("cycle".to_string(), Value::from(cycle_index)),
556 (
557 "error".to_string(),
558 Value::String("Operation was cancelled".to_string()),
559 ),
560 ]),
561 );
562 return Some(cancelled_agent_result(
563 messages.clone(),
564 cycles.clone(),
565 shared_state.clone(),
566 task_token_usage(&controls),
567 ));
568 }
569 let (patched_call, short_circuit_result) = hook_manager.apply_before_tool_call(
570 &task,
571 cycle_index,
572 call.clone(),
573 &context,
574 );
575 let checkpoint_plan = checkpoint.plan_tool(
576 cycle_index,
577 &patched_call,
578 || {
579 let idempotency = super::run_definition::tool_idempotency_for(
580 &self.tool_registry,
581 &patched_call.name,
582 );
583 let budget_usage = budget_snapshot(&budget_controller);
584 (idempotency, budget_usage)
585 },
586 messages,
587 cycles,
588 shared_state,
589 );
590 let checkpoint_plan = match checkpoint_plan {
591 CheckpointToolPlan::Continue(plan) => plan,
592 CheckpointToolPlan::Stop(result) => return Some(*result),
593 };
594 let tool_kind = self
595 .tool_registry
596 .get(&patched_call.name)
597 .map(|spec| spec.kind)
598 .ok();
599 let mut approval_failure = None;
600 let mut execution = if let Some(mut result) = short_circuit_result {
601 if needs_tool_call_id(&result.tool_call_id) {
602 result.tool_call_id = call.id.clone();
603 }
604 tool_orchestrator.observe_result_without_execution(
605 patched_call.clone(),
606 result,
607 &tool_run_options,
608 )
609 } else if let Some(result) = checkpoint_plan
610 .as_ref()
611 .and_then(|plan| plan.replay_result.clone())
612 {
613 context.idempotency_key = checkpoint_plan
614 .as_ref()
615 .and_then(|plan| plan.idempotency_key.clone());
616 tool_orchestrator.observe_result_without_execution(
617 patched_call.clone(),
618 result,
619 &tool_run_options,
620 )
621 } else {
622 let effective_tool_run_options = checkpoint.before_tool_dispatch(
623 tool_run_options.clone().idempotency_key(
624 checkpoint_plan
625 .as_ref()
626 .and_then(|plan| plan.idempotency_key.clone()),
627 ),
628 cycle_index,
629 );
630 let execution = match block_on_engine_tool_run(
631 tool_orchestrator.run_one_with_approval_and_metadata_deferred(
632 patched_call.clone(),
633 &mut context,
634 effective_tool_run_options.clone(),
635 |call, effective_requirement, approval_context, tool_metadata| {
636 let result = match approval_provider_result(
637 self,
638 &controls,
639 &task,
640 cycle_index,
641 call,
642 effective_requirement,
643 tool_metadata,
644 ) {
645 Ok(result) => result,
646 Err(error) => {
647 approval_failure = Some(error);
648 return Some(approval_error_result(
649 call,
650 "approval_provider_error",
651 "Approval provider failed.",
652 ));
653 }
654 };
655 if result.as_ref().is_some_and(|result| {
656 result.error_code.as_deref()
657 == Some("tool_approval_required")
658 }) {
659 self.capture_pending_tool_approval(
660 PendingToolApprovalCapture {
661 task: &task,
662 hook_manager: &hook_manager,
663 cycle_index,
664 call,
665 context: approval_context,
666 options: &effective_tool_run_options,
667 orchestrator: &tool_orchestrator,
668 result: result
669 .as_ref()
670 .expect("checked approval result"),
671 },
672 );
673 }
674 result
675 },
676 ),
677 ) {
678 Ok(execution) => execution,
679 Err(error) => crate::tools::orchestrator::DeferredToolExecution::without_lifecycle(
680 approval_error_result(
681 &patched_call,
682 "tool_orchestrator_error",
683 error.to_string(),
684 ),
685 ),
686 };
687 if let Some(result) =
688 checkpoint.pending_failure(messages, cycles, shared_state)
689 {
690 return Some(result);
691 }
692 execution
693 };
694 let execution_started = execution.execution_started();
695 let mut result = execution.result().clone();
696 if needs_tool_call_id(&result.tool_call_id) {
697 result.tool_call_id = patched_call.id.clone();
698 }
699 result = hook_manager.apply_after_tool_call(
700 &task,
701 cycle_index,
702 &patched_call,
703 &context,
704 result,
705 );
706 if needs_tool_call_id(&result.tool_call_id) {
707 result.tool_call_id = patched_call.id.clone();
708 }
709 let behavior_reason =
710 apply_tool_use_behavior(&task, &patched_call, &mut result);
711 execution.replace_result(result);
712 let result = execution.complete();
713 if let Some(error) = approval_failure {
714 *shared_state = context.shared_state.clone();
715 if let Some(controller) = &budget_controller {
716 lock_budget(controller).tool_batch_complete(
717 &controls,
718 cycle_index,
719 true,
720 false,
721 );
722 }
723 cycles.push(cycle);
724 self.emit_log(
725 &controls,
726 "cycle_failed",
727 BTreeMap::from([
728 ("cycle".to_string(), Value::from(cycle_index)),
729 ("error".to_string(), Value::String(error.to_string())),
730 ]),
731 );
732 return Some(failed_agent_result(
733 messages.clone(),
734 cycles.clone(),
735 shared_state.clone(),
736 error.to_string(),
737 task_token_usage(&controls),
738 ));
739 }
740 if let Some(result) = checkpoint.finish_tool(
741 cycle_index,
742 &patched_call,
743 &result,
744 || budget_snapshot(&budget_controller),
745 (messages, cycles, shared_state),
746 ) {
747 return Some(result);
748 }
749 if matches!(
750 tool_kind,
751 Some(ToolSpecKind::Agent | ToolSpecKind::BackgroundAgent)
752 ) && execution_started
753 {
754 self.emit_log(
755 &controls,
756 "sub_run_completed",
757 BTreeMap::from([
758 ("task_id".to_string(), Value::String(task.task_id.clone())),
759 (
760 "agent_name".to_string(),
761 Value::String(
762 task.metadata
763 .get("agent_name")
764 .and_then(Value::as_str)
765 .unwrap_or(&task.task_id)
766 .to_string(),
767 ),
768 ),
769 ("cycle".to_string(), Value::from(cycle_index)),
770 (
771 "parent_run_id".to_string(),
772 Value::String(task.task_id.clone()),
773 ),
774 (
775 "parent_tool_call_id".to_string(),
776 Value::String(patched_call.id.clone()),
777 ),
778 (
779 "status".to_string(),
780 self::logging::tool_result_status_value(result.status),
781 ),
782 (
783 "final_output".to_string(),
784 Value::String(result.content.clone()),
785 ),
786 ]),
787 );
788 }
789 self.emit_tool_result(&controls, cycle_index, &patched_call, &result);
790
791 let interruption_messages = collect_interruption_messages(&controls);
792 let steering_prompts = drain_steering_queue(&controls);
793 let steering_count = interruption_messages.len() + steering_prompts.len();
794 if steering_count == 0 && result.directive != ToolDirective::Continue {
795 directive_completion_reason =
796 behavior_reason.or(Some(match result.directive {
797 ToolDirective::WaitUser => CompletionReason::WaitUser,
798 ToolDirective::Finish => CompletionReason::ToolFinish,
799 ToolDirective::Continue => unreachable!(),
800 }));
801 directive_completion_tool_name = Some(patched_call.name.clone());
802 directive_result = Some(result.clone());
803 }
804 messages.push(result.to_message());
805 if let Some(image_notification) =
806 image_notification_from_tool_result(&result, task.native_multimodal)
807 {
808 image_notifications.push(image_notification);
809 }
810 cycle.tool_results.push(result);
811 if steering_count > 0 {
812 for prompt in &steering_prompts {
813 self.emit_log(
814 &controls,
815 "session_steer_interrupt",
816 BTreeMap::from([
817 ("cycle".to_string(), Value::from(cycle_index)),
818 (
819 "after_tool_call_id".to_string(),
820 Value::String(patched_call.id.clone()),
821 ),
822 (
823 "after_tool_name".to_string(),
824 Value::String(patched_call.name.clone()),
825 ),
826 ("prompt".to_string(), Value::String(prompt.clone())),
827 ]),
828 );
829 }
830 for skipped_call in response.tool_calls.iter().skip(call_index + 1) {
831 let skipped = skipped_tool_result(
832 skipped_call,
833 "skipped_due_to_steering",
834 "Tool skipped due to queued steering message.",
835 );
836 self.emit_skipped_tool_result(
837 &controls,
838 cycle_index,
839 skipped_call,
840 &skipped,
841 );
842 messages.push(skipped.to_message());
843 cycle.tool_results.push(skipped);
844 }
845 for prompt in &steering_prompts {
846 messages.push(crate::types::Message::user(prompt.clone()));
847 }
848 messages.extend(interruption_messages);
849 self.emit_log(
850 &controls,
851 "run_steered",
852 BTreeMap::from([
853 ("cycle".to_string(), Value::from(cycle_index)),
854 (
855 "after_tool_call_id".to_string(),
856 Value::String(patched_call.id.clone()),
857 ),
858 (
859 "after_tool_name".to_string(),
860 Value::String(patched_call.name.clone()),
861 ),
862 (
863 "prompt_count".to_string(),
864 Value::from(steering_count as u64),
865 ),
866 (
867 "steering_count".to_string(),
868 Value::from(steering_count as u64),
869 ),
870 ]),
871 );
872 break;
873 }
874 if directive_result.is_some() {
875 let (error_code, message) = match directive_result
876 .as_ref()
877 .map(|result| result.directive)
878 .unwrap_or(ToolDirective::Continue)
879 {
880 ToolDirective::WaitUser => (
881 "skipped_due_to_wait_user",
882 "Tool skipped because a previous tool requested user input.",
883 ),
884 ToolDirective::Finish => (
885 "skipped_due_to_finish",
886 "Tool skipped because a previous tool finished the task.",
887 ),
888 ToolDirective::Continue => {
889 ("skipped_due_to_directive", "Tool skipped.")
890 }
891 };
892 for skipped_call in response.tool_calls.iter().skip(call_index + 1) {
893 let skipped = skipped_tool_result(skipped_call, error_code, message);
894 self.emit_skipped_tool_result(
895 &controls,
896 cycle_index,
897 skipped_call,
898 &skipped,
899 );
900 messages.push(skipped.to_message());
901 cycle.tool_results.push(skipped);
902 }
903 break;
904 }
905 }
906 messages.extend(image_notifications);
907 *shared_state = context.shared_state.clone();
908
909 cycles.push(cycle);
910 let tool_boundary_result = observe_tool_batch_completion(
911 &budget_controller,
912 &controls,
913 cycle_index,
914 cancellation_token,
915 messages,
916 cycles,
917 shared_state,
918 );
919 if let Some(result) = checkpoint.update_budget_usage(
920 || budget_snapshot(&budget_controller),
921 messages,
922 cycles,
923 shared_state,
924 ) {
925 return Some(result);
926 }
927 if let Some(result) = tool_boundary_result {
928 return Some(result);
929 }
930 finalize_tool_cycle(ToolCycleFinalization {
931 runtime: self,
932 controls: &controls,
933 task: &task,
934 cycle_index,
935 directive_result: directive_result.as_ref(),
936 completion_reason: directive_completion_reason,
937 completion_tool_name: directive_completion_tool_name.as_deref(),
938 messages,
939 cycles,
940 shared_state,
941 checkpoint: &checkpoint,
942 budget_controller: &budget_controller,
943 persisted_denials: &active_after_cycle_denials,
944 })
945 },
946 effective_cancellation_token.as_ref(),
947 cycle_index_start,
948 cycle_count,
949 effective_budget_limits,
950 controls.initial_budget_usage.clone(),
951 controls
952 .checkpoint_controller
953 .clone()
954 .map(CheckpointRuntimeControl::into_controller),
955 );
956 if let Some(error) = checkpoint.take_llm_error() {
957 return Err(error);
958 }
959 if let Some(error) = pending_error {
960 return Err(error);
961 }
962 if backend_manages_checkpoint_cycles && !checkpoint.refresh_model_call_ledger()? {
963 model_call_ledger
964 .replace(result.token_usage.model_calls.clone())
965 .map_err(LlmError::Request)?;
966 }
967 result.token_usage = model_call_ledger.usage();
968 result = finalize_run_budget(
969 &budget_controller,
970 &controls,
971 effective_cancellation_token.as_ref(),
972 result,
973 );
974 result = finalize_terminal_projection(
975 self,
976 &controls,
977 effective_cancellation_token.as_ref(),
978 result,
979 );
980 Ok(result)
981 }
982}