rho_coding_agent/app/
automation.rs

1use std::{
2    fmt,
3    io::{self, Read, Write},
4    path::PathBuf,
5    sync::Arc,
6};
7
8use rho_sdk::{
9    CapabilityRequest, PolicyDecision, SessionOptions, SystemPrompt, UserInput, Workspace,
10    WorkspacePolicy,
11};
12
13use crate::{
14    cli::Command,
15    config::Config,
16    credentials::OsCredentialStore,
17    diagnostics::RuntimeDiagnostics,
18    herdr::{HerdrReporter, HerdrState},
19    prompt,
20    providers::build_automation_provider,
21    subagent::{self, Preset, RunState, RunStatus},
22    tools::sdk_registry::{AppToolSet, ToolSetOptions},
23};
24
25use super::{
26    runtime_builder::{build_runtime, configured_context_window, RuntimeBuildOptions},
27    sdk_config::SdkBootstrapOptions,
28};
29
30/// Error returned after an automation run handles an interrupt and completes cleanup.
31#[derive(Debug)]
32pub struct AutomationInterrupted {
33    signal: ShutdownSignal,
34}
35
36impl AutomationInterrupted {
37    fn new(signal: ShutdownSignal) -> Self {
38        Self { signal }
39    }
40
41    /// Returns the conventional process exit code for the received signal.
42    pub fn exit_code(&self) -> u8 {
43        match self.signal {
44            ShutdownSignal::Interrupt => 130,
45            ShutdownSignal::Terminate => 143,
46        }
47    }
48}
49
50impl fmt::Display for AutomationInterrupted {
51    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
52        write!(formatter, "rho run interrupted by {}", self.signal)
53    }
54}
55
56impl std::error::Error for AutomationInterrupted {}
57
58#[derive(Clone, Copy, Debug)]
59enum ShutdownSignal {
60    Interrupt,
61    Terminate,
62}
63
64impl fmt::Display for ShutdownSignal {
65    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
66        match self {
67            Self::Interrupt => formatter.write_str("SIGINT"),
68            Self::Terminate => formatter.write_str("SIGTERM"),
69        }
70    }
71}
72
73#[derive(Debug)]
74struct SubagentCancelled;
75
76impl fmt::Display for SubagentCancelled {
77    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
78        formatter.write_str("subagent cancellation requested")
79    }
80}
81
82impl std::error::Error for SubagentCancelled {}
83
84pub(super) struct Startup<'a> {
85    pub config: &'a Config,
86    pub config_path: PathBuf,
87    pub cwd: PathBuf,
88    pub no_system_prompt: bool,
89    pub no_tools: bool,
90    pub no_subagents: bool,
91    pub preset: Option<Preset>,
92    pub output_file: Option<PathBuf>,
93    pub diagnostics: RuntimeDiagnostics,
94    pub herdr: HerdrReporter,
95}
96
97pub(super) fn prompt_for_command(command: &Option<Command>) -> anyhow::Result<Option<String>> {
98    match command {
99        Some(Command::Run { prompt, stdin, .. }) => {
100            prompt_from_stdin(prompt.clone(), *stdin).map(Some)
101        }
102        Some(Command::Login { .. }) | Some(Command::Update) | None => Ok(None),
103    }
104}
105
106pub(super) async fn run(prompt_text: String, startup: Startup<'_>) -> anyhow::Result<()> {
107    // The reporter exists before anything that can fail, so a parent process
108    // watching the output file always sees a terminal state — even when the
109    // run dies during startup (bad auth, broken workspace, ...).
110    let mut reporter = startup
111        .output_file
112        .as_ref()
113        .map(|path| {
114            RunReporter::new(
115                path.clone(),
116                startup.preset.as_ref().map(|preset| preset.name.clone()),
117            )
118        })
119        .transpose()?;
120    let result = run_session(prompt_text, &startup, reporter.as_mut()).await;
121    if let Some(reporter) = reporter.as_mut() {
122        reporter.finish(&result);
123    }
124    let answer = result?;
125    let mut stdout = io::stdout().lock();
126    if reporter.is_some() {
127        // The answer already streamed above and is in the result file.
128        writeln!(stdout, "\n[subagent run complete]")?;
129    } else {
130        writeln!(stdout, "{}", answer.text())?;
131    }
132    stdout.flush()?;
133    Ok(())
134}
135
136async fn run_session(
137    prompt_text: String,
138    startup: &Startup<'_>,
139    reporter: Option<&mut RunReporter>,
140) -> anyhow::Result<rho_sdk::RunOutcome> {
141    let sdk_options = SdkBootstrapOptions::from_config(startup.config, &startup.cwd)?;
142    let credentials = crate::auth::provider_credentials::ApplicationCredentialSource::new(
143        Arc::new(OsCredentialStore),
144    );
145    let provider = build_automation_provider(sdk_options.provider, &credentials)?;
146    let subagents_enabled = startup.config.enable_subagents && !startup.no_subagents;
147    let mut tool_set = if startup.no_tools {
148        AppToolSet::disabled()
149    } else {
150        let subagents = subagents_enabled.then(|| startup.cwd.clone());
151        AppToolSet::new(
152            startup.config,
153            startup.diagnostics.clone(),
154            ToolSetOptions::default()
155                .subagents(subagents)
156                .subagent_config_path(startup.config_path.clone()),
157        )
158    };
159    if let Some(allowed) = startup
160        .preset
161        .as_ref()
162        .and_then(|preset| preset.tools.as_ref())
163    {
164        // The preset's tool list is the subagent's permission boundary:
165        // anything not listed is never registered, so it cannot run.
166        tool_set.retain_named(allowed);
167    }
168    let tool_specs = tool_set.specs();
169    let system_prompt = if startup.no_system_prompt {
170        startup.diagnostics.update_prompt_sources(Vec::new());
171        SystemPrompt::None
172    } else {
173        let system_prompt = prompt::system_prompt(&tool_specs, &startup.cwd);
174        startup
175            .diagnostics
176            .update_prompt_sources(system_prompt.sources);
177        let mut text = system_prompt.text;
178        if !subagents_enabled {
179            prompt::append_subagents_disabled_instruction(&mut text);
180        }
181        if let Some(preset) = &startup.preset {
182            if !preset.prompt.is_empty() {
183                text.push_str("\n\n# Subagent instructions\n\n");
184                text.push_str(&preset.prompt);
185            }
186        }
187        SystemPrompt::Custom(text)
188    };
189    startup.diagnostics.update_tools(&tool_specs);
190
191    let workspace = Workspace::new(&sdk_options.workspace.root)?;
192    let context_window = configured_context_window(startup.config);
193    let compaction = sdk_options.runtime.compaction.clone();
194    startup.diagnostics.update_compaction_config(&compaction);
195    let runtime = build_runtime(RuntimeBuildOptions {
196        provider,
197        tools: tool_set.tools(),
198        workspace,
199        workspace_policy: AutomationWorkspacePolicy,
200        system_prompt,
201        reasoning: sdk_options.runtime.reasoning,
202        compaction,
203        context_window,
204    })?;
205    let session = runtime.session(SessionOptions::default()).await?;
206
207    startup
208        .herdr
209        .report_state(HerdrState::Working, None, None)
210        .await;
211    let result = complete_run(&session, prompt_text, reporter).await;
212
213    runtime.shutdown();
214    tool_set.shutdown().await;
215    startup
216        .herdr
217        .report_state(HerdrState::Idle, None, None)
218        .await;
219    startup.herdr.release().await;
220
221    result
222}
223
224async fn complete_run(
225    session: &rho_sdk::Session,
226    prompt_text: String,
227    reporter: Option<&mut RunReporter>,
228) -> anyhow::Result<rho_sdk::RunOutcome> {
229    let mut run = session.start(UserInput::text(prompt_text)).await?;
230    let cancellation = run.cancellation_handle();
231    let cancel_file = reporter
232        .as_ref()
233        .map(|reporter| reporter.cancel_file.clone());
234    tokio::select! {
235        outcome = drive_headless_run(&mut run, reporter) => outcome,
236        signal = shutdown_signal() => {
237            let signal = signal?;
238            cancellation.cancel();
239            let _ = run.outcome().await;
240            Err(AutomationInterrupted::new(signal).into())
241        }
242        cancelled = wait_for_cancel_request(cancel_file) => {
243            cancelled?;
244            cancellation.cancel();
245            let _ = run.outcome().await;
246            Err(SubagentCancelled.into())
247        }
248    }
249}
250
251/// Drains run events with no interactive host attached.
252///
253/// Host input requests cannot be answered headlessly; cancel instead of
254/// leaving the requesting tool suspended until a signal arrives.
255async fn drive_headless_run(
256    run: &mut rho_sdk::Run,
257    mut reporter: Option<&mut RunReporter>,
258) -> anyhow::Result<rho_sdk::RunOutcome> {
259    let mut heartbeat = tokio::time::interval(REPORT_HEARTBEAT);
260    heartbeat.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
261    loop {
262        let event = tokio::select! {
263            event = run.next_event() => event,
264            _ = heartbeat.tick(), if reporter.is_some() => {
265                if let Some(reporter) = reporter.as_deref_mut() {
266                    reporter.write();
267                }
268                continue;
269            }
270        };
271        let Some(event) = event else {
272            break;
273        };
274        if let Some(reporter) = reporter.as_deref_mut() {
275            reporter.on_event(&event);
276        }
277        if let rho_sdk::RunEvent::HostInputRequested { request } = event {
278            run.cancel();
279            let _ = run.outcome().await;
280            anyhow::bail!(
281                "rho run cannot answer host input request '{}' ({}); run without tools that require interactive input",
282                request.id(),
283                request.title(),
284            );
285        }
286    }
287    Ok(run.outcome().await?)
288}
289
290/// Maintains the `--output-file` status contract for subagent runs and
291/// streams progress to stdout so a watching pane shows live activity.
292struct RunReporter {
293    path: PathBuf,
294    cancel_file: PathBuf,
295    status: RunStatus,
296    last_write: std::time::Instant,
297}
298
299/// Longest a status-file write is deferred while text streams.
300const REPORT_THROTTLE: std::time::Duration = std::time::Duration::from_secs(2);
301/// Keeps the status file fresh while a provider or tool call emits no events.
302const REPORT_HEARTBEAT: std::time::Duration = std::time::Duration::from_secs(10);
303const LAST_TEXT_BYTES: usize = 400;
304
305impl RunReporter {
306    fn new(path: PathBuf, preset: Option<String>) -> anyhow::Result<Self> {
307        let cancel_file = subagent::cancel_file_for(&path);
308        match std::fs::remove_file(&cancel_file) {
309            Ok(()) => {}
310            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
311            Err(error) => return Err(error.into()),
312        }
313        let status = RunStatus {
314            state: RunState::Starting,
315            pid: Some(std::process::id()),
316            preset,
317            ..RunStatus::default()
318        };
319        subagent::write_status(&path, &status)?;
320        Ok(Self {
321            path,
322            cancel_file,
323            status,
324            last_write: std::time::Instant::now(),
325        })
326    }
327
328    fn on_event(&mut self, event: &rho_sdk::RunEvent) {
329        use rho_sdk::RunEvent;
330
331        match event {
332            RunEvent::StepStarted { step } => {
333                self.status.state = RunState::Running;
334                self.status.turns = *step as u64;
335                self.write();
336            }
337            RunEvent::ToolStarted { name, .. } => {
338                self.status.last_activity = Some(format!("tool: {name}"));
339                self.stream(&format!("\n[tool] {name}\n"));
340                self.write();
341            }
342            RunEvent::AssistantTextDelta { text } => {
343                self.status.last_activity = Some("assistant text".into());
344                append_tail(
345                    self.status.last_text.get_or_insert_with(String::new),
346                    text,
347                    LAST_TEXT_BYTES,
348                );
349                self.stream(text);
350                self.write_throttled();
351            }
352            RunEvent::UsageUpdated { usage } => {
353                self.status.input_tokens = usage.total_input_tokens().unwrap_or(0);
354                self.status.output_tokens = usage.output_tokens.unwrap_or(0);
355            }
356            _ => {}
357        }
358    }
359
360    fn finish(&mut self, result: &anyhow::Result<rho_sdk::RunOutcome>) {
361        match result {
362            Ok(outcome) => {
363                self.status.state = RunState::Ok;
364                self.status.result = Some(outcome.text().to_string());
365                let usage = outcome.usage();
366                self.status.input_tokens = usage.total_input_tokens().unwrap_or(0);
367                self.status.output_tokens = usage.output_tokens.unwrap_or(0);
368            }
369            Err(error)
370                if error.is::<AutomationInterrupted>() || error.is::<SubagentCancelled>() =>
371            {
372                self.status.state = RunState::Stopped;
373                self.status.result = self
374                    .status
375                    .last_text
376                    .as_ref()
377                    .map(|text| format!("(partial, stopped before finishing)\n{text}"));
378            }
379            Err(error) => {
380                self.status.state = RunState::Error;
381                self.status.error = Some(format!("{error:#}"));
382            }
383        }
384        self.write();
385    }
386
387    fn stream(&self, text: &str) {
388        let mut stdout = io::stdout().lock();
389        let _ = stdout.write_all(text.as_bytes());
390        let _ = stdout.flush();
391    }
392
393    fn write_throttled(&mut self) {
394        if self.last_write.elapsed() >= REPORT_THROTTLE {
395            self.write();
396        }
397    }
398
399    fn write(&mut self) {
400        self.last_write = std::time::Instant::now();
401        let _ = subagent::write_status(&self.path, &self.status);
402    }
403}
404
405const CANCEL_POLL_INTERVAL: std::time::Duration = std::time::Duration::from_millis(100);
406
407async fn wait_for_cancel_request(cancel_file: Option<PathBuf>) -> io::Result<()> {
408    let Some(cancel_file) = cancel_file else {
409        return std::future::pending().await;
410    };
411    loop {
412        match tokio::fs::metadata(&cancel_file).await {
413            Ok(_) => return Ok(()),
414            Err(error) if error.kind() == io::ErrorKind::NotFound => {}
415            Err(error) => return Err(error),
416        }
417        tokio::time::sleep(CANCEL_POLL_INTERVAL).await;
418    }
419}
420
421/// Appends to a rolling tail buffer capped at `max` bytes.
422fn append_tail(buffer: &mut String, text: &str, max: usize) {
423    buffer.push_str(text);
424    if buffer.len() > max {
425        let cut = buffer.len() - max;
426        let boundary = (cut..buffer.len())
427            .find(|index| buffer.is_char_boundary(*index))
428            .unwrap_or(buffer.len());
429        buffer.drain(..boundary);
430    }
431}
432
433#[cfg(unix)]
434async fn shutdown_signal() -> io::Result<ShutdownSignal> {
435    use tokio::signal::unix::{signal, SignalKind};
436
437    let mut interrupt = signal(SignalKind::interrupt())?;
438    let mut terminate = signal(SignalKind::terminate())?;
439    tokio::select! {
440        _ = interrupt.recv() => Ok(ShutdownSignal::Interrupt),
441        _ = terminate.recv() => Ok(ShutdownSignal::Terminate),
442    }
443}
444
445#[cfg(not(unix))]
446async fn shutdown_signal() -> io::Result<ShutdownSignal> {
447    tokio::signal::ctrl_c().await?;
448    Ok(ShutdownSignal::Interrupt)
449}
450
451#[derive(Clone, Copy, Debug)]
452struct AutomationWorkspacePolicy;
453
454impl WorkspacePolicy for AutomationWorkspacePolicy {
455    fn evaluate(&self, _request: &CapabilityRequest) -> PolicyDecision {
456        PolicyDecision::Allow
457    }
458}
459
460fn prompt_from_stdin(parts: Vec<String>, read_stdin: bool) -> anyhow::Result<String> {
461    prompt_from_reader(parts, read_stdin, &mut io::stdin())
462}
463
464fn prompt_from_reader(
465    parts: Vec<String>,
466    read_stdin: bool,
467    stdin: &mut impl Read,
468) -> anyhow::Result<String> {
469    let mut chunks = Vec::new();
470    let inline = parts.join(" ").trim().to_string();
471    if !inline.is_empty() {
472        chunks.push(inline);
473    }
474    if read_stdin {
475        let mut buffer = String::new();
476        stdin.read_to_string(&mut buffer)?;
477        let buffer = buffer.trim().to_string();
478        if !buffer.is_empty() {
479            chunks.push(buffer);
480        }
481    }
482
483    let prompt = chunks.join("\n\n");
484    if prompt.is_empty() {
485        anyhow::bail!("rho run requires a prompt argument or --stdin");
486    }
487    Ok(prompt)
488}
489
490#[cfg(test)]
491#[path = "automation_tests.rs"]
492mod tests;
rho_coding_agent/app/automation.rs

rho_coding_agent/app/
automation.rs
