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

rho_coding_agent/app/
automation.rs
