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