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, ToolsAndPrompt, 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::Mcp { .. }
162            | Command::Plugins { .. }
163            | Command::Workflow { .. }
164            | Command::WorkflowPlannerWorker
165            | Command::Update,
166        )
167        | None => Ok(None),
168    }
169}
170
171pub(super) fn emit_startup_failure(message: impl Into<String>) -> anyhow::Result<()> {
172    let mut adapter = JsonlAdapter::new();
173    let event = adapter.failed(TerminalReason::ConfigurationError, message.into(), None);
174    emit(event)
175}
176
177pub(super) async fn run(prompt_text: String, startup: Startup<'_>) -> anyhow::Result<()> {
178    let mut jsonl = (startup.output == OutputFormat::Jsonl).then(JsonlAdapter::new);
179    let deadline = startup
180        .timeout
181        .map(|timeout| tokio::time::Instant::now() + timeout);
182    // The reporter exists before anything that can fail, so a parent process
183    // watching the output file always sees a terminal state, including startup failures.
184    let reporter_result = startup
185        .output_file
186        .as_ref()
187        .map(|path| {
188            RunReporter::new(
189                path.clone(),
190                RunArtifactIdentity {
191                    agent_id: startup.agent.id().to_string(),
192                    agent_fingerprint: startup.agent.fingerprint().to_string(),
193                    provider: startup.config.provider.clone(),
194                    model: startup.config.model.clone(),
195                    runtime: crate::agent::AgentRuntime::Rho,
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    let terminal = classify_run_terminal(result, timed_out);
245    if let Some(reporter) = reporter.as_mut() {
246        reporter.finish_terminal(&terminal);
247    }
248    emit_and_exit_terminal(terminal, &mut jsonl, reporter.is_some())
249}
250
251fn write_text_answer(answer: &rho_sdk::RunOutcome, has_reporter: bool) -> anyhow::Result<()> {
252    let result = (|| -> io::Result<()> {
253        let mut stdout = io::stdout().lock();
254        if has_reporter {
255            writeln!(stdout, "\n[subagent run complete]")?;
256        } else {
257            writeln!(stdout, "{}", answer.text())?;
258        }
259        stdout.flush()
260    })();
261    result.map_err(|error| {
262        AutomationExit::new(
263            1,
264            TerminalReason::OutputError,
265            format!("could not write output: {error}"),
266        )
267        .into()
268    })
269}
270
271pub(super) fn emit(event: WireEvent) -> anyhow::Result<()> {
272    let mut stdout = io::stdout().lock();
273    write_event(&mut stdout, &event).map_err(|error| {
274        AutomationExit::new(
275            1,
276            TerminalReason::OutputError,
277            format!("could not write JSONL output: {error}"),
278        )
279        .into()
280    })
281}
282
283fn emit_stopped(adapter: &mut Option<JsonlAdapter>, reason: TerminalReason) -> anyhow::Result<()> {
284    if let Some(adapter) = adapter.as_mut() {
285        let text = adapter.partial_text();
286        let event = adapter.stopped(reason, text);
287        emit(event)?;
288    }
289    Ok(())
290}
291
292fn emit_failure(
293    adapter: &mut Option<JsonlAdapter>,
294    reason: TerminalReason,
295    error: &anyhow::Error,
296) -> anyhow::Result<()> {
297    if let Some(adapter) = adapter.as_mut() {
298        let text = adapter.partial_text();
299        let message = terminal_error_message(reason, error);
300        let event = adapter.failed(reason, message, text);
301        emit(event)?;
302    }
303    Ok(())
304}
305
306const MAX_STEPS_MESSAGE: &str = "rho run reached its model-step limit";
307const TIMEOUT_MESSAGE: &str = "rho run timed out";
308
309/// Single classification of a finished automation session before reporter,
310/// JSONL/text emission, and process exit share one decision.
311enum RunTerminal {
312    Completed(rho_sdk::RunOutcome),
313    MaxSteps(rho_sdk::RunOutcome),
314    Timeout,
315    Failed(anyhow::Error),
316}
317
318fn classify_run_terminal(
319    result: anyhow::Result<rho_sdk::RunOutcome>,
320    timed_out: bool,
321) -> RunTerminal {
322    if timed_out {
323        return RunTerminal::Timeout;
324    }
325    match result {
326        Ok(answer) if answer.stop_reason() == rho_sdk::StopReason::MaxSteps => {
327            RunTerminal::MaxSteps(answer)
328        }
329        Ok(answer) => RunTerminal::Completed(answer),
330        Err(error) => RunTerminal::Failed(error),
331    }
332}
333
334fn emit_and_exit_terminal(
335    terminal: RunTerminal,
336    jsonl: &mut Option<JsonlAdapter>,
337    has_reporter: bool,
338) -> anyhow::Result<()> {
339    match terminal {
340        RunTerminal::Timeout => {
341            emit_stopped(jsonl, TerminalReason::Timeout)?;
342            Err(AutomationExit::new(124, TerminalReason::Timeout, TIMEOUT_MESSAGE).into())
343        }
344        RunTerminal::MaxSteps(answer) => {
345            if let Some(adapter) = jsonl.as_mut() {
346                let text = (!answer.text().is_empty()).then(|| answer.text().into());
347                let event = adapter.stopped(TerminalReason::MaxSteps, text);
348                emit(event)?;
349            } else {
350                write_text_answer(&answer, has_reporter)?;
351            }
352            Err(AutomationExit::new(124, TerminalReason::MaxSteps, MAX_STEPS_MESSAGE).into())
353        }
354        RunTerminal::Completed(answer) => {
355            if let Some(adapter) = jsonl.as_mut() {
356                let event = adapter.completed(answer.text().into());
357                emit(event)?;
358            } else {
359                write_text_answer(&answer, has_reporter)?;
360            }
361            Ok(())
362        }
363        RunTerminal::Failed(error) => {
364            let (reason, code) = classify_error(&error);
365            if reason == TerminalReason::Interrupted {
366                emit_stopped(jsonl, reason)?;
367            } else if reason != TerminalReason::OutputError {
368                emit_failure(jsonl, reason, &error)?;
369            }
370            let message = terminal_error_message(reason, &error);
371            if error.is::<AutomationInterrupted>() {
372                return Err(error);
373            }
374            Err(AutomationExit::new(code, reason, message).into())
375        }
376    }
377}
378
379/// Builds the human-readable terminal message for a classified automation error.
380///
381/// Reason codes stay stable machine labels. The message carries actionable detail
382/// except for authentication failures, which stay generic so credentials and
383/// token material never leave the process through stdout, stderr, or JSONL.
384fn terminal_error_message(reason: TerminalReason, error: &anyhow::Error) -> String {
385    match reason {
386        TerminalReason::Authentication => "authentication failed".to_string(),
387        TerminalReason::ProviderError
388        | TerminalReason::ToolHostError
389        | TerminalReason::ConfigurationError
390        | TerminalReason::OutputError
391        | TerminalReason::OtherError
392        | TerminalReason::Interrupted
393        | TerminalReason::MaxSteps
394        | TerminalReason::Timeout
395        | TerminalReason::Completed => error.to_string(),
396    }
397}
398
399fn classify_error(error: &anyhow::Error) -> (TerminalReason, u8) {
400    if let Some(interrupted) = error.downcast_ref::<AutomationInterrupted>() {
401        return (TerminalReason::Interrupted, interrupted.exit_code());
402    }
403    if let Some(exit) = error.downcast_ref::<AutomationExit>() {
404        return (exit.reason(), exit.exit_code());
405    }
406    for cause in error.chain() {
407        if let Some(error) = cause.downcast_ref::<rho_sdk::Error>() {
408            return match error {
409                rho_sdk::Error::Authentication { .. } => (TerminalReason::Authentication, 1),
410                rho_sdk::Error::Provider(provider)
411                    if provider.kind() == rho_sdk::ProviderErrorKind::Authentication =>
412                {
413                    (TerminalReason::Authentication, 1)
414                }
415                rho_sdk::Error::Provider(_) => (TerminalReason::ProviderError, 1),
416                rho_sdk::Error::Tool(_) => (TerminalReason::ToolHostError, 1),
417                rho_sdk::Error::InvalidConfiguration { .. } => {
418                    (TerminalReason::ConfigurationError, 2)
419                }
420                _ => (TerminalReason::OtherError, 1),
421            };
422        }
423        if let Some(error) = cause.downcast_ref::<rho_providers::model::ModelError>() {
424            use rho_providers::model::ModelError;
425            return match error {
426                ModelError::MissingCredentials(_) | ModelError::Credentials(_) => {
427                    (TerminalReason::Authentication, 1)
428                }
429                ModelError::UnsupportedReasoning { .. } | ModelError::UnsupportedProvider(_) => {
430                    (TerminalReason::ConfigurationError, 2)
431                }
432                _ => (TerminalReason::ProviderError, 1),
433            };
434        }
435    }
436    (TerminalReason::OtherError, 1)
437}
438
439pub(crate) async fn run_session(
440    prompt_text: String,
441    startup: &Startup<'_>,
442    reporter: Option<&mut RunReporter>,
443    cancellation: Option<rho_tools::cancellation::RunCancellation>,
444) -> anyhow::Result<rho_sdk::RunOutcome> {
445    run_session_with_output(prompt_text, startup, reporter, cancellation, None).await
446}
447
448async fn run_session_with_output(
449    prompt_text: String,
450    startup: &Startup<'_>,
451    reporter: Option<&mut RunReporter>,
452    cancellation: Option<rho_tools::cancellation::RunCancellation>,
453    mut jsonl: Option<&mut JsonlAdapter>,
454) -> anyhow::Result<rho_sdk::RunOutcome> {
455    let sdk_options = SdkBootstrapOptions::from_config(startup.config, &startup.cwd)?;
456    let credentials = rho_providers::auth::provider_credentials::ApplicationCredentialSource::new(
457        Arc::new(AppCredentialStore),
458    );
459    let provider = build_automation_provider(sdk_options.provider, &credentials)?;
460    let workspace_root = sdk_options.workspace.root.clone();
461    let workspace = sdk_options.workspace.build_workspace()?;
462    let ToolsAndPrompt {
463        tools: tool_set,
464        system_prompt,
465        ..
466    } = assemble_tools_and_prompt(ToolsAndPromptOptions {
467        config: startup.config,
468        config_path: startup.config_path.clone(),
469        cwd: &startup.cwd,
470        no_system_prompt: startup.no_system_prompt,
471        no_tools: startup.no_tools,
472        no_subagents: startup.no_subagents,
473        // Automation keeps questionnaire capability when the agent exposes it.
474        questionnaire_enabled: true,
475        // An automation run can only show a server's question when the caller
476        // supplied a responder for host input; without one the run would fail
477        // on the first question instead of declining it.
478        mcp_elicitation: match startup.host_input {
479            Some(_) => crate::tools::mcp::McpElicitationSupport::Available,
480            None => crate::tools::mcp::McpElicitationSupport::Unavailable,
481        },
482        // Automation binds no model for sampling, so it never declares the
483        // capability and rejects any request that arrives anyway.
484        mcp_sampling: crate::app::tools_prompt::McpSamplingSupport::Unavailable,
485        background_subagents: BackgroundSubagents::Disabled,
486        diagnostics: &startup.diagnostics,
487        agent: &startup.agent,
488    })
489    .await?;
490
491    let context_window = configured_context_window(startup.config);
492    let compaction = sdk_options.runtime.compaction.clone();
493    startup.diagnostics.update_compaction_config(&compaction);
494    let usage_recording = crate::usage::default_recording().await;
495    let hooks = crate::hooks::start_for_cwd(&workspace_root);
496    if let Some(hooks) = hooks.as_ref() {
497        startup.diagnostics.attach_hooks(hooks);
498    }
499    let startup_result: anyhow::Result<_> = async {
500        let runtime = build_runtime_with_max_steps(
501            RuntimeBuildOptions {
502                provider,
503                tools: tool_set.tools(),
504                workspace,
505                workspace_policy: AppPolicy::for_mode(startup.config.permission_mode),
506                approval_session: startup.approval_session.clone(),
507                system_prompt,
508                reasoning: sdk_options.runtime.reasoning,
509                service_tier: sdk_options.runtime.service_tier,
510                compaction,
511                context_window,
512                usage_purpose: startup.usage_purpose,
513                usage_parent_session_id: startup.parent_session_id.clone(),
514                usage_recording,
515                hook_host_labels: startup.hook_host_labels.clone(),
516                hooks: hooks.as_ref(),
517            },
518            startup.max_steps,
519        )?;
520        let session = match runtime.session(SessionOptions::default()).await {
521            Ok(session) => session,
522            Err(error) => {
523                runtime.shutdown();
524                return Err(error.into());
525            }
526        };
527        anyhow::Ok((runtime, session))
528    }
529    .await;
530    let (runtime, session) = match startup_result {
531        Ok(startup) => startup,
532        Err(error) => {
533            if let Some(hooks) = hooks {
534                hooks.shutdown(crate::hooks::DRAIN_GRACE).await;
535            }
536            tool_set.shutdown().await;
537            return Err(error);
538        }
539    };
540    if let Some(advisor) = tool_set.advisor() {
541        advisor.bind_session(session.clone());
542    }
543    if let Some(adapter) = jsonl.as_deref_mut() {
544        adapter.set_run_context(session.id(), &workspace_root);
545    }
546    startup
547        .herdr
548        .report_state(HerdrState::Working, None, None)
549        .await;
550    let result = complete_run(
551        &session,
552        prompt_text,
553        HeadlessRunDeps {
554            reporter,
555            external_cancellation: cancellation,
556            jsonl,
557            host_input: startup.host_input.as_deref(),
558        },
559    )
560    .await;
561
562    let session_hooks = runtime.hooks();
563    let session_id = session.id().clone();
564    match &result {
565        Ok(_) => {
566            session_hooks.session_completed(&session_id, /* completed_runs */ 1)
567        }
568        Err(error) => session_hooks.session_failed(
569            &session_id,
570            rho_sdk::hooks::HookSessionFailureKind::RunFailed,
571            &error.to_string(),
572        ),
573    }
574    runtime.shutdown();
575    drop(session);
576    drop(runtime);
577    if let Some(hooks) = hooks {
578        hooks.shutdown(crate::hooks::DRAIN_GRACE).await;
579    }
580    tool_set.shutdown().await;
581    startup
582        .herdr
583        .report_state(HerdrState::Idle, None, None)
584        .await;
585    startup.herdr.release().await;
586
587    result
588}
589
590async fn complete_run(
591    session: &rho_sdk::Session,
592    prompt_text: String,
593    dependencies: HeadlessRunDeps<'_>,
594) -> anyhow::Result<rho_sdk::RunOutcome> {
595    let HeadlessRunDeps {
596        reporter,
597        external_cancellation,
598        jsonl,
599        host_input,
600    } = dependencies;
601    let mut run = session.start(UserInput::text(prompt_text)).await?;
602    let cancellation = run.cancellation_handle();
603    let external_cancellation = external_cancellation.unwrap_or_default();
604    tokio::select! {
605        outcome = headless_run::drive(&mut run, reporter, jsonl, host_input) => outcome,
606        signal = shutdown_signal() => {
607            let signal = signal?;
608            cancellation.cancel();
609            let _ = run.outcome().await;
610            Err(AutomationInterrupted::new(signal).into())
611        }
612        () = external_cancellation.cancelled() => {
613            cancellation.cancel();
614            let _ = run.outcome().await;
615            Err(SubagentCancelled.into())
616        }
617    }
618}
619
620pub(crate) use crate::run_artifacts::RunArtifactIdentity;
621
622/// Maintains the `--output-file` status contract for subagent runs and
623/// streams progress to stdout so a watching pane shows live activity.
624pub(crate) struct RunReporter {
625    sink: crate::run_artifacts::RunArtifactSink,
626    adapter: crate::tui::event_adapter::SdkEventAdapter,
627    stream_output: bool,
628}
629
630impl RunReporter {
631    pub(crate) fn new(
632        path: PathBuf,
633        identity: RunArtifactIdentity,
634        cwd: PathBuf,
635        prompt: &str,
636        stream_output: bool,
637        status_tx: Option<tokio::sync::watch::Sender<RunStatus>>,
638    ) -> anyhow::Result<Self> {
639        let sink = crate::run_artifacts::RunArtifactSink::open(path, &identity, prompt, status_tx)?;
640        Ok(Self {
641            sink,
642            adapter: crate::tui::event_adapter::SdkEventAdapter::new(cwd),
643            stream_output,
644        })
645    }
646
647    /// Resume after the executor already wrote the Starting boundary.
648    pub(crate) fn continue_from(
649        path: PathBuf,
650        started_status: RunStatus,
651        cwd: PathBuf,
652        prompt: &str,
653        stream_output: bool,
654        status_tx: Option<tokio::sync::watch::Sender<RunStatus>>,
655    ) -> anyhow::Result<Self> {
656        let sink = crate::run_artifacts::RunArtifactSink::continue_from(
657            path,
658            started_status,
659            prompt,
660            status_tx,
661        )?;
662        Ok(Self {
663            sink,
664            adapter: crate::tui::event_adapter::SdkEventAdapter::new(cwd),
665            stream_output,
666        })
667    }
668
669    pub(super) fn on_event(&mut self, event: &rho_sdk::RunEvent) {
670        use rho_sdk::RunEvent;
671
672        let attachments = crate::tui::translate_run_event(&mut self.adapter, event);
673        for attachment in attachments {
674            // Reasoning is deliberately kept out of `last_text`: the status file
675            // carries the answer, not the thinking.
676            if let crate::run_artifacts::AttachmentEvent::AssistantTextDelta(text) = &attachment {
677                if !text.is_empty() {
678                    self.sink.append_last_text(text);
679                }
680            }
681            self.sink.record_attachment(attachment);
682        }
683        match event {
684            RunEvent::StepStarted { step, .. } => {
685                self.sink.status.state = RunState::Running;
686                self.sink.status.turns = *step as u64;
687                self.sink.publish();
688            }
689            RunEvent::ToolStarted { name, .. } => {
690                self.sink.status.last_activity = Some(format!("tool: {name}"));
691                self.stream(&format!("\n[tool] {name}\n"));
692                self.sink.publish();
693            }
694            RunEvent::HostInputRequested { request }
695            | RunEvent::ToolHostInputRequested { request, .. } => {
696                self.sink.status.last_activity =
697                    Some(format!("waiting for questionnaire: {}", request.title()));
698                self.sink.publish();
699            }
700            RunEvent::AssistantTextDelta { text } => {
701                self.sink.status.last_activity = Some("assistant text".into());
702                self.stream(text);
703                // Attachment path already published throttled when translated.
704            }
705            RunEvent::ProviderStreamReset { .. } => {
706                self.sink.status.last_activity = Some("retrying provider response".into());
707                self.sink.status.last_text = None;
708                self.stream("\n[provider response discarded; retrying]\n");
709                self.sink.publish();
710            }
711            RunEvent::UsageUpdated { usage } => {
712                self.sink.status.input_tokens = usage.total_input_tokens();
713                self.sink.status.output_tokens = usage.output_tokens;
714            }
715            _ => {}
716        }
717    }
718
719    #[cfg(test)]
720    pub(crate) fn status(&self) -> &RunStatus {
721        &self.sink.status
722    }
723
724    pub(super) fn write(&mut self) {
725        self.sink.publish();
726    }
727
728    pub(crate) fn finish(&mut self, result: &anyhow::Result<rho_sdk::RunOutcome>) {
729        match result {
730            Ok(outcome) => {
731                let usage = outcome.usage();
732                self.sink.status.input_tokens = usage.total_input_tokens();
733                self.sink.status.output_tokens = usage.output_tokens;
734                self.sink.finish_ok(Some(outcome.text().to_string()));
735            }
736            Err(error)
737                if error.is::<AutomationInterrupted>()
738                    || error.downcast_ref::<AutomationExit>().is_some_and(|exit| {
739                        matches!(
740                            exit.reason(),
741                            TerminalReason::MaxSteps | TerminalReason::Timeout
742                        )
743                    })
744                    || error.is::<SubagentCancelled>() =>
745            {
746                self.sink.finish_stopped("stopped");
747            }
748            Err(error) => {
749                self.sink.finish_error(format!("{error:#}"));
750            }
751        }
752    }
753
754    fn finish_terminal(&mut self, terminal: &RunTerminal) {
755        match terminal {
756            RunTerminal::Completed(outcome) => {
757                let usage = outcome.usage();
758                self.sink.status.input_tokens = usage.total_input_tokens();
759                self.sink.status.output_tokens = usage.output_tokens;
760                self.sink.finish_ok(Some(outcome.text().to_string()));
761            }
762            RunTerminal::MaxSteps(_) | RunTerminal::Timeout => {
763                self.sink.finish_stopped("stopped");
764            }
765            RunTerminal::Failed(error)
766                if error.is::<AutomationInterrupted>() || error.is::<SubagentCancelled>() =>
767            {
768                self.sink.finish_stopped("stopped");
769            }
770            RunTerminal::Failed(error) => {
771                self.sink.finish_error(format!("{error:#}"));
772            }
773        }
774    }
775
776    fn stream(&self, text: &str) {
777        if !self.stream_output {
778            return;
779        }
780        let mut stdout = io::stdout().lock();
781        let _ = stdout.write_all(text.as_bytes());
782        let _ = stdout.flush();
783    }
784}
785
786#[cfg(unix)]
787async fn shutdown_signal() -> io::Result<ShutdownSignal> {
788    use tokio::signal::unix::{signal, SignalKind};
789
790    let mut interrupt = signal(SignalKind::interrupt())?;
791    let mut terminate = signal(SignalKind::terminate())?;
792    tokio::select! {
793        _ = interrupt.recv() => Ok(ShutdownSignal::Interrupt),
794        _ = terminate.recv() => Ok(ShutdownSignal::Terminate),
795    }
796}
797
798#[cfg(not(unix))]
799async fn shutdown_signal() -> io::Result<ShutdownSignal> {
800    tokio::signal::ctrl_c().await?;
801    Ok(ShutdownSignal::Interrupt)
802}
803
804fn prompt_from_stdin(parts: Vec<String>, read_stdin: bool) -> anyhow::Result<String> {
805    if !read_stdin && crate::stdio::stdin_is_redirected() {
806        anyhow::bail!(
807            "stdin is redirected but --stdin was not set; pass --stdin to include piped input"
808        );
809    }
810    prompt_from_reader(parts, read_stdin, &mut io::stdin())
811}
812
813fn prompt_from_reader(
814    parts: Vec<String>,
815    read_stdin: bool,
816    stdin: &mut impl Read,
817) -> anyhow::Result<String> {
818    let mut chunks = Vec::new();
819    let inline = parts.join(" ").trim().to_string();
820    if !inline.is_empty() {
821        chunks.push(inline);
822    }
823    if read_stdin {
824        let mut buffer = String::new();
825        stdin.read_to_string(&mut buffer)?;
826        let buffer = buffer.trim().to_string();
827        if !buffer.is_empty() {
828            chunks.push(buffer);
829        }
830    }
831
832    let prompt = chunks.join("\n\n");
833    if prompt.is_empty() {
834        anyhow::bail!("rho run requires a prompt argument or --stdin");
835    }
836    Ok(prompt)
837}
838
839#[cfg(test)]
840#[path = "automation_tests.rs"]
841mod tests;
rho_coding_agent/app/automation.rs

rho_coding_agent/app/
automation.rs
