ras_agent/application/
run_step.rs

1use std::sync::Arc;
2use std::time::Instant;
3
4use chrono::Utc;
5use ras_cdp::BrowserPort;
6use ras_dom::DomExtractor;
7use ras_errors::AppError;
8use ras_events::EventBus;
9use ras_llm::{ChatMessage, ChatResponse, InvokeOptions, LlmClient};
10use ras_tools::domain::registry::{ActionRegistry, ToolContext};
11use ras_types::{ActionResult, StepId};
12use url::Url;
13
14use crate::application::compute_action_hash::compute_action_hash;
15use crate::application::detect_loop::{build_budget_warning, build_loop_nudge};
16use crate::application::fallback_llm::should_switch_to_fallback;
17use crate::application::parse_output::parse_agent_output;
18use crate::domain::agent_history::StepRecord;
19use crate::domain::loop_detector::ActionLoopDetector;
20use crate::domain::step_metadata::StepMetadata;
21
22pub struct RunStep {
23    primary_llm: Arc<dyn LlmClient>,
24    fallback_llm: Option<Arc<dyn LlmClient>>,
25    registry: Arc<ActionRegistry>,
26    browser: Arc<dyn BrowserPort>,
27    events: Arc<dyn EventBus>,
28    dom_extractor: Option<Arc<dyn DomExtractor>>,
29}
30
31impl RunStep {
32    #[must_use]
33    pub fn new(
34        primary: Arc<dyn LlmClient>,
35        fallback: Option<Arc<dyn LlmClient>>,
36        registry: Arc<ActionRegistry>,
37        browser: Arc<dyn BrowserPort>,
38        events: Arc<dyn EventBus>,
39        dom_extractor: Option<Arc<dyn DomExtractor>>,
40    ) -> Self {
41        Self {
42            primary_llm: primary,
43            fallback_llm: fallback,
44            registry,
45            browser,
46            events,
47            dom_extractor,
48        }
49    }
50
51    pub async fn execute(
52        &self,
53        step: StepId,
54        max_steps: u32,
55        prompt: Vec<ChatMessage>,
56        detector: &mut ActionLoopDetector,
57    ) -> Result<StepRecord, AppError> {
58        let started = Instant::now();
59        let mut messages = prompt;
60        if let Some(nudge) = build_loop_nudge(detector) {
61            messages.push(nudge);
62        }
63        if let Some(warn) = build_budget_warning(step.0, max_steps) {
64            messages.push(warn);
65        }
66        let response = self.invoke_with_fallback(messages).await?;
67        let output = parse_agent_output(&response)?;
68
69        let target = self.browser.focused_target().await.ok();
70        let page_url = match &target {
71            Some(t) => self
72                .browser
73                .evaluate(t, "location.href")
74                .await
75                .ok()
76                .and_then(|v| v.as_str().and_then(|s| Url::parse(s).ok())),
77            None => None,
78        };
79
80        let mut results = Vec::new();
81        for action in &output.action {
82            detector.record_action(compute_action_hash(action));
83            let Some(reg) = self.registry.get(&action.name) else {
84                results.push(ActionResult::err(format!(
85                    "unknown action: {}",
86                    action.name.0
87                )));
88                break;
89            };
90            let ctx = ToolContext {
91                browser: self.browser.clone(),
92                events: self.events.clone(),
93                page_url: page_url.clone(),
94                available_files: Vec::new(),
95            };
96            match reg.handler.execute(action.parameters.clone(), ctx).await {
97                Ok(r) => {
98                    let terminates = reg.metadata.terminates_sequence;
99                    let is_done = r.is_done;
100                    let is_err = r.is_error();
101                    results.push(r);
102                    if terminates || is_done || is_err {
103                        break;
104                    }
105                }
106                Err(e) => {
107                    results.push(ActionResult::err(e.to_string()));
108                    break;
109                }
110            }
111        }
112
113        let summary = match (&self.dom_extractor, &target) {
114            (Some(extractor), Some(t)) => match extractor.snapshot(t).await {
115                Ok(s) => Some(s),
116                Err(e) => {
117                    tracing::warn!(error = %e, "dom snapshot failed; continuing without grounding");
118                    None
119                }
120            },
121            _ => None,
122        };
123
124        let metadata = StepMetadata {
125            duration_ms: started.elapsed().as_millis() as u64,
126            step_interval_ms: None,
127            usage: response.usage,
128            model: Some(response.model.clone()),
129            fallback_used: false,
130        };
131        Ok(StepRecord {
132            step,
133            started_at: Utc::now(),
134            url: page_url,
135            output,
136            results,
137            metadata,
138            summary,
139        })
140    }
141
142    async fn invoke_with_fallback(
143        &self,
144        messages: Vec<ChatMessage>,
145    ) -> Result<ChatResponse, AppError> {
146        let opts = InvokeOptions::default();
147        match self
148            .primary_llm
149            .ainvoke(messages.clone(), opts.clone())
150            .await
151        {
152            Ok(r) => Ok(r),
153            Err(e) if should_switch_to_fallback(&e) => match &self.fallback_llm {
154                Some(fb) => fb.ainvoke(messages, opts).await,
155                None => Err(e),
156            },
157            Err(e) => Err(e),
158        }
159    }
160}
161
162#[must_use]
163pub fn done_result(text: impl Into<String>) -> ActionResult {
164    ActionResult::done(text)
165}
ras_agent/application/run_step.rs

ras_agent/application/
run_step.rs
