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