rho_coding_agent/app/
automation.rs

1use std::{
2    fmt,
3    io::{self, Read, Write},
4    num::NonZeroUsize,
5    path::PathBuf,
6    sync::Arc,
7    time::Duration,
8};
9
10use rho_sdk::{SessionOptions, SystemPrompt, UserInput};
11
12use {
13    crate::agent::{PromptPolicy, ToolCapability},
14    crate::cli::{Command, OutputFormat},
15    crate::config::Config,
16    crate::credential_store::AppCredentialStore,
17    crate::diagnostics::RuntimeDiagnostics,
18    crate::herdr::{HerdrReporter, HerdrState},
19    crate::prompt,
20    crate::subagent::{RunState, RunStatus},
21    crate::tools::{
22        agent::BackgroundSubagents,
23        sdk_registry::{AppToolSet, DelegationConfig, ToolSetOptions},
24    },
25    rho_providers::providers::build_automation_provider,
26};
27
28use super::{
29    agent_binding::BoundAgent,
30    automation_protocol::{write_event, JsonlAdapter, TerminalReason, WireEvent},
31    headless_run::{self, HeadlessRunDeps, HostInputResponder},
32    policy::AppPolicy,
33    runtime_builder::{
34        build_runtime_with_max_steps, configured_context_window, RuntimeBuildOptions,
35    },
36    sdk_config::SdkBootstrapOptions,
37};
38
39/// Error returned after an automation run has cleaned up and selected a stable exit code.
40#[derive(Debug)]
41pub struct AutomationExit {
42    code: u8,
43    reason: TerminalReason,
44    message: String,
45}
46
47impl AutomationExit {
48    pub(super) fn new(code: u8, reason: TerminalReason, message: impl Into<String>) -> Self {
49        Self {
50            code,
51            reason,
52            message: message.into(),
53        }
54    }
55
56    /// Returns the documented process exit code for this automation result.
57    pub fn exit_code(&self) -> u8 {
58        self.code
59    }
60
61    fn reason(&self) -> TerminalReason {
62        self.reason
63    }
64}
65
66impl fmt::Display for AutomationExit {
67    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
68        formatter.write_str(&self.message)
69    }
70}
71
72impl std::error::Error for AutomationExit {}
73
74/// Error returned after an automation run handles an interrupt and completes cleanup.
75#[derive(Debug)]
76pub struct AutomationInterrupted {
77    signal: ShutdownSignal,
78}
79
80impl AutomationInterrupted {
81    fn new(signal: ShutdownSignal) -> Self {
82        Self { signal }
83    }
84
85    /// Returns the conventional process exit code for the received signal.
86    pub fn exit_code(&self) -> u8 {
87        self.signal.exit_code()
88    }
89}
90
91impl fmt::Display for AutomationInterrupted {
92    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
93        write!(formatter, "rho run interrupted by {}", self.signal)
94    }
95}
96
97impl std::error::Error for AutomationInterrupted {}
98
99#[derive(Clone, Copy, Debug)]
100enum ShutdownSignal {
101    Interrupt,
102    Terminate,
103}
104
105impl ShutdownSignal {
106    fn exit_code(self) -> u8 {
107        match self {
108            Self::Interrupt => 130,
109            Self::Terminate => 143,
110        }
111    }
112}
113
114impl fmt::Display for ShutdownSignal {
115    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
116        match self {
117            Self::Interrupt => formatter.write_str("SIGINT"),
118            Self::Terminate => formatter.write_str("SIGTERM"),
119        }
120    }
121}
122
123#[derive(Debug)]
124struct SubagentCancelled;
125
126impl fmt::Display for SubagentCancelled {
127    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
128        formatter.write_str("subagent cancellation requested")
129    }
130}
131
132impl std::error::Error for SubagentCancelled {}
133
134pub(super) struct Startup<'a> {
135    pub config: &'a Config,
136    pub config_path: PathBuf,
137    pub cwd: PathBuf,
138    pub no_system_prompt: bool,
139    pub no_tools: bool,
140    pub no_subagents: bool,
141    pub usage_purpose: &'static str,
142    pub parent_session_id: Option<rho_sdk::SessionId>,
143    pub agent: BoundAgent,
144    pub output_file: Option<PathBuf>,
145    pub output: OutputFormat,
146    pub max_steps: Option<NonZeroUsize>,
147    pub timeout: Option<Duration>,
148    pub diagnostics: RuntimeDiagnostics,
149    pub herdr: HerdrReporter,
150    pub host_input: Option<Arc<dyn HostInputResponder>>,
151}
152
153pub(super) fn prompt_for_command(command: &Option<Command>) -> anyhow::Result<Option<String>> {
154    match command {
155        Some(Command::Run { prompt, stdin, .. }) => {
156            prompt_from_stdin(prompt.clone(), *stdin).map(Some)
157        }
158        Some(
159            Command::Attach { .. }
160            | Command::Login { .. }
161            | Command::CredentialStore { .. }
162            | Command::Update,
163        )
164        | None => Ok(None),
165    }
166}
167
168pub(super) fn emit_startup_failure() -> anyhow::Result<()> {
169    let mut adapter = JsonlAdapter::new();
170    let event = adapter.failed(
171        TerminalReason::ConfigurationError,
172        "configuration failed".into(),
173        None,
174    );
175    emit(event)
176}
177
178pub(super) async fn run(prompt_text: String, startup: Startup<'_>) -> anyhow::Result<()> {
179    let mut jsonl = (startup.output == OutputFormat::Jsonl).then(JsonlAdapter::new);
180    let deadline = startup
181        .timeout
182        .map(|timeout| tokio::time::Instant::now() + timeout);
183    // The reporter exists before anything that can fail, so a parent process
184    // watching the output file always sees a terminal state, including startup failures.
185    let reporter_result = startup
186        .output_file
187        .as_ref()
188        .map(|path| {
189            RunReporter::new(
190                path.clone(),
191                RunArtifactIdentity {
192                    agent_id: startup.agent.id().to_string(),
193                    agent_fingerprint: startup.agent.fingerprint().to_string(),
194                    provider: startup.config.provider.clone(),
195                    model: startup.config.model.clone(),
196                },
197                startup.cwd.clone(),
198                &prompt_text,
199                /* stream_output */ startup.output == OutputFormat::Text,
200                None,
201            )
202        })
203        .transpose();
204    let mut reporter = match reporter_result {
205        Ok(reporter) => reporter,
206        Err(error) => {
207            emit_failure(&mut jsonl, TerminalReason::OutputError, &error)?;
208            return Err(
209                AutomationExit::new(1, TerminalReason::OutputError, error.to_string()).into(),
210            );
211        }
212    };
213
214    let cancellation = rho_tools::cancellation::RunCancellation::default();
215    let (result, timed_out) = if let Some(deadline) = deadline {
216        let future = run_session_with_output(
217            prompt_text,
218            &startup,
219            reporter.as_mut(),
220            Some(cancellation.clone()),
221            jsonl.as_mut(),
222        );
223        tokio::pin!(future);
224        tokio::select! {
225            result = &mut future => (result, false),
226            () = tokio::time::sleep_until(deadline) => {
227                cancellation.cancel();
228                (future.await, true)
229            }
230        }
231    } else {
232        (
233            run_session_with_output(
234                prompt_text,
235                &startup,
236                reporter.as_mut(),
237                None,
238                jsonl.as_mut(),
239            )
240            .await,
241            false,
242        )
243    };
244    if let Some(reporter) = reporter.as_mut() {
245        let reached_step_limit = result.as_ref().is_ok_and(|outcome| {
246            outcome.stop_reason() == rho_sdk::StopReason::MaxSteps
247                && (jsonl.is_some() || startup.max_steps.is_some())
248        });
249        if reached_step_limit {
250            let stopped = Err(AutomationExit::new(
251                124,
252                TerminalReason::MaxSteps,
253                "rho run reached its model-step limit",
254            )
255            .into());
256            reporter.finish(&stopped);
257        } else {
258            reporter.finish(&result);
259        }
260    }
261
262    if timed_out {
263        emit_stopped(&mut jsonl, TerminalReason::Timeout)?;
264        return Err(AutomationExit::new(124, TerminalReason::Timeout, "rho run timed out").into());
265    }
266
267    match result {
268        Ok(answer) => {
269            let max_steps = answer.stop_reason() == rho_sdk::StopReason::MaxSteps;
270            if max_steps && (jsonl.is_some() || startup.max_steps.is_some()) {
271                if let Some(adapter) = jsonl.as_mut() {
272                    let text = (!answer.text().is_empty()).then(|| answer.text().into());
273                    let event = adapter.stopped(TerminalReason::MaxSteps, text);
274                    emit(event)?;
275                } else {
276                    write_text_answer(&answer, reporter.is_some())?;
277                }
278                return Err(AutomationExit::new(
279                    124,
280                    TerminalReason::MaxSteps,
281                    "rho run reached its model-step limit",
282                )
283                .into());
284            }
285            if let Some(adapter) = jsonl.as_mut() {
286                let event = adapter.completed(answer.text().into());
287                emit(event)?;
288            } else {
289                write_text_answer(&answer, reporter.is_some())?;
290            }
291            Ok(())
292        }
293        Err(error) => {
294            let (reason, code) = classify_error(&error);
295            if reason == TerminalReason::Interrupted {
296                emit_stopped(&mut jsonl, reason)?;
297            } else if reason != TerminalReason::OutputError {
298                emit_failure(&mut jsonl, reason, &error)?;
299            }
300            let message = terminal_error_message(reason, &error);
301            if error.is::<AutomationInterrupted>() {
302                return Err(error);
303            }
304            Err(AutomationExit::new(code, reason, message).into())
305        }
306    }
307}
308
309fn write_text_answer(answer: &rho_sdk::RunOutcome, has_reporter: bool) -> anyhow::Result<()> {
310    let result = (|| -> io::Result<()> {
311        let mut stdout = io::stdout().lock();
312        if has_reporter {
313            writeln!(stdout, "\n[subagent run complete]")?;
314        } else {
315            writeln!(stdout, "{}", answer.text())?;
316        }
317        stdout.flush()
318    })();
319    result.map_err(|error| {
320        AutomationExit::new(
321            1,
322            TerminalReason::OutputError,
323            format!("could not write output: {error}"),
324        )
325        .into()
326    })
327}
328
329pub(super) fn emit(event: WireEvent) -> anyhow::Result<()> {
330    let mut stdout = io::stdout().lock();
331    write_event(&mut stdout, &event).map_err(|error| {
332        AutomationExit::new(
333            1,
334            TerminalReason::OutputError,
335            format!("could not write JSONL output: {error}"),
336        )
337        .into()
338    })
339}
340
341fn emit_stopped(adapter: &mut Option<JsonlAdapter>, reason: TerminalReason) -> anyhow::Result<()> {
342    if let Some(adapter) = adapter.as_mut() {
343        let text = adapter.partial_text();
344        let event = adapter.stopped(reason, text);
345        emit(event)?;
346    }
347    Ok(())
348}
349
350fn emit_failure(
351    adapter: &mut Option<JsonlAdapter>,
352    reason: TerminalReason,
353    error: &anyhow::Error,
354) -> anyhow::Result<()> {
355    if let Some(adapter) = adapter.as_mut() {
356        let text = adapter.partial_text();
357        let message = terminal_error_message(reason, error);
358        let event = adapter.failed(reason, message, text);
359        emit(event)?;
360    }
361    Ok(())
362}
363
364fn terminal_error_message(reason: TerminalReason, error: &anyhow::Error) -> String {
365    match reason {
366        TerminalReason::Authentication => "authentication failed".to_string(),
367        TerminalReason::ConfigurationError => "configuration failed".to_string(),
368        TerminalReason::OutputError => "output failed".to_string(),
369        TerminalReason::OtherError => "run failed".to_string(),
370        _ => error.to_string(),
371    }
372}
373
374fn classify_error(error: &anyhow::Error) -> (TerminalReason, u8) {
375    if let Some(interrupted) = error.downcast_ref::<AutomationInterrupted>() {
376        return (TerminalReason::Interrupted, interrupted.exit_code());
377    }
378    if let Some(exit) = error.downcast_ref::<AutomationExit>() {
379        return (exit.reason(), exit.exit_code());
380    }
381    for cause in error.chain() {
382        if let Some(error) = cause.downcast_ref::<rho_sdk::Error>() {
383            return match error {
384                rho_sdk::Error::Authentication { .. } => (TerminalReason::Authentication, 1),
385                rho_sdk::Error::Provider(provider)
386                    if provider.kind() == rho_sdk::ProviderErrorKind::Authentication =>
387                {
388                    (TerminalReason::Authentication, 1)
389                }
390                rho_sdk::Error::Provider(_) => (TerminalReason::ProviderError, 1),
391                rho_sdk::Error::Tool(_) => (TerminalReason::ToolHostError, 1),
392                rho_sdk::Error::InvalidConfiguration { .. } => {
393                    (TerminalReason::ConfigurationError, 2)
394                }
395                _ => (TerminalReason::OtherError, 1),
396            };
397        }
398        if let Some(error) = cause.downcast_ref::<rho_providers::model::ModelError>() {
399            use rho_providers::model::ModelError;
400            return match error {
401                ModelError::MissingApiKey
402                | ModelError::MissingCodexAuth
403                | ModelError::MissingAnthropicApiKey
404                | ModelError::MissingGoogleApiKey
405                | ModelError::MissingGithubCopilotAuth
406                | ModelError::MissingMoonshotApiKey
407                | ModelError::MissingPoolsideApiKey
408                | ModelError::MissingOpenRouterApiKey
409                | ModelError::MissingCredentialProfile(_)
410                | ModelError::MissingKimiAuth
411                | ModelError::MissingXaiApiKey
412                | ModelError::MissingXaiAuth
413                | ModelError::Credentials(_) => (TerminalReason::Authentication, 1),
414                ModelError::UnsupportedReasoning { .. } | ModelError::UnsupportedProvider(_) => {
415                    (TerminalReason::ConfigurationError, 2)
416                }
417                _ => (TerminalReason::ProviderError, 1),
418            };
419        }
420    }
421    (TerminalReason::OtherError, 1)
422}
423
424pub(crate) async fn run_session(
425    prompt_text: String,
426    startup: &Startup<'_>,
427    reporter: Option<&mut RunReporter>,
428    cancellation: Option<rho_tools::cancellation::RunCancellation>,
429) -> anyhow::Result<rho_sdk::RunOutcome> {
430    run_session_with_output(prompt_text, startup, reporter, cancellation, None).await
431}
432
433async fn run_session_with_output(
434    prompt_text: String,
435    startup: &Startup<'_>,
436    reporter: Option<&mut RunReporter>,
437    cancellation: Option<rho_tools::cancellation::RunCancellation>,
438    mut jsonl: Option<&mut JsonlAdapter>,
439) -> anyhow::Result<rho_sdk::RunOutcome> {
440    let sdk_options = SdkBootstrapOptions::from_config(startup.config, &startup.cwd)?;
441    let credentials = rho_providers::auth::provider_credentials::ApplicationCredentialSource::new(
442        Arc::new(AppCredentialStore),
443    );
444    let provider = build_automation_provider(sdk_options.provider, &credentials)?;
445    let mut capabilities = startup
446        .agent
447        .rho_capabilities()
448        .cloned()
449        .unwrap_or_default();
450    if startup.no_subagents {
451        capabilities.remove(&ToolCapability::Agent);
452        capabilities.remove(&ToolCapability::Agents);
453    }
454    let launch_delegation_enabled = capabilities.contains(&ToolCapability::Agent);
455    let delegation_enabled =
456        launch_delegation_enabled || capabilities.contains(&ToolCapability::Agents);
457    let tool_set = if startup.no_tools {
458        AppToolSet::disabled()
459    } else {
460        let mut options = ToolSetOptions::new(capabilities);
461        if delegation_enabled {
462            options = options.delegation(DelegationConfig::new(
463                startup.cwd.clone(),
464                startup.config_path.clone(),
465                BackgroundSubagents::Disabled,
466            ));
467        }
468        AppToolSet::new(startup.config, startup.diagnostics.clone(), options)
469    };
470    let tool_specs = tool_set.specs();
471    let system_prompt = if startup.no_system_prompt {
472        startup.diagnostics.update_prompt_sources(Vec::new());
473        SystemPrompt::None
474    } else {
475        let mut text = match startup.agent.prompt() {
476            PromptPolicy::Replace(text) => text.clone(),
477            PromptPolicy::Extend(extra) => {
478                let built = prompt::system_prompt(&tool_specs, &startup.cwd);
479                startup.diagnostics.update_prompt_sources(built.sources);
480                let mut text = built.text;
481                if !launch_delegation_enabled {
482                    prompt::append_subagents_disabled_instruction(&mut text);
483                }
484                if !extra.is_empty() {
485                    text.push_str("\n\n# Agent instructions\n\n");
486                    text.push_str(extra);
487                }
488                text
489            }
490        };
491        if text.is_empty() {
492            text = "You are a coding agent.".into();
493        }
494        SystemPrompt::Custom(text)
495    };
496    startup.diagnostics.update_tools(&tool_specs);
497
498    let workspace_root = sdk_options.workspace.root.clone();
499    let workspace = sdk_options.workspace.build_workspace()?;
500    let context_window = configured_context_window(startup.config);
501    let compaction = sdk_options.runtime.compaction.clone();
502    startup.diagnostics.update_compaction_config(&compaction);
503    let usage_recording = crate::usage::default_recording().await;
504    let runtime = build_runtime_with_max_steps(
505        RuntimeBuildOptions {
506            provider,
507            tools: tool_set.tools(),
508            workspace,
509            workspace_policy: AppPolicy::for_mode(startup.config.permission_mode),
510            approval_handler: None,
511            system_prompt,
512            reasoning: sdk_options.runtime.reasoning,
513            compaction,
514            context_window,
515            usage_purpose: startup.usage_purpose,
516            usage_parent_session_id: startup.parent_session_id.clone(),
517            usage_recording,
518        },
519        startup.max_steps,
520    )?;
521    let session = runtime.session(SessionOptions::default()).await?;
522    if let Some(adapter) = jsonl.as_deref_mut() {
523        adapter.set_run_context(session.id(), &workspace_root);
524    }
525    if let Some(manager) = tool_set.subagents() {
526        manager.set_session(session.id().to_string());
527    }
528
529    startup
530        .herdr
531        .report_state(HerdrState::Working, None, None)
532        .await;
533    let result = complete_run(
534        &session,
535        prompt_text,
536        HeadlessRunDeps {
537            reporter,
538            external_cancellation: cancellation,
539            jsonl,
540            host_input: startup.host_input.as_deref(),
541        },
542    )
543    .await;
544
545    runtime.shutdown();
546    tool_set.shutdown().await;
547    startup
548        .herdr
549        .report_state(HerdrState::Idle, None, None)
550        .await;
551    startup.herdr.release().await;
552
553    result
554}
555
556async fn complete_run(
557    session: &rho_sdk::Session,
558    prompt_text: String,
559    dependencies: HeadlessRunDeps<'_>,
560) -> anyhow::Result<rho_sdk::RunOutcome> {
561    let HeadlessRunDeps {
562        reporter,
563        external_cancellation,
564        jsonl,
565        host_input,
566    } = dependencies;
567    let mut run = session.start(UserInput::text(prompt_text)).await?;
568    let cancellation = run.cancellation_handle();
569    let external_cancellation = external_cancellation.unwrap_or_default();
570    tokio::select! {
571        outcome = headless_run::drive(&mut run, reporter, jsonl, host_input) => outcome,
572        signal = shutdown_signal() => {
573            let signal = signal?;
574            cancellation.cancel();
575            let _ = run.outcome().await;
576            Err(AutomationInterrupted::new(signal).into())
577        }
578        () = external_cancellation.cancelled() => {
579            cancellation.cancel();
580            let _ = run.outcome().await;
581            Err(SubagentCancelled.into())
582        }
583    }
584}
585
586pub(crate) use crate::run_artifacts::RunArtifactIdentity;
587
588/// Maintains the `--output-file` status contract for subagent runs and
589/// streams progress to stdout so a watching pane shows live activity.
590pub(crate) struct RunReporter {
591    sink: crate::run_artifacts::RunArtifactSink,
592    adapter: crate::tui::event_adapter::SdkEventAdapter,
593    stream_output: bool,
594}
595
596impl RunReporter {
597    pub(crate) fn new(
598        path: PathBuf,
599        identity: RunArtifactIdentity,
600        cwd: PathBuf,
601        prompt: &str,
602        stream_output: bool,
603        status_tx: Option<tokio::sync::watch::Sender<RunStatus>>,
604    ) -> anyhow::Result<Self> {
605        let sink = crate::run_artifacts::RunArtifactSink::open(path, &identity, prompt, status_tx)?;
606        Ok(Self {
607            sink,
608            adapter: crate::tui::event_adapter::SdkEventAdapter::new(cwd),
609            stream_output,
610        })
611    }
612
613    /// Resume after the executor already wrote the Starting boundary.
614    pub(crate) fn continue_from(
615        path: PathBuf,
616        started_status: RunStatus,
617        cwd: PathBuf,
618        prompt: &str,
619        stream_output: bool,
620        status_tx: Option<tokio::sync::watch::Sender<RunStatus>>,
621    ) -> anyhow::Result<Self> {
622        let sink = crate::run_artifacts::RunArtifactSink::continue_from(
623            path,
624            started_status,
625            prompt,
626            status_tx,
627        )?;
628        Ok(Self {
629            sink,
630            adapter: crate::tui::event_adapter::SdkEventAdapter::new(cwd),
631            stream_output,
632        })
633    }
634
635    pub(super) fn on_event(&mut self, event: &rho_sdk::RunEvent) {
636        use rho_sdk::RunEvent;
637
638        let attachments = crate::tui::translate_run_event(&mut self.adapter, event);
639        if !attachments.is_empty() {
640            let mut saw_text_delta = false;
641            let mut needs_immediate_publish = false;
642            for attachment in attachments {
643                match &attachment {
644                    crate::run_artifacts::AttachmentEvent::AssistantTextDelta(text)
645                        if !text.is_empty() =>
646                    {
647                        self.sink.append_last_text(text);
648                        saw_text_delta = true;
649                        self.sink.write_attachment(attachment);
650                    }
651                    _ => {
652                        needs_immediate_publish = true;
653                        self.sink.write_attachment(attachment);
654                    }
655                }
656            }
657            if needs_immediate_publish {
658                self.sink.publish();
659            } else if saw_text_delta {
660                self.sink.publish_throttled();
661            }
662        }
663        match event {
664            RunEvent::StepStarted { step } => {
665                self.sink.status.state = RunState::Running;
666                self.sink.status.turns = *step as u64;
667                self.sink.publish();
668            }
669            RunEvent::ToolStarted { name, .. } => {
670                self.sink.status.last_activity = Some(format!("tool: {name}"));
671                self.stream(&format!("\n[tool] {name}\n"));
672                self.sink.publish();
673            }
674            RunEvent::HostInputRequested { request }
675            | RunEvent::ToolHostInputRequested { request, .. } => {
676                self.sink.status.last_activity =
677                    Some(format!("waiting for questionnaire: {}", request.title()));
678                self.sink.publish();
679            }
680            RunEvent::AssistantTextDelta { text } => {
681                self.sink.status.last_activity = Some("assistant text".into());
682                self.stream(text);
683                // Attachment path already published throttled when translated.
684            }
685            RunEvent::ProviderStreamReset { .. } => {
686                self.sink.status.last_activity = Some("retrying provider response".into());
687                self.sink.status.last_text = None;
688                self.stream("\n[provider response discarded; retrying]\n");
689                self.sink.publish();
690            }
691            RunEvent::UsageUpdated { usage } => {
692                self.sink.status.input_tokens = usage.total_input_tokens();
693                self.sink.status.output_tokens = usage.output_tokens;
694            }
695            _ => {}
696        }
697    }
698
699    #[cfg(test)]
700    pub(crate) fn status(&self) -> &RunStatus {
701        &self.sink.status
702    }
703
704    pub(super) fn write(&mut self) {
705        self.sink.publish();
706    }
707
708    pub(crate) fn finish(&mut self, result: &anyhow::Result<rho_sdk::RunOutcome>) {
709        match result {
710            Ok(outcome) => {
711                let usage = outcome.usage();
712                self.sink.status.input_tokens = usage.total_input_tokens();
713                self.sink.status.output_tokens = usage.output_tokens;
714                self.sink.finish_ok(Some(outcome.text().to_string()));
715            }
716            Err(error)
717                if error.is::<AutomationInterrupted>()
718                    || error.downcast_ref::<AutomationExit>().is_some_and(|exit| {
719                        matches!(
720                            exit.reason(),
721                            TerminalReason::MaxSteps | TerminalReason::Timeout
722                        )
723                    })
724                    || error.is::<SubagentCancelled>() =>
725            {
726                self.sink.finish_stopped("stopped");
727            }
728            Err(error) => {
729                self.sink.finish_error(format!("{error:#}"));
730            }
731        }
732    }
733
734    fn stream(&self, text: &str) {
735        if !self.stream_output {
736            return;
737        }
738        let mut stdout = io::stdout().lock();
739        let _ = stdout.write_all(text.as_bytes());
740        let _ = stdout.flush();
741    }
742}
743
744#[cfg(unix)]
745async fn shutdown_signal() -> io::Result<ShutdownSignal> {
746    use tokio::signal::unix::{signal, SignalKind};
747
748    let mut interrupt = signal(SignalKind::interrupt())?;
749    let mut terminate = signal(SignalKind::terminate())?;
750    tokio::select! {
751        _ = interrupt.recv() => Ok(ShutdownSignal::Interrupt),
752        _ = terminate.recv() => Ok(ShutdownSignal::Terminate),
753    }
754}
755
756#[cfg(not(unix))]
757async fn shutdown_signal() -> io::Result<ShutdownSignal> {
758    tokio::signal::ctrl_c().await?;
759    Ok(ShutdownSignal::Interrupt)
760}
761
762fn prompt_from_stdin(parts: Vec<String>, read_stdin: bool) -> anyhow::Result<String> {
763    prompt_from_reader(parts, read_stdin, &mut io::stdin())
764}
765
766fn prompt_from_reader(
767    parts: Vec<String>,
768    read_stdin: bool,
769    stdin: &mut impl Read,
770) -> anyhow::Result<String> {
771    let mut chunks = Vec::new();
772    let inline = parts.join(" ").trim().to_string();
773    if !inline.is_empty() {
774        chunks.push(inline);
775    }
776    if read_stdin {
777        let mut buffer = String::new();
778        stdin.read_to_string(&mut buffer)?;
779        let buffer = buffer.trim().to_string();
780        if !buffer.is_empty() {
781            chunks.push(buffer);
782        }
783    }
784
785    let prompt = chunks.join("\n\n");
786    if prompt.is_empty() {
787        anyhow::bail!("rho run requires a prompt argument or --stdin");
788    }
789    Ok(prompt)
790}
791
792#[cfg(test)]
793#[path = "automation_tests.rs"]
794mod tests;
rho_coding_agent/app/automation.rs

rho_coding_agent/app/
automation.rs
