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() -> anyhow::Result<()> {
170    let mut adapter = JsonlAdapter::new();
171    let event = adapter.failed(
172        TerminalReason::ConfigurationError,
173        "configuration failed".into(),
174        None,
175    );
176    emit(event)
177}
178
179pub(super) async fn run(prompt_text: String, startup: Startup<'_>) -> anyhow::Result<()> {
180    let mut jsonl = (startup.output == OutputFormat::Jsonl).then(JsonlAdapter::new);
181    let deadline = startup
182        .timeout
183        .map(|timeout| tokio::time::Instant::now() + timeout);
184    // The reporter exists before anything that can fail, so a parent process
185    // watching the output file always sees a terminal state, including startup failures.
186    let reporter_result = startup
187        .output_file
188        .as_ref()
189        .map(|path| {
190            RunReporter::new(
191                path.clone(),
192                RunArtifactIdentity {
193                    agent_id: startup.agent.id().to_string(),
194                    agent_fingerprint: startup.agent.fingerprint().to_string(),
195                    provider: startup.config.provider.clone(),
196                    model: startup.config.model.clone(),
197                },
198                startup.cwd.clone(),
199                &prompt_text,
200                /* stream_output */ startup.output == OutputFormat::Text,
201                None,
202            )
203        })
204        .transpose();
205    let mut reporter = match reporter_result {
206        Ok(reporter) => reporter,
207        Err(error) => {
208            emit_failure(&mut jsonl, TerminalReason::OutputError, &error)?;
209            return Err(
210                AutomationExit::new(1, TerminalReason::OutputError, error.to_string()).into(),
211            );
212        }
213    };
214
215    let cancellation = rho_tools::cancellation::RunCancellation::default();
216    let (result, timed_out) = if let Some(deadline) = deadline {
217        let future = run_session_with_output(
218            prompt_text,
219            &startup,
220            reporter.as_mut(),
221            Some(cancellation.clone()),
222            jsonl.as_mut(),
223        );
224        tokio::pin!(future);
225        tokio::select! {
226            result = &mut future => (result, false),
227            () = tokio::time::sleep_until(deadline) => {
228                cancellation.cancel();
229                (future.await, true)
230            }
231        }
232    } else {
233        (
234            run_session_with_output(
235                prompt_text,
236                &startup,
237                reporter.as_mut(),
238                None,
239                jsonl.as_mut(),
240            )
241            .await,
242            false,
243        )
244    };
245    if let Some(reporter) = reporter.as_mut() {
246        let reached_step_limit = result.as_ref().is_ok_and(|outcome| {
247            outcome.stop_reason() == rho_sdk::StopReason::MaxSteps
248                && (jsonl.is_some() || startup.max_steps.is_some())
249        });
250        if reached_step_limit {
251            let stopped = Err(AutomationExit::new(
252                124,
253                TerminalReason::MaxSteps,
254                "rho run reached its model-step limit",
255            )
256            .into());
257            reporter.finish(&stopped);
258        } else {
259            reporter.finish(&result);
260        }
261    }
262
263    if timed_out {
264        emit_stopped(&mut jsonl, TerminalReason::Timeout)?;
265        return Err(AutomationExit::new(124, TerminalReason::Timeout, "rho run timed out").into());
266    }
267
268    match result {
269        Ok(answer) => {
270            let max_steps = answer.stop_reason() == rho_sdk::StopReason::MaxSteps;
271            if max_steps && (jsonl.is_some() || startup.max_steps.is_some()) {
272                if let Some(adapter) = jsonl.as_mut() {
273                    let text = (!answer.text().is_empty()).then(|| answer.text().into());
274                    let event = adapter.stopped(TerminalReason::MaxSteps, text);
275                    emit(event)?;
276                } else {
277                    write_text_answer(&answer, reporter.is_some())?;
278                }
279                return Err(AutomationExit::new(
280                    124,
281                    TerminalReason::MaxSteps,
282                    "rho run reached its model-step limit",
283                )
284                .into());
285            }
286            if let Some(adapter) = jsonl.as_mut() {
287                let event = adapter.completed(answer.text().into());
288                emit(event)?;
289            } else {
290                write_text_answer(&answer, reporter.is_some())?;
291            }
292            Ok(())
293        }
294        Err(error) => {
295            let (reason, code) = classify_error(&error);
296            if reason == TerminalReason::Interrupted {
297                emit_stopped(&mut jsonl, reason)?;
298            } else if reason != TerminalReason::OutputError {
299                emit_failure(&mut jsonl, reason, &error)?;
300            }
301            let message = terminal_error_message(reason, &error);
302            if error.is::<AutomationInterrupted>() {
303                return Err(error);
304            }
305            Err(AutomationExit::new(code, reason, message).into())
306        }
307    }
308}
309
310fn write_text_answer(answer: &rho_sdk::RunOutcome, has_reporter: bool) -> anyhow::Result<()> {
311    let result = (|| -> io::Result<()> {
312        let mut stdout = io::stdout().lock();
313        if has_reporter {
314            writeln!(stdout, "\n[subagent run complete]")?;
315        } else {
316            writeln!(stdout, "{}", answer.text())?;
317        }
318        stdout.flush()
319    })();
320    result.map_err(|error| {
321        AutomationExit::new(
322            1,
323            TerminalReason::OutputError,
324            format!("could not write output: {error}"),
325        )
326        .into()
327    })
328}
329
330pub(super) fn emit(event: WireEvent) -> anyhow::Result<()> {
331    let mut stdout = io::stdout().lock();
332    write_event(&mut stdout, &event).map_err(|error| {
333        AutomationExit::new(
334            1,
335            TerminalReason::OutputError,
336            format!("could not write JSONL output: {error}"),
337        )
338        .into()
339    })
340}
341
342fn emit_stopped(adapter: &mut Option<JsonlAdapter>, reason: TerminalReason) -> anyhow::Result<()> {
343    if let Some(adapter) = adapter.as_mut() {
344        let text = adapter.partial_text();
345        let event = adapter.stopped(reason, text);
346        emit(event)?;
347    }
348    Ok(())
349}
350
351fn emit_failure(
352    adapter: &mut Option<JsonlAdapter>,
353    reason: TerminalReason,
354    error: &anyhow::Error,
355) -> anyhow::Result<()> {
356    if let Some(adapter) = adapter.as_mut() {
357        let text = adapter.partial_text();
358        let message = terminal_error_message(reason, error);
359        let event = adapter.failed(reason, message, text);
360        emit(event)?;
361    }
362    Ok(())
363}
364
365fn terminal_error_message(reason: TerminalReason, error: &anyhow::Error) -> String {
366    match reason {
367        TerminalReason::Authentication => "authentication failed".to_string(),
368        TerminalReason::ConfigurationError => "configuration failed".to_string(),
369        TerminalReason::OutputError => "output failed".to_string(),
370        TerminalReason::OtherError => "run failed".to_string(),
371        _ => error.to_string(),
372    }
373}
374
375fn classify_error(error: &anyhow::Error) -> (TerminalReason, u8) {
376    if let Some(interrupted) = error.downcast_ref::<AutomationInterrupted>() {
377        return (TerminalReason::Interrupted, interrupted.exit_code());
378    }
379    if let Some(exit) = error.downcast_ref::<AutomationExit>() {
380        return (exit.reason(), exit.exit_code());
381    }
382    for cause in error.chain() {
383        if let Some(error) = cause.downcast_ref::<rho_sdk::Error>() {
384            return match error {
385                rho_sdk::Error::Authentication { .. } => (TerminalReason::Authentication, 1),
386                rho_sdk::Error::Provider(provider)
387                    if provider.kind() == rho_sdk::ProviderErrorKind::Authentication =>
388                {
389                    (TerminalReason::Authentication, 1)
390                }
391                rho_sdk::Error::Provider(_) => (TerminalReason::ProviderError, 1),
392                rho_sdk::Error::Tool(_) => (TerminalReason::ToolHostError, 1),
393                rho_sdk::Error::InvalidConfiguration { .. } => {
394                    (TerminalReason::ConfigurationError, 2)
395                }
396                _ => (TerminalReason::OtherError, 1),
397            };
398        }
399        if let Some(error) = cause.downcast_ref::<rho_providers::model::ModelError>() {
400            use rho_providers::model::ModelError;
401            return match error {
402                ModelError::MissingCredentials(_) | ModelError::Credentials(_) => {
403                    (TerminalReason::Authentication, 1)
404                }
405                ModelError::UnsupportedReasoning { .. } | ModelError::UnsupportedProvider(_) => {
406                    (TerminalReason::ConfigurationError, 2)
407                }
408                _ => (TerminalReason::ProviderError, 1),
409            };
410        }
411    }
412    (TerminalReason::OtherError, 1)
413}
414
415pub(crate) async fn run_session(
416    prompt_text: String,
417    startup: &Startup<'_>,
418    reporter: Option<&mut RunReporter>,
419    cancellation: Option<rho_tools::cancellation::RunCancellation>,
420) -> anyhow::Result<rho_sdk::RunOutcome> {
421    run_session_with_output(prompt_text, startup, reporter, cancellation, None).await
422}
423
424async fn run_session_with_output(
425    prompt_text: String,
426    startup: &Startup<'_>,
427    reporter: Option<&mut RunReporter>,
428    cancellation: Option<rho_tools::cancellation::RunCancellation>,
429    mut jsonl: Option<&mut JsonlAdapter>,
430) -> anyhow::Result<rho_sdk::RunOutcome> {
431    let sdk_options = SdkBootstrapOptions::from_config(startup.config, &startup.cwd)?;
432    let credentials = rho_providers::auth::provider_credentials::ApplicationCredentialSource::new(
433        Arc::new(AppCredentialStore),
434    );
435    let provider = build_automation_provider(sdk_options.provider, &credentials)?;
436    let (tool_set, system_prompt) = assemble_tools_and_prompt(ToolsAndPromptOptions {
437        config: startup.config,
438        config_path: startup.config_path.clone(),
439        cwd: &startup.cwd,
440        no_system_prompt: startup.no_system_prompt,
441        no_tools: startup.no_tools,
442        no_subagents: startup.no_subagents,
443        // Automation keeps questionnaire capability when the agent exposes it.
444        questionnaire_enabled: true,
445        background_subagents: BackgroundSubagents::Disabled,
446        diagnostics: &startup.diagnostics,
447        agent: &startup.agent,
448    })?;
449
450    let workspace_root = sdk_options.workspace.root.clone();
451    let workspace = sdk_options.workspace.build_workspace()?;
452    let context_window = configured_context_window(startup.config);
453    let compaction = sdk_options.runtime.compaction.clone();
454    startup.diagnostics.update_compaction_config(&compaction);
455    let usage_recording = crate::usage::default_recording().await;
456    let hooks = crate::hooks::start_for_cwd(&workspace_root);
457    if let Some(hooks) = hooks.as_ref() {
458        startup.diagnostics.attach_hooks(hooks);
459    }
460    let runtime = build_runtime_with_max_steps(
461        RuntimeBuildOptions {
462            provider,
463            tools: tool_set.tools(),
464            workspace,
465            workspace_policy: AppPolicy::for_mode(startup.config.permission_mode),
466            approval_session: startup.approval_session.clone(),
467            system_prompt,
468            reasoning: sdk_options.runtime.reasoning,
469            service_tier: sdk_options.runtime.service_tier,
470            compaction,
471            context_window,
472            usage_purpose: startup.usage_purpose,
473            usage_parent_session_id: startup.parent_session_id.clone(),
474            usage_recording,
475            hook_host_labels: startup.hook_host_labels.clone(),
476            hooks: hooks.as_ref(),
477        },
478        startup.max_steps,
479    )?;
480    let session = runtime.session(SessionOptions::default()).await?;
481    if let Some(adapter) = jsonl.as_deref_mut() {
482        adapter.set_run_context(session.id(), &workspace_root);
483    }
484    startup
485        .herdr
486        .report_state(HerdrState::Working, None, None)
487        .await;
488    let result = complete_run(
489        &session,
490        prompt_text,
491        HeadlessRunDeps {
492            reporter,
493            external_cancellation: cancellation,
494            jsonl,
495            host_input: startup.host_input.as_deref(),
496        },
497    )
498    .await;
499
500    let session_hooks = runtime.hooks();
501    let session_id = session.id().clone();
502    match &result {
503        Ok(_) => {
504            session_hooks.session_completed(&session_id, /* completed_runs */ 1)
505        }
506        Err(error) => session_hooks.session_failed(
507            &session_id,
508            rho_sdk::hooks::HookSessionFailureKind::RunFailed,
509            &error.to_string(),
510        ),
511    }
512    runtime.shutdown();
513    drop(session);
514    drop(runtime);
515    if let Some(hooks) = hooks {
516        hooks.shutdown(crate::hooks::DRAIN_GRACE).await;
517    }
518    tool_set.shutdown().await;
519    startup
520        .herdr
521        .report_state(HerdrState::Idle, None, None)
522        .await;
523    startup.herdr.release().await;
524
525    result
526}
527
528async fn complete_run(
529    session: &rho_sdk::Session,
530    prompt_text: String,
531    dependencies: HeadlessRunDeps<'_>,
532) -> anyhow::Result<rho_sdk::RunOutcome> {
533    let HeadlessRunDeps {
534        reporter,
535        external_cancellation,
536        jsonl,
537        host_input,
538    } = dependencies;
539    let mut run = session.start(UserInput::text(prompt_text)).await?;
540    let cancellation = run.cancellation_handle();
541    let external_cancellation = external_cancellation.unwrap_or_default();
542    tokio::select! {
543        outcome = headless_run::drive(&mut run, reporter, jsonl, host_input) => outcome,
544        signal = shutdown_signal() => {
545            let signal = signal?;
546            cancellation.cancel();
547            let _ = run.outcome().await;
548            Err(AutomationInterrupted::new(signal).into())
549        }
550        () = external_cancellation.cancelled() => {
551            cancellation.cancel();
552            let _ = run.outcome().await;
553            Err(SubagentCancelled.into())
554        }
555    }
556}
557
558pub(crate) use crate::run_artifacts::RunArtifactIdentity;
559
560/// Maintains the `--output-file` status contract for subagent runs and
561/// streams progress to stdout so a watching pane shows live activity.
562pub(crate) struct RunReporter {
563    sink: crate::run_artifacts::RunArtifactSink,
564    adapter: crate::tui::event_adapter::SdkEventAdapter,
565    stream_output: bool,
566}
567
568impl RunReporter {
569    pub(crate) fn new(
570        path: PathBuf,
571        identity: RunArtifactIdentity,
572        cwd: PathBuf,
573        prompt: &str,
574        stream_output: bool,
575        status_tx: Option<tokio::sync::watch::Sender<RunStatus>>,
576    ) -> anyhow::Result<Self> {
577        let sink = crate::run_artifacts::RunArtifactSink::open(path, &identity, prompt, status_tx)?;
578        Ok(Self {
579            sink,
580            adapter: crate::tui::event_adapter::SdkEventAdapter::new(cwd),
581            stream_output,
582        })
583    }
584
585    /// Resume after the executor already wrote the Starting boundary.
586    pub(crate) fn continue_from(
587        path: PathBuf,
588        started_status: RunStatus,
589        cwd: PathBuf,
590        prompt: &str,
591        stream_output: bool,
592        status_tx: Option<tokio::sync::watch::Sender<RunStatus>>,
593    ) -> anyhow::Result<Self> {
594        let sink = crate::run_artifacts::RunArtifactSink::continue_from(
595            path,
596            started_status,
597            prompt,
598            status_tx,
599        )?;
600        Ok(Self {
601            sink,
602            adapter: crate::tui::event_adapter::SdkEventAdapter::new(cwd),
603            stream_output,
604        })
605    }
606
607    pub(super) fn on_event(&mut self, event: &rho_sdk::RunEvent) {
608        use rho_sdk::RunEvent;
609
610        let attachments = crate::tui::translate_run_event(&mut self.adapter, event);
611        if !attachments.is_empty() {
612            let mut saw_text_delta = false;
613            let mut needs_immediate_publish = false;
614            for attachment in attachments {
615                match &attachment {
616                    crate::run_artifacts::AttachmentEvent::AssistantTextDelta(text)
617                        if !text.is_empty() =>
618                    {
619                        self.sink.append_last_text(text);
620                        saw_text_delta = true;
621                        self.sink.write_attachment(attachment);
622                    }
623                    _ => {
624                        needs_immediate_publish = true;
625                        self.sink.write_attachment(attachment);
626                    }
627                }
628            }
629            if needs_immediate_publish {
630                self.sink.publish();
631            } else if saw_text_delta {
632                self.sink.publish_throttled();
633            }
634        }
635        match event {
636            RunEvent::StepStarted { step } => {
637                self.sink.status.state = RunState::Running;
638                self.sink.status.turns = *step as u64;
639                self.sink.publish();
640            }
641            RunEvent::ToolStarted { name, .. } => {
642                self.sink.status.last_activity = Some(format!("tool: {name}"));
643                self.stream(&format!("\n[tool] {name}\n"));
644                self.sink.publish();
645            }
646            RunEvent::HostInputRequested { request }
647            | RunEvent::ToolHostInputRequested { request, .. } => {
648                self.sink.status.last_activity =
649                    Some(format!("waiting for questionnaire: {}", request.title()));
650                self.sink.publish();
651            }
652            RunEvent::AssistantTextDelta { text } => {
653                self.sink.status.last_activity = Some("assistant text".into());
654                self.stream(text);
655                // Attachment path already published throttled when translated.
656            }
657            RunEvent::ProviderStreamReset { .. } => {
658                self.sink.status.last_activity = Some("retrying provider response".into());
659                self.sink.status.last_text = None;
660                self.stream("\n[provider response discarded; retrying]\n");
661                self.sink.publish();
662            }
663            RunEvent::UsageUpdated { usage } => {
664                self.sink.status.input_tokens = usage.total_input_tokens();
665                self.sink.status.output_tokens = usage.output_tokens;
666            }
667            _ => {}
668        }
669    }
670
671    #[cfg(test)]
672    pub(crate) fn status(&self) -> &RunStatus {
673        &self.sink.status
674    }
675
676    pub(super) fn write(&mut self) {
677        self.sink.publish();
678    }
679
680    pub(crate) fn finish(&mut self, result: &anyhow::Result<rho_sdk::RunOutcome>) {
681        match result {
682            Ok(outcome) => {
683                let usage = outcome.usage();
684                self.sink.status.input_tokens = usage.total_input_tokens();
685                self.sink.status.output_tokens = usage.output_tokens;
686                self.sink.finish_ok(Some(outcome.text().to_string()));
687            }
688            Err(error)
689                if error.is::<AutomationInterrupted>()
690                    || error.downcast_ref::<AutomationExit>().is_some_and(|exit| {
691                        matches!(
692                            exit.reason(),
693                            TerminalReason::MaxSteps | TerminalReason::Timeout
694                        )
695                    })
696                    || error.is::<SubagentCancelled>() =>
697            {
698                self.sink.finish_stopped("stopped");
699            }
700            Err(error) => {
701                self.sink.finish_error(format!("{error:#}"));
702            }
703        }
704    }
705
706    fn stream(&self, text: &str) {
707        if !self.stream_output {
708            return;
709        }
710        let mut stdout = io::stdout().lock();
711        let _ = stdout.write_all(text.as_bytes());
712        let _ = stdout.flush();
713    }
714}
715
716#[cfg(unix)]
717async fn shutdown_signal() -> io::Result<ShutdownSignal> {
718    use tokio::signal::unix::{signal, SignalKind};
719
720    let mut interrupt = signal(SignalKind::interrupt())?;
721    let mut terminate = signal(SignalKind::terminate())?;
722    tokio::select! {
723        _ = interrupt.recv() => Ok(ShutdownSignal::Interrupt),
724        _ = terminate.recv() => Ok(ShutdownSignal::Terminate),
725    }
726}
727
728#[cfg(not(unix))]
729async fn shutdown_signal() -> io::Result<ShutdownSignal> {
730    tokio::signal::ctrl_c().await?;
731    Ok(ShutdownSignal::Interrupt)
732}
733
734fn prompt_from_stdin(parts: Vec<String>, read_stdin: bool) -> anyhow::Result<String> {
735    prompt_from_reader(parts, read_stdin, &mut io::stdin())
736}
737
738fn prompt_from_reader(
739    parts: Vec<String>,
740    read_stdin: bool,
741    stdin: &mut impl Read,
742) -> anyhow::Result<String> {
743    let mut chunks = Vec::new();
744    let inline = parts.join(" ").trim().to_string();
745    if !inline.is_empty() {
746        chunks.push(inline);
747    }
748    if read_stdin {
749        let mut buffer = String::new();
750        stdin.read_to_string(&mut buffer)?;
751        let buffer = buffer.trim().to_string();
752        if !buffer.is_empty() {
753            chunks.push(buffer);
754        }
755    }
756
757    let prompt = chunks.join("\n\n");
758    if prompt.is_empty() {
759        anyhow::bail!("rho run requires a prompt argument or --stdin");
760    }
761    Ok(prompt)
762}
763
764#[cfg(test)]
765#[path = "automation_tests.rs"]
766mod tests;
rho_coding_agent/app/automation.rs

rho_coding_agent/app/
automation.rs
