1use std::collections::HashMap;
2use std::fmt::Write;
3use std::sync::Arc;
4
5use async_trait::async_trait;
6use orchestral_core::action::ActionResult;
7#[cfg(test)]
8use orchestral_core::action::{ActionContext, ActionInput};
9use orchestral_core::executor::{
10 execute_action_with_registry_with_options, ActionExecutionOptions, ActionRegistry,
11 AgentStepExecutor, ExecutorContext,
12};
13use orchestral_core::planner::SkillInstruction;
14use orchestral_core::types::Step;
15use serde_json::Value;
16use tokio::sync::RwLock;
17use tracing::{debug, info, warn};
18
19use crate::planner::{LlmClient, LlmRequest, LlmResponse, ToolDefinition};
20
21mod executor;
22mod finalize;
23mod materialize;
24mod parsing;
25mod preflight;
26mod progress;
27mod prompt;
28mod types;
29
30pub use self::executor::{LlmAgentExecutor, LlmAgentExecutorConfig};
31use self::finalize::{attempt_forced_finalization, ForcedFinalizationContext};
32use self::materialize::{
33 derive_evidence_value, materialize_final_exports, validate_agent_action_success,
34};
35#[cfg(test)]
36use self::materialize::{parse_path_segments, render_path_segments};
37use self::parsing::{
38 extract_json, normalize_agent_action_params, parse_agent_decision, parse_agent_params,
39 parse_tool_call_decision,
40};
41#[cfg(test)]
42use self::preflight::action_preflight_error;
43pub(crate) use self::preflight::default_action_preflight_hook;
44use self::preflight::deterministic_action_error_reason;
45use self::progress::report_agent_progress;
46use self::prompt::{
47 agent_tool_definitions, build_agent_system_prompt, build_agent_user_prompt,
48 log_agent_debug_text_payloads, summarize_recent_observations, truncate_log_text, truncate_text,
49};
50#[cfg(test)]
51use self::prompt::{
52 build_bound_inputs_block, collect_agent_debug_text_payloads, leaf_execute_action_schema,
53 summarize_bound_input_value,
54};
55use self::types::{AgentDecision, AgentEvidenceStore, AgentStepParams};
56
57const MAX_PROMPT_OBSERVATION_CHARS: usize = 600;
58const MAX_AGENT_LOG_TEXT_CHARS: usize = 1_200;
59const MAX_BOUND_INPUT_VALUE_CHARS: usize = 1_200;
60const MAX_BOUND_SKILL_VALUE_CHARS: usize = 12_000;
61const MAX_BOUND_INPUT_KEYS: usize = 4;
62const MAX_RECENT_OBSERVATIONS: usize = 4;
63
64#[async_trait]
65impl<C: LlmClient> AgentStepExecutor for LlmAgentExecutor<C> {
66 async fn execute_agent_step(
67 &self,
68 step: &Step,
69 resolved_params: Value,
70 execution_id: &str,
71 ctx: &ExecutorContext,
72 action_registry: Arc<RwLock<ActionRegistry>>,
73 ) -> ActionResult {
74 let params = match parse_agent_params(step.id.as_ref(), &resolved_params) {
75 Ok(v) => v,
76 Err(err) => {
77 warn!(
78 step_id = %step.id,
79 error = %err,
80 "agent step params invalid"
81 );
82 return ActionResult::error(err);
83 }
84 };
85 let mut allowed_actions = params.allowed_actions.iter().cloned().collect::<Vec<_>>();
86 allowed_actions.sort_unstable();
87 info!(
88 step_id = %step.id,
89 execution_id = %execution_id,
90 mode = %params.mode.as_str(),
91 max_iterations = params.max_iterations,
92 output_keys = ?params.output_keys,
93 allowed_actions = ?allowed_actions,
94 "agent step started"
95 );
96
97 let mut observations: Vec<String> = Vec::new();
98 let mut last_success_exports: Option<HashMap<String, Value>> = None;
99 let mut evidence = AgentEvidenceStore::default();
100 let action_execution_options =
101 ActionExecutionOptions::default().with_preflight_hook(default_action_preflight_hook());
102 for iteration in 1..=params.max_iterations {
103 report_agent_progress(
104 ctx,
105 step,
106 "iteration",
107 None,
108 format!("iteration {}/{}", iteration, params.max_iterations),
109 )
110 .await;
111 info!(
112 step_id = %step.id,
113 execution_id = %execution_id,
114 iteration = iteration,
115 max_iterations = params.max_iterations,
116 observation_count = observations.len(),
117 "agent iteration started"
118 );
119 let request = LlmRequest {
120 system: build_agent_system_prompt(¶ms, ctx),
121 user: build_agent_user_prompt(iteration, &observations),
122 model: self.config.model.clone(),
123 temperature: self.config.temperature,
124 };
125 debug!(
126 step_id = %step.id,
127 iteration = iteration,
128 model = %self.config.model,
129 temperature = self.config.temperature,
130 system_prompt = %truncate_log_text(&request.system),
131 user_prompt = %truncate_log_text(&request.user),
132 "agent llm request"
133 );
134
135 let tools = agent_tool_definitions(¶ms);
136 let llm_response = match self.client.complete_with_tools(request, &tools).await {
137 Ok(v) => v,
138 Err(err) => {
139 warn!(
140 step_id = %step.id,
141 iteration = iteration,
142 error = %err,
143 "agent llm call failed"
144 );
145 report_agent_progress(
146 ctx,
147 step,
148 "note",
149 None,
150 format!("llm call failed: {}", truncate_text(&err.to_string())),
151 )
152 .await;
153 return ActionResult::error(format!("agent llm call failed: {}", err));
154 }
155 };
156
157 let decision = match &llm_response {
158 LlmResponse::ToolCall {
159 name, arguments, ..
160 } => {
161 debug!(
162 step_id = %step.id,
163 iteration = iteration,
164 tool_name = %name,
165 tool_args = %truncate_log_text(&arguments.to_string()),
166 "agent llm tool_call response"
167 );
168 log_agent_debug_text_payloads(
169 step.id.as_ref(),
170 iteration,
171 "tool_call_arguments",
172 arguments,
173 );
174 match parse_tool_call_decision(name, arguments.clone()) {
175 Ok(v) => v,
176 Err(err) => {
177 warn!(
178 step_id = %step.id,
179 iteration = iteration,
180 error = %err,
181 tool_name = %name,
182 "agent tool_call parse failed"
183 );
184 observations
185 .push(format!("invalid_tool_call: {}", truncate_text(&err)));
186 continue;
187 }
188 }
189 }
190 LlmResponse::Text(raw) => {
191 debug!(
192 step_id = %step.id,
193 iteration = iteration,
194 llm_output = %truncate_log_text(raw),
195 "agent llm text response (fallback)"
196 );
197 let json = extract_json(raw).unwrap_or_else(|| raw.clone());
198 match parse_agent_decision(&json) {
199 Ok(v) => v,
200 Err(err) => {
201 warn!(
202 step_id = %step.id,
203 iteration = iteration,
204 error = %err,
205 response = %truncate_log_text(&json),
206 "agent decision parse failed"
207 );
208 report_agent_progress(
209 ctx,
210 step,
211 "note",
212 None,
213 format!("invalid agent response: {}", truncate_text(&err)),
214 )
215 .await;
216 observations.push(format!("invalid_decision: {}", truncate_text(&err)));
217 continue;
218 }
219 }
220 }
221 };
222
223 match decision {
224 AgentDecision::Action {
225 name: action_name,
226 params: mut action_params,
227 save_as,
228 capture,
229 } => {
230 info!(
231 step_id = %step.id,
232 iteration = iteration,
233 action = %action_name,
234 "agent decided action"
235 );
236 log_agent_debug_text_payloads(
237 step.id.as_ref(),
238 iteration,
239 "decision_action_params",
240 &action_params,
241 );
242 report_agent_progress(
243 ctx,
244 step,
245 "action_started",
246 Some(action_name.as_str()),
247 format!("run {}", action_name),
248 )
249 .await;
250 if !params.allowed_actions.contains(action_name.as_str()) {
251 warn!(
252 step_id = %step.id,
253 iteration = iteration,
254 action = %action_name,
255 "agent action blocked by allowed_actions"
256 );
257 report_agent_progress(
258 ctx,
259 step,
260 "action_failed",
261 Some(action_name.as_str()),
262 format!("blocked {}", action_name),
263 )
264 .await;
265 observations.push(format!(
266 "blocked_action: '{}' is not in allowed_actions",
267 action_name
268 ));
269 continue;
270 }
271
272 action_params = normalize_agent_action_params(&action_name, action_params);
273
274 let action_result = execute_action_with_registry_with_options(
275 action_registry.clone(),
276 ctx,
277 &step.id,
278 &action_name,
279 &format!("{}-agent-{}", execution_id, iteration),
280 action_params,
281 &action_execution_options,
282 )
283 .await;
284 match action_result {
285 ActionResult::Success { exports } => {
286 let captured_value =
287 match derive_evidence_value(&exports, capture.as_ref()) {
288 Ok(value) => value,
289 Err(err) => {
290 observations.push(format!(
291 "capture_error {} => {}",
292 action_name,
293 truncate_text(&err)
294 ));
295 Value::Object(
296 exports
297 .iter()
298 .map(|(k, v)| (k.clone(), v.clone()))
299 .collect(),
300 )
301 }
302 };
303 if let Err(err) = validate_agent_action_success(
304 ¶ms,
305 &evidence,
306 &action_name,
307 &exports,
308 &captured_value,
309 save_as.as_deref(),
310 ) {
311 let summary = err.summary();
312 warn!(
313 step_id = %step.id,
314 iteration = iteration,
315 action = %action_name,
316 reason = %summary,
317 "agent action produced invalid materialized output"
318 );
319 report_agent_progress(
320 ctx,
321 step,
322 "action_failed",
323 Some(action_name.as_str()),
324 format!(
325 "{} invalid structured output: {}",
326 action_name,
327 truncate_text(&summary)
328 ),
329 )
330 .await;
331 observations.push(format!(
332 "invalid_action_output {} => {}",
333 action_name,
334 truncate_text(&summary)
335 ));
336 continue;
337 }
338 evidence.record_success(
339 &action_name,
340 exports.clone(),
341 captured_value,
342 save_as.as_deref(),
343 );
344 last_success_exports = Some(exports.clone());
345 info!(
346 step_id = %step.id,
347 iteration = iteration,
348 action = %action_name,
349 export_keys = ?exports.keys().collect::<Vec<_>>(),
350 "agent action succeeded"
351 );
352 let status = exports.get("status").and_then(|v| v.as_i64());
353 let progress_line = status
354 .map(|code| format!("{} completed status={}", action_name, code))
355 .unwrap_or_else(|| format!("{} completed", action_name));
356 report_agent_progress(
357 ctx,
358 step,
359 "action_completed",
360 Some(action_name.as_str()),
361 progress_line,
362 )
363 .await;
364 observations.push(format!(
365 "action_success {} => {}",
366 action_name,
367 truncate_text(
368 &Value::Object(
369 exports
370 .iter()
371 .map(|(k, v)| (k.clone(), v.clone()))
372 .collect()
373 )
374 .to_string()
375 )
376 ));
377 if let Some(slot) = save_as.as_deref() {
378 observations
379 .push(format!("saved_slot {} <= {}", slot, action_name));
380 }
381 }
382 ActionResult::NeedClarification { .. }
383 | ActionResult::NeedApproval { .. } => {
384 info!(
385 step_id = %step.id,
386 iteration = iteration,
387 action = %action_name,
388 "agent action requested user interaction"
389 );
390 report_agent_progress(
391 ctx,
392 step,
393 "action_completed",
394 Some(action_name.as_str()),
395 format!("{} requires user input", action_name),
396 )
397 .await;
398 return action_result;
399 }
400 ActionResult::RetryableError { message, .. } => {
401 if let Some(reason) = deterministic_action_error_reason(&message) {
402 warn!(
403 step_id = %step.id,
404 iteration = iteration,
405 action = %action_name,
406 reason = reason,
407 error = %truncate_log_text(&message),
408 "agent deterministic action error detected (fast fail)"
409 );
410 report_agent_progress(
411 ctx,
412 step,
413 "action_failed",
414 Some(action_name.as_str()),
415 format!(
416 "{} deterministic error ({}): {}",
417 action_name,
418 reason,
419 truncate_text(&message)
420 ),
421 )
422 .await;
423 return ActionResult::error(format!(
424 "agent action '{}' deterministic error ({}): {}",
425 action_name, reason, message
426 ));
427 }
428 warn!(
429 step_id = %step.id,
430 iteration = iteration,
431 action = %action_name,
432 error = %truncate_log_text(&message),
433 "agent action retryable error"
434 );
435 report_agent_progress(
436 ctx,
437 step,
438 "action_failed",
439 Some(action_name.as_str()),
440 format!(
441 "{} retryable error: {}",
442 action_name,
443 truncate_text(&message)
444 ),
445 )
446 .await;
447 observations.push(format!(
448 "action_retryable_error {} => {}",
449 action_name,
450 truncate_text(&message)
451 ));
452 }
453 ActionResult::Error { message } => {
454 if let Some(reason) = deterministic_action_error_reason(&message) {
455 warn!(
456 step_id = %step.id,
457 iteration = iteration,
458 action = %action_name,
459 reason = reason,
460 error = %truncate_log_text(&message),
461 "agent deterministic action error detected (fast fail)"
462 );
463 report_agent_progress(
464 ctx,
465 step,
466 "action_failed",
467 Some(action_name.as_str()),
468 format!(
469 "{} deterministic error ({}): {}",
470 action_name,
471 reason,
472 truncate_text(&message)
473 ),
474 )
475 .await;
476 return ActionResult::error(format!(
477 "agent action '{}' deterministic error ({}): {}",
478 action_name, reason, message
479 ));
480 }
481 warn!(
482 step_id = %step.id,
483 iteration = iteration,
484 action = %action_name,
485 error = %truncate_log_text(&message),
486 "agent action failed"
487 );
488 report_agent_progress(
489 ctx,
490 step,
491 "action_failed",
492 Some(action_name.as_str()),
493 format!("{} failed: {}", action_name, truncate_text(&message)),
494 )
495 .await;
496 observations.push(format!(
497 "action_error {} => {}",
498 action_name,
499 truncate_text(&message)
500 ));
501 }
502 }
503 }
504 AgentDecision::Final { exports } => {
505 let filtered = match materialize_final_exports(¶ms, &evidence) {
506 Ok(filtered) => filtered,
507 Err(err) => {
508 warn!(
509 step_id = %step.id,
510 iteration = iteration,
511 missing_output_key = %err.key,
512 reason_code = err.reason_code,
513 reason_detail = %truncate_log_text(&err.detail),
514 available_slots = ?evidence.slots.keys().collect::<Vec<_>>(),
515 available_keys = ?exports.keys().collect::<Vec<_>>(),
516 "agent return_final missing materialized output key"
517 );
518 report_agent_progress(
519 ctx,
520 step,
521 "note",
522 None,
523 format!(
524 "return_final materialize failed '{}' ({}); retrying",
525 err.key, err.reason_code
526 ),
527 )
528 .await;
529 observations.push(format!(
530 "invalid_final: output_key='{}' reason='{}' detail='{}'",
531 err.key,
532 err.reason_code,
533 truncate_text(&err.detail)
534 ));
535 continue;
536 }
537 };
538 info!(
539 step_id = %step.id,
540 iteration = iteration,
541 output_keys = ?params.output_keys,
542 "agent step completed"
543 );
544 report_agent_progress(ctx, step, "note", None, "agent completed").await;
545 return ActionResult::success_with(filtered);
546 }
547 AgentDecision::Finish => {
548 let filtered = match materialize_final_exports(¶ms, &evidence) {
549 Ok(filtered) => filtered,
550 Err(err) => {
551 warn!(
552 step_id = %step.id,
553 iteration = iteration,
554 missing_output_key = %err.key,
555 reason_code = err.reason_code,
556 reason_detail = %truncate_log_text(&err.detail),
557 available_slots = ?evidence.slots.keys().collect::<Vec<_>>(),
558 "agent finish missing materialized output key"
559 );
560 report_agent_progress(
561 ctx,
562 step,
563 "note",
564 None,
565 format!(
566 "finish materialize failed '{}' ({}); retrying",
567 err.key, err.reason_code
568 ),
569 )
570 .await;
571 observations.push(format!(
572 "invalid_finish: output_key='{}' reason='{}' detail='{}'",
573 err.key,
574 err.reason_code,
575 truncate_text(&err.detail)
576 ));
577 continue;
578 }
579 };
580 info!(
581 step_id = %step.id,
582 iteration = iteration,
583 output_keys = ?params.output_keys,
584 "agent step completed via materialized finish"
585 );
586 report_agent_progress(ctx, step, "note", None, "agent completed").await;
587 return ActionResult::success_with(filtered);
588 }
589 }
590 }
591
592 if let Ok(filtered) = materialize_final_exports(¶ms, &evidence) {
593 info!(
594 step_id = %step.id,
595 output_keys = ?params.output_keys,
596 "agent step completed via deterministic materialization"
597 );
598 report_agent_progress(
599 ctx,
600 step,
601 "note",
602 None,
603 "agent completed via deterministic materialization",
604 )
605 .await;
606 return ActionResult::success_with(filtered);
607 }
608
609 let mut forced_finalization_failure: Option<String> = None;
610 if !observations.is_empty() {
611 match attempt_forced_finalization(
612 &self.client,
613 &self.config,
614 ForcedFinalizationContext {
615 step,
616 params: ¶ms,
617 ctx,
618 observations: &observations,
619 last_success_exports: last_success_exports.as_ref(),
620 evidence: &evidence,
621 },
622 )
623 .await
624 {
625 Ok(filtered) => {
626 info!(
627 step_id = %step.id,
628 output_keys = ?params.output_keys,
629 "agent step completed via forced finalization"
630 );
631 report_agent_progress(
632 ctx,
633 step,
634 "note",
635 None,
636 "agent completed via forced finalization",
637 )
638 .await;
639 return ActionResult::success_with(filtered);
640 }
641 Err(err) => {
642 warn!(
643 step_id = %step.id,
644 error = %err,
645 "forced agent finalization failed"
646 );
647 report_agent_progress(
648 ctx,
649 step,
650 "note",
651 None,
652 format!("forced finalization failed: {}", truncate_text(&err)),
653 )
654 .await;
655 forced_finalization_failure = Some(err);
656 }
657 }
658 }
659
660 let observation_tail =
661 summarize_recent_observations(&observations, MAX_RECENT_OBSERVATIONS);
662 warn!(
663 step_id = %step.id,
664 max_iterations = params.max_iterations,
665 recent_observations = %observation_tail,
666 "agent step exhausted max_iterations"
667 );
668 report_agent_progress(
669 ctx,
670 step,
671 "note",
672 None,
673 format!("agent exhausted after {} iterations", params.max_iterations),
674 )
675 .await;
676 let mut error_message = format!(
677 "agent step '{}' reached max_iterations without a valid final result; recent_observations={}",
678 step.id,
679 observation_tail
680 );
681 if let Some(reason) = forced_finalization_failure {
682 let _ = write!(
683 error_message,
684 "; forced_finalization={}",
685 truncate_text(&reason)
686 );
687 }
688 ActionResult::error(error_message)
689 }
690}
691
692#[cfg(test)]
693mod tests;