1mod completion;
2mod construction;
3mod controls;
4mod cycle_inputs;
5mod helpers;
6mod logging;
7mod memory;
8mod planning;
9mod run_setup;
10mod state;
11
12use std::collections::{BTreeMap, BTreeSet};
13
14use futures_util::FutureExt;
15use serde_json::Value;
16
17use crate::approval::{block_on_approval_future, ApprovalRequest};
18use crate::events::RunEvent;
19use crate::llm::{LlmClient, LlmError, LlmRequest};
20use crate::memory::{
21 provider::block_on_memory_future, CompactionExhaustedError, MemoryManager, MemoryProvider,
22};
23use crate::tools::{
24 ApprovalDecision, ToolContext, ToolError, ToolOrchestrator, ToolRunOptions, ToolSpecKind,
25};
26use crate::types::{
27 AgentResult, AgentStatus, AgentTask, Message, ToolCall, ToolDirective, ToolExecutionResult,
28 ToolResultStatus,
29};
30
31use super::cancellation::CancellationToken;
32use super::context::ExecutionContext;
33
34use super::cycle_runner::{is_prompt_too_long_error, MAX_PROMPT_TOO_LONG_RETRIES};
35use super::results::assistant_message_from_response;
36use super::token_usage::normalize_token_usage;
37use super::tool_call_runner::{needs_tool_call_id, skipped_tool_result};
38
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, image_notification_from_tool_result,
45 previous_cycle_memory_usage,
46};
47use self::run_setup::{prepare_run_setup, PreparedRun};
48pub use self::state::AgentRuntime;
49
50pub use crate::runtime::sub_agent_sessions::{
51 _register_sub_agent_session, _unregister_sub_agent_session, get_sub_agent_session,
52 steer_sub_agent_session, subscribe_sub_agent_session,
53};
54pub use controls::{
55 BeforeCycleMessageProvider, InterruptionMessageProvider, RuntimeEventHandler,
56 RuntimeLogCallback, RuntimeLogHandler, RuntimeRunControls,
57};
58
59impl<C: LlmClient + Clone + 'static> AgentRuntime<C> {
60 pub fn run(&self, task: AgentTask) -> Result<AgentResult, LlmError> {
61 self.run_with_controls(task, RuntimeRunControls::default())
62 }
63
64 pub fn run_with_controls(
65 &self,
66 task: AgentTask,
67 controls: RuntimeRunControls,
68 ) -> Result<AgentResult, LlmError> {
69 let PreparedRun {
70 task,
71 messages,
72 cycles,
73 shared_state,
74 workspace_path,
75 workspace_backend,
76 sub_task_manager,
77 mut memory_manager,
78 } = prepare_run_setup(self, task, &controls);
79 self.emit_run_started(&controls, &task, &workspace_path);
80
81 if controls_cancelled(&controls) {
82 self.emit_log(
83 &controls,
84 "run_cancelled",
85 BTreeMap::from([(
86 "error".to_string(),
87 Value::String("Operation was cancelled".to_string()),
88 )]),
89 );
90 return Ok(cancelled_agent_result(messages, cycles, shared_state));
91 }
92
93 let effective_cancellation_token = controls.effective_cancellation_token();
94 let effective_stream_callback = controls.effective_stream_callback();
95 let mut pending_error = None;
96 let result = self.execution_backend.execute(
97 &task,
98 messages,
99 shared_state,
100 |cycle_index, messages, cycles, shared_state, cancellation_token| {
101 if cancellation_token.is_some_and(CancellationToken::is_cancelled)
102 || controls_cancelled(&controls)
103 {
104 self.emit_log(
105 &controls,
106 "run_cancelled",
107 BTreeMap::from([
108 ("cycle".to_string(), Value::from(cycle_index)),
109 (
110 "error".to_string(),
111 Value::String("Operation was cancelled".to_string()),
112 ),
113 ]),
114 );
115 return Some(cancelled_agent_result(
116 messages.clone(),
117 cycles.clone(),
118 shared_state.clone(),
119 ));
120 }
121 self.apply_cycle_inputs(&controls, cycle_index, messages, shared_state);
122 if cancellation_token.is_some_and(CancellationToken::is_cancelled)
123 || controls_cancelled(&controls)
124 {
125 self.emit_log(
126 &controls,
127 "run_cancelled",
128 BTreeMap::from([
129 ("cycle".to_string(), Value::from(cycle_index)),
130 (
131 "error".to_string(),
132 Value::String("Operation was cancelled".to_string()),
133 ),
134 ]),
135 );
136 return Some(cancelled_agent_result(
137 messages.clone(),
138 cycles.clone(),
139 shared_state.clone(),
140 ));
141 }
142 self.emit_log(
143 &controls,
144 "cycle_started",
145 BTreeMap::from([
146 ("cycle".to_string(), Value::from(cycle_index)),
147 ("max_cycles".to_string(), Value::from(task.max_cycles)),
148 ("message_count".to_string(), Value::from(messages.len())),
149 ]),
150 );
151 let hook_manager = self.hook_manager();
152 let pre_compact_messages = hook_manager.apply_before_memory_compact(
153 &task,
154 cycle_index,
155 messages.clone(),
156 shared_state,
157 );
158 let pre_compact_messages =
159 memory_manager.apply_session_memory_context(&pre_compact_messages);
160 let (previous_prompt_tokens, recent_tool_call_ids) =
161 previous_cycle_memory_usage(cycles);
162 let memory_compact_event = memory_compact_started_event(
163 controls.execution_context.as_ref(),
164 &memory_manager,
165 &task,
166 cycle_index,
167 &pre_compact_messages,
168 previous_prompt_tokens,
169 recent_tool_call_ids.as_ref(),
170 );
171 if let Some(event) = memory_compact_event.as_ref() {
172 notify_memory_before_compact(controls.execution_context.as_ref(), event);
173 }
174 let (prepared_messages, memory_compacted) = memory_manager
175 .compact_for_cycle_with_usage(
176 &pre_compact_messages,
177 cycle_index,
178 false,
179 previous_prompt_tokens,
180 recent_tool_call_ids.as_ref(),
181 );
182 if memory_compacted {
183 if let Some(started_event) = memory_compact_event.as_ref() {
184 let completed = RunEvent::memory_compact_completed(
185 started_event.run_id().to_string(),
186 started_event.trace_id().to_string(),
187 started_event.agent_name().unwrap_or("agent").to_string(),
188 cycle_index,
189 pre_compact_messages.len(),
190 prepared_messages.len(),
191 None,
192 );
193 notify_memory_after_compact(
194 controls.execution_context.as_ref(),
195 &completed,
196 );
197 }
198 }
199 *messages = prepared_messages;
200 let tool_schemas = self.planned_tool_schemas(&task);
201 let llm_messages = memory_manager.apply_session_memory_context(messages);
202 let (request_messages, request_tool_schemas) = hook_manager.apply_before_llm(
203 &task,
204 cycle_index,
205 llm_messages,
206 tool_schemas,
207 shared_state,
208 );
209 let mut request_messages = request_messages;
210 let mut request_tool_schemas = request_tool_schemas;
211 let mut memory_compacted = memory_compacted;
212 let mut prompt_too_long_retries = 0;
213 let response = loop {
214 let mut request = LlmRequest::new(task.model.clone(), request_messages.clone());
215 request.tools = request_tool_schemas.clone();
216 if let Some(execution_context) = controls.execution_context.as_ref() {
217 request.metadata = serde_json::to_value(&execution_context.metadata)
218 .unwrap_or(Value::Null);
219 }
220 match self
221 .llm_client
222 .complete_with_stream(request, effective_stream_callback.clone())
223 {
224 Ok(response) => break response,
225 Err(error) if is_prompt_too_long_error(&error) => {
226 prompt_too_long_retries += 1;
227 if prompt_too_long_retries > MAX_PROMPT_TOO_LONG_RETRIES {
228 let error =
229 LlmError::CompactionExhausted(CompactionExhaustedError::new(
230 prompt_too_long_retries,
231 Some(error.to_string()),
232 ));
233 let message = error.to_string();
234 pending_error = Some(error);
235 return Some(failed_agent_result(
236 messages.clone(),
237 cycles.clone(),
238 shared_state.clone(),
239 message,
240 ));
241 }
242 memory_compacted = true;
243 let retry_messages = if prompt_too_long_retries == 1 {
244 let (compacted, _) = memory_manager.compact_for_cycle(
245 &request_messages,
246 cycle_index,
247 true,
248 );
249 compacted
250 } else {
251 memory_manager.emergency_compact(
252 &request_messages,
253 (0.2 * prompt_too_long_retries as f64).min(0.95),
254 )
255 };
256 let retry_tool_schemas = self.planned_tool_schemas(&task);
257 let llm_messages =
258 memory_manager.apply_session_memory_context(&retry_messages);
259 (request_messages, request_tool_schemas) = hook_manager
260 .apply_before_llm(
261 &task,
262 cycle_index,
263 llm_messages,
264 retry_tool_schemas,
265 shared_state,
266 );
267 }
268 Err(error) => {
269 let message = error.to_string();
270 pending_error = Some(error);
271 return Some(failed_agent_result(
272 messages.clone(),
273 cycles.clone(),
274 shared_state.clone(),
275 message,
276 ));
277 }
278 }
279 };
280 let response = hook_manager.apply_after_llm(
281 &task,
282 cycle_index,
283 &request_messages,
284 &request_tool_schemas,
285 response,
286 shared_state,
287 );
288 *messages = request_messages;
289 messages.push(assistant_message_from_response(&response));
290 let mut cycle = crate::types::CycleRecord::from_response(
291 cycle_index,
292 &response,
293 Vec::<ToolExecutionResult>::new(),
294 );
295 cycle.memory_compacted = memory_compacted;
296 if !cycle.token_usage.has_usage() {
297 cycle.token_usage =
298 normalize_token_usage(response.raw.get("usage").unwrap_or(&Value::Null));
299 }
300 self.emit_cycle_llm_response(&controls, &cycle);
301
302 if response.tool_calls.is_empty() {
303 if let Some(result) = handle_no_tool_response(NoToolResponseRequest {
304 runtime: self,
305 controls: &controls,
306 task: &task,
307 cycle_index,
308 response: &response,
309 messages,
310 cycles,
311 cycle,
312 shared_state,
313 }) {
314 return Some(result);
315 }
316 return None;
317 }
318
319 let sub_task_runner = self.build_sub_task_runner(
320 &task,
321 workspace_path.clone(),
322 workspace_backend.clone(),
323 shared_state.clone(),
324 sub_task_manager.clone(),
325 super::sub_agents::SubTaskCallbacks {
326 stream_callback: effective_stream_callback.clone(),
327 parent_log_handler: self.log_handler.clone(),
328 parent_event_handler: controls.log_handler.clone(),
329 },
330 );
331 let mut tool_metadata = controls
332 .execution_context
333 .as_ref()
334 .map(|context| context.metadata.clone())
335 .unwrap_or_default();
336 tool_metadata.extend(task.metadata.clone());
337 let mut context = ToolContext {
338 workspace: workspace_path.clone(),
339 shared_state: shared_state.clone(),
340 cycle_index,
341 task_id: task.task_id.clone(),
342 metadata: tool_metadata,
343 workspace_backend: workspace_backend.clone(),
344 model_provider: controls.model_provider.clone(),
345 sub_task_runner,
346 sub_task_manager: Some(sub_task_manager.clone()),
347 execution_backend: Some(self.execution_backend.clone()),
348 };
349
350 let mut directive_result = None;
351 let mut image_notifications = Vec::new();
352 let tool_orchestrator =
353 ToolOrchestrator::from_tools(self.tool_registry.executors());
354 for (call_index, call) in response.tool_calls.iter().enumerate() {
355 if cancellation_token.is_some_and(CancellationToken::is_cancelled)
356 || controls_cancelled(&controls)
357 {
358 *shared_state = context.shared_state.clone();
359 cycles.push(cycle);
360 self.emit_log(
361 &controls,
362 "run_cancelled",
363 BTreeMap::from([
364 ("cycle".to_string(), Value::from(cycle_index)),
365 (
366 "error".to_string(),
367 Value::String("Operation was cancelled".to_string()),
368 ),
369 ]),
370 );
371 return Some(cancelled_agent_result(
372 messages.clone(),
373 cycles.clone(),
374 shared_state.clone(),
375 ));
376 }
377 let (patched_call, short_circuit_result) = hook_manager.apply_before_tool_call(
378 &task,
379 cycle_index,
380 call.clone(),
381 &context,
382 );
383 self.emit_log(
384 &controls,
385 "tool_call_started",
386 BTreeMap::from([
387 ("task_id".to_string(), Value::String(task.task_id.clone())),
388 (
389 "agent_name".to_string(),
390 Value::String(
391 task.metadata
392 .get("agent_name")
393 .and_then(Value::as_str)
394 .unwrap_or(&task.task_id)
395 .to_string(),
396 ),
397 ),
398 ("cycle".to_string(), Value::from(cycle_index)),
399 (
400 "tool_name".to_string(),
401 Value::String(patched_call.name.clone()),
402 ),
403 (
404 "tool_arguments".to_string(),
405 Value::Object(patched_call.arguments.clone().into_iter().collect()),
406 ),
407 (
408 "tool_call_id".to_string(),
409 Value::String(patched_call.id.clone()),
410 ),
411 ]),
412 );
413 let tool_kind = self
414 .tool_registry
415 .get(&patched_call.name)
416 .map(|spec| spec.kind)
417 .ok();
418 if matches!(
419 tool_kind,
420 Some(ToolSpecKind::Agent | ToolSpecKind::BackgroundAgent)
421 ) {
422 self.emit_log(
423 &controls,
424 "sub_run_started",
425 BTreeMap::from([
426 ("task_id".to_string(), Value::String(task.task_id.clone())),
427 (
428 "agent_name".to_string(),
429 Value::String(
430 task.metadata
431 .get("agent_name")
432 .and_then(Value::as_str)
433 .unwrap_or(&task.task_id)
434 .to_string(),
435 ),
436 ),
437 ("cycle".to_string(), Value::from(cycle_index)),
438 (
439 "parent_run_id".to_string(),
440 Value::String(task.task_id.clone()),
441 ),
442 (
443 "parent_tool_call_id".to_string(),
444 Value::String(patched_call.id.clone()),
445 ),
446 (
447 "task_id_hint".to_string(),
448 Value::String(format!("sub_run:{}", patched_call.id)),
449 ),
450 ]),
451 );
452 }
453 let provider_approval_result = approval_provider_result(
454 self,
455 &controls,
456 &task,
457 cycle_index,
458 &patched_call,
459 );
460 let mut result = if let Some(mut result) = short_circuit_result {
461 if needs_tool_call_id(&result.tool_call_id) {
462 result.tool_call_id = call.id.clone();
463 }
464 result
465 } else if let Some(result) = provider_approval_result {
466 result
467 } else {
468 let mut result = match block_on_engine_tool_run(tool_orchestrator.run_one(
469 patched_call.clone(),
470 &mut context,
471 ToolRunOptions::default(),
472 )) {
473 Ok(result) => result,
474 Err(error) => approval_error_result(
475 &patched_call,
476 "tool_orchestrator_error",
477 error.to_string(),
478 ),
479 };
480 if needs_tool_call_id(&result.tool_call_id) {
481 result.tool_call_id = patched_call.id.clone();
482 }
483 result
484 };
485 result = hook_manager.apply_after_tool_call(
486 &task,
487 cycle_index,
488 &patched_call,
489 &context,
490 result,
491 );
492 if needs_tool_call_id(&result.tool_call_id) {
493 result.tool_call_id = patched_call.id.clone();
494 }
495 if matches!(
496 tool_kind,
497 Some(ToolSpecKind::Agent | ToolSpecKind::BackgroundAgent)
498 ) {
499 self.emit_log(
500 &controls,
501 "sub_run_completed",
502 BTreeMap::from([
503 ("task_id".to_string(), Value::String(task.task_id.clone())),
504 (
505 "agent_name".to_string(),
506 Value::String(
507 task.metadata
508 .get("agent_name")
509 .and_then(Value::as_str)
510 .unwrap_or(&task.task_id)
511 .to_string(),
512 ),
513 ),
514 ("cycle".to_string(), Value::from(cycle_index)),
515 (
516 "parent_run_id".to_string(),
517 Value::String(task.task_id.clone()),
518 ),
519 (
520 "parent_tool_call_id".to_string(),
521 Value::String(patched_call.id.clone()),
522 ),
523 (
524 "status".to_string(),
525 serde_json::to_value(result.status).unwrap_or(Value::Null),
526 ),
527 (
528 "final_output".to_string(),
529 Value::String(result.content.clone()),
530 ),
531 ]),
532 );
533 }
534 self.emit_tool_result(&controls, cycle_index, &patched_call, &result);
535
536 let interruption_messages = collect_interruption_messages(&controls);
537 let steering_prompts = drain_steering_queue(&controls);
538 let steering_count = interruption_messages.len() + steering_prompts.len();
539 if steering_count == 0 && result.directive != ToolDirective::Continue {
540 directive_result = Some(result.clone());
541 }
542 messages.push(result.to_message());
543 if let Some(image_notification) =
544 image_notification_from_tool_result(&result, task.native_multimodal)
545 {
546 image_notifications.push(image_notification);
547 }
548 cycle.tool_results.push(result);
549 if steering_count > 0 {
550 for prompt in &steering_prompts {
551 self.emit_log(
552 &controls,
553 "session_steer_interrupt",
554 BTreeMap::from([
555 ("cycle".to_string(), Value::from(cycle_index)),
556 (
557 "after_tool_call_id".to_string(),
558 Value::String(patched_call.id.clone()),
559 ),
560 (
561 "after_tool_name".to_string(),
562 Value::String(patched_call.name.clone()),
563 ),
564 ("prompt".to_string(), Value::String(prompt.clone())),
565 ]),
566 );
567 }
568 for skipped_call in response.tool_calls.iter().skip(call_index + 1) {
569 let skipped = skipped_tool_result(
570 skipped_call,
571 "skipped_due_to_steering",
572 "Tool skipped due to queued steering message.",
573 );
574 self.emit_tool_result(&controls, cycle_index, skipped_call, &skipped);
575 messages.push(skipped.to_message());
576 cycle.tool_results.push(skipped);
577 }
578 for prompt in &steering_prompts {
579 messages.push(crate::types::Message::user(prompt.clone()));
580 }
581 messages.extend(interruption_messages);
582 self.emit_log(
583 &controls,
584 "run_steered",
585 BTreeMap::from([
586 ("cycle".to_string(), Value::from(cycle_index)),
587 (
588 "after_tool_call_id".to_string(),
589 Value::String(patched_call.id.clone()),
590 ),
591 (
592 "after_tool_name".to_string(),
593 Value::String(patched_call.name.clone()),
594 ),
595 (
596 "prompt_count".to_string(),
597 Value::from(steering_count as u64),
598 ),
599 (
600 "steering_count".to_string(),
601 Value::from(steering_count as u64),
602 ),
603 ]),
604 );
605 break;
606 }
607 if directive_result.is_some() {
608 let (error_code, message) = match directive_result
609 .as_ref()
610 .map(|result| result.directive)
611 .unwrap_or(ToolDirective::Continue)
612 {
613 ToolDirective::WaitUser => (
614 "skipped_due_to_wait_user",
615 "Tool skipped because a previous tool requested user input.",
616 ),
617 ToolDirective::Finish => (
618 "skipped_due_to_finish",
619 "Tool skipped because a previous tool finished the task.",
620 ),
621 ToolDirective::Continue => {
622 ("skipped_due_to_directive", "Tool skipped.")
623 }
624 };
625 for skipped_call in response.tool_calls.iter().skip(call_index + 1) {
626 let skipped = skipped_tool_result(skipped_call, error_code, message);
627 self.emit_tool_result(&controls, cycle_index, skipped_call, &skipped);
628 messages.push(skipped.to_message());
629 cycle.tool_results.push(skipped);
630 }
631 break;
632 }
633 }
634 messages.extend(image_notifications);
635 *shared_state = context.shared_state.clone();
636
637 cycles.push(cycle);
638 if let Some(directive_result) = directive_result.as_ref() {
639 if let Some(result) = handle_directive_result(DirectiveResultRequest {
640 runtime: self,
641 controls: &controls,
642 task: &task,
643 cycle_index,
644 result: directive_result,
645 messages,
646 cycles,
647 shared_state,
648 }) {
649 return Some(result);
650 }
651 }
652 None
653 },
654 effective_cancellation_token.as_ref(),
655 task.max_cycles,
656 );
657 if let Some(error) = pending_error {
658 return Err(error);
659 }
660 if result.status == AgentStatus::MaxCycles {
661 self.emit_run_max_cycles(&controls, &result);
662 }
663 Ok(result)
664 }
665}
666
667fn approval_provider_result<C: LlmClient>(
668 runtime: &AgentRuntime<C>,
669 controls: &RuntimeRunControls,
670 task: &AgentTask,
671 cycle_index: u32,
672 call: &ToolCall,
673) -> Option<ToolExecutionResult> {
674 if call.name == "task_finish" {
675 return None;
676 }
677 let execution_context = controls.execution_context.as_ref()?;
678 let provider = execution_context.approval_provider.as_ref()?;
679 let broker = execution_context.approval_broker.as_ref()?;
680 let agent_name = task
681 .metadata
682 .get("agent_name")
683 .and_then(Value::as_str)
684 .unwrap_or(&task.task_id)
685 .to_string();
686 let request = ApprovalRequest::for_tool_call(
687 task.task_id.clone(),
688 task.task_id.clone(),
689 agent_name,
690 cycle_index,
691 call,
692 );
693 if !provider.should_request(&request) {
694 return None;
695 }
696
697 let decision = match block_on_approval_future(provider.decide(&request)) {
698 Ok(Some(ApprovalDecision::Approved)) => return None,
699 Ok(Some(ApprovalDecision::Denied(reason))) => ApprovalDecision::Denied(reason),
700 Ok(Some(ApprovalDecision::TimedOut(reason))) => ApprovalDecision::TimedOut(reason),
701 Ok(Some(ApprovalDecision::NeedsApproval)) | Ok(None) => {
702 if let Err(error) = broker.register(request.clone()) {
703 return Some(approval_error_result(
704 call,
705 "approval_broker_error",
706 error.to_string(),
707 ));
708 }
709 runtime.emit_log(
710 controls,
711 "approval_requested",
712 BTreeMap::from([
713 ("task_id".to_string(), Value::String(request.run_id.clone())),
714 (
715 "agent_name".to_string(),
716 Value::String(request.agent_name.clone()),
717 ),
718 ("cycle".to_string(), Value::from(cycle_index)),
719 (
720 "request_id".to_string(),
721 Value::String(request.request_id.clone()),
722 ),
723 (
724 "tool_call_id".to_string(),
725 Value::String(request.tool_call_id.clone()),
726 ),
727 (
728 "tool_name".to_string(),
729 Value::String(request.tool_name.clone()),
730 ),
731 (
732 "preview".to_string(),
733 Value::String(request.preview.clone()),
734 ),
735 ]),
736 );
737 broker
738 .wait_blocking(&request.request_id, execution_context.approval_timeout)
739 .unwrap_or_else(|error| ApprovalDecision::deny(error.to_string()))
740 }
741 Err(error) => ApprovalDecision::deny(error.to_string()),
742 };
743
744 runtime.emit_log(
745 controls,
746 "approval_resolved",
747 BTreeMap::from([
748 ("task_id".to_string(), Value::String(request.run_id.clone())),
749 (
750 "agent_name".to_string(),
751 Value::String(request.agent_name.clone()),
752 ),
753 ("cycle".to_string(), Value::from(cycle_index)),
754 (
755 "request_id".to_string(),
756 Value::String(request.request_id.clone()),
757 ),
758 (
759 "tool_call_id".to_string(),
760 Value::String(request.tool_call_id.clone()),
761 ),
762 (
763 "tool_name".to_string(),
764 Value::String(request.tool_name.clone()),
765 ),
766 (
767 "approved".to_string(),
768 Value::Bool(matches!(decision, ApprovalDecision::Approved)),
769 ),
770 ]),
771 );
772
773 match decision {
774 ApprovalDecision::Approved => None,
775 ApprovalDecision::NeedsApproval => Some(approval_error_result(
776 call,
777 "approval_unresolved",
778 "Approval was not resolved.",
779 )),
780 ApprovalDecision::Denied(reason) => {
781 Some(approval_error_result(call, "approval_denied", reason))
782 }
783 ApprovalDecision::TimedOut(reason) => {
784 Some(approval_error_result(call, "approval_timeout", reason))
785 }
786 }
787}
788
789fn memory_compact_started_event(
790 execution_context: Option<&ExecutionContext>,
791 memory_manager: &MemoryManager,
792 task: &AgentTask,
793 cycle_index: u32,
794 messages: &[Message],
795 previous_prompt_tokens: Option<u64>,
796 recent_tool_call_ids: Option<&BTreeSet<String>>,
797) -> Option<RunEvent> {
798 let providers = memory_providers(execution_context);
799 if providers.is_empty() {
800 return None;
801 }
802 let usage = memory_manager.estimate_memory_usage_percentage(
803 messages,
804 previous_prompt_tokens,
805 recent_tool_call_ids,
806 );
807 if usage <= 100 {
808 return None;
809 }
810 let agent_name = task
811 .metadata
812 .get("agent_name")
813 .and_then(Value::as_str)
814 .unwrap_or(&task.task_id)
815 .to_string();
816 Some(RunEvent::memory_compact_started(
817 task.task_id.clone(),
818 task.task_id.clone(),
819 agent_name,
820 cycle_index,
821 messages.len(),
822 previous_prompt_tokens,
823 ))
824}
825
826fn notify_memory_before_compact(execution_context: Option<&ExecutionContext>, event: &RunEvent) {
827 for provider in memory_providers(execution_context) {
828 let _ = block_on_memory_future(provider.before_compact(event));
829 }
830}
831
832fn notify_memory_after_compact(execution_context: Option<&ExecutionContext>, event: &RunEvent) {
833 for provider in memory_providers(execution_context) {
834 let _ = block_on_memory_future(provider.after_compact(event));
835 }
836}
837
838fn memory_providers(
839 execution_context: Option<&ExecutionContext>,
840) -> Vec<&std::sync::Arc<dyn MemoryProvider>> {
841 execution_context
842 .map(|context| context.memory_providers.iter().collect())
843 .unwrap_or_default()
844}
845
846fn approval_error_result(
847 call: &ToolCall,
848 error_code: &str,
849 message: impl Into<String>,
850) -> ToolExecutionResult {
851 let message = message.into();
852 ToolExecutionResult {
853 tool_call_id: call.id.clone(),
854 content: serde_json::json!({
855 "ok": false,
856 "error": message,
857 "error_code": error_code,
858 "tool_name": call.name.clone(),
859 })
860 .to_string(),
861 status: ToolResultStatus::Error,
862 directive: ToolDirective::Continue,
863 error_code: Some(error_code.to_string()),
864 metadata: BTreeMap::new(),
865 image_url: None,
866 image_path: None,
867 }
868}
869
870fn block_on_engine_tool_run<'a>(
871 future: impl std::future::Future<Output = Result<ToolExecutionResult, ToolError>> + 'a,
872) -> Result<ToolExecutionResult, ToolError> {
873 if let Ok(handle) = tokio::runtime::Handle::try_current() {
874 if handle.runtime_flavor() == tokio::runtime::RuntimeFlavor::MultiThread {
875 tokio::task::block_in_place(|| handle.block_on(future))
876 } else {
877 future.now_or_never().unwrap_or_else(|| {
878 Err(ToolError::new(
879 "tool future cannot be driven from a current-thread runtime",
880 ))
881 })
882 }
883 } else {
884 tokio::runtime::Builder::new_current_thread()
885 .enable_all()
886 .build()
887 .map_err(|error| ToolError::new(error.to_string()))?
888 .block_on(future)
889 }
890}