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, Workspace};
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::{self, RunState, RunStatus},
21    crate::tools::{
22        agent::BackgroundSubagents,
23        sdk_registry::{AppToolSet, DelegationConfig, ToolSetOptions},
24    },
25    crate::tui::AttachmentWriter,
26    rho_providers::providers::build_automation_provider,
27};
28
29use super::{
30    agent_binding::BoundAgent,
31    automation_protocol::{write_event, JsonlAdapter, TerminalReason, WireEvent},
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}
151
152pub(super) fn prompt_for_command(command: &Option<Command>) -> anyhow::Result<Option<String>> {
153    match command {
154        Some(Command::Run { prompt, stdin, .. }) => {
155            prompt_from_stdin(prompt.clone(), *stdin).map(Some)
156        }
157        Some(
158            Command::Attach { .. }
159            | Command::Login { .. }
160            | Command::CredentialStore { .. }
161            | Command::Update,
162        )
163        | None => Ok(None),
164    }
165}
166
167pub(super) fn emit_startup_failure() -> anyhow::Result<()> {
168    let mut adapter = JsonlAdapter::new();
169    let event = adapter.failed(
170        TerminalReason::ConfigurationError,
171        "configuration failed".into(),
172        None,
173    );
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                },
196                startup.cwd.clone(),
197                &prompt_text,
198                /* stream_output */ startup.output == OutputFormat::Text,
199                None,
200            )
201        })
202        .transpose();
203    let mut reporter = match reporter_result {
204        Ok(reporter) => reporter,
205        Err(error) => {
206            emit_failure(&mut jsonl, TerminalReason::OutputError, &error)?;
207            return Err(
208                AutomationExit::new(1, TerminalReason::OutputError, error.to_string()).into(),
209            );
210        }
211    };
212
213    let cancellation = rho_tools::cancellation::RunCancellation::default();
214    let (result, timed_out) = if let Some(deadline) = deadline {
215        let future = run_session_with_output(
216            prompt_text,
217            &startup,
218            reporter.as_mut(),
219            Some(cancellation.clone()),
220            jsonl.as_mut(),
221        );
222        tokio::pin!(future);
223        tokio::select! {
224            result = &mut future => (result, false),
225            () = tokio::time::sleep_until(deadline) => {
226                cancellation.cancel();
227                (future.await, true)
228            }
229        }
230    } else {
231        (
232            run_session_with_output(
233                prompt_text,
234                &startup,
235                reporter.as_mut(),
236                None,
237                jsonl.as_mut(),
238            )
239            .await,
240            false,
241        )
242    };
243    if let Some(reporter) = reporter.as_mut() {
244        let reached_step_limit = result.as_ref().is_ok_and(|outcome| {
245            outcome.stop_reason() == rho_sdk::StopReason::MaxSteps
246                && (jsonl.is_some() || startup.max_steps.is_some())
247        });
248        if reached_step_limit {
249            let stopped = Err(AutomationExit::new(
250                124,
251                TerminalReason::MaxSteps,
252                "rho run reached its model-step limit",
253            )
254            .into());
255            reporter.finish(&stopped);
256        } else {
257            reporter.finish(&result);
258        }
259    }
260
261    if timed_out {
262        emit_stopped(&mut jsonl, TerminalReason::Timeout)?;
263        return Err(AutomationExit::new(124, TerminalReason::Timeout, "rho run timed out").into());
264    }
265
266    match result {
267        Ok(answer) => {
268            let max_steps = answer.stop_reason() == rho_sdk::StopReason::MaxSteps;
269            if max_steps && (jsonl.is_some() || startup.max_steps.is_some()) {
270                if let Some(adapter) = jsonl.as_mut() {
271                    let text = (!answer.text().is_empty()).then(|| answer.text().into());
272                    let event = adapter.stopped(TerminalReason::MaxSteps, text);
273                    emit(event)?;
274                } else {
275                    write_text_answer(&answer, reporter.is_some())?;
276                }
277                return Err(AutomationExit::new(
278                    124,
279                    TerminalReason::MaxSteps,
280                    "rho run reached its model-step limit",
281                )
282                .into());
283            }
284            if let Some(adapter) = jsonl.as_mut() {
285                let event = adapter.completed(answer.text().into());
286                emit(event)?;
287            } else {
288                write_text_answer(&answer, reporter.is_some())?;
289            }
290            Ok(())
291        }
292        Err(error) => {
293            let (reason, code) = classify_error(&error);
294            if reason == TerminalReason::Interrupted {
295                emit_stopped(&mut jsonl, reason)?;
296            } else if reason != TerminalReason::OutputError {
297                emit_failure(&mut jsonl, reason, &error)?;
298            }
299            let message = terminal_error_message(reason, &error);
300            if error.is::<AutomationInterrupted>() {
301                return Err(error);
302            }
303            Err(AutomationExit::new(code, reason, message).into())
304        }
305    }
306}
307
308fn write_text_answer(answer: &rho_sdk::RunOutcome, has_reporter: bool) -> anyhow::Result<()> {
309    let result = (|| -> io::Result<()> {
310        let mut stdout = io::stdout().lock();
311        if has_reporter {
312            writeln!(stdout, "\n[subagent run complete]")?;
313        } else {
314            writeln!(stdout, "{}", answer.text())?;
315        }
316        stdout.flush()
317    })();
318    result.map_err(|error| {
319        AutomationExit::new(
320            1,
321            TerminalReason::OutputError,
322            format!("could not write output: {error}"),
323        )
324        .into()
325    })
326}
327
328fn emit(event: WireEvent) -> anyhow::Result<()> {
329    let mut stdout = io::stdout().lock();
330    write_event(&mut stdout, &event).map_err(|error| {
331        AutomationExit::new(
332            1,
333            TerminalReason::OutputError,
334            format!("could not write JSONL output: {error}"),
335        )
336        .into()
337    })
338}
339
340fn emit_stopped(adapter: &mut Option<JsonlAdapter>, reason: TerminalReason) -> anyhow::Result<()> {
341    if let Some(adapter) = adapter.as_mut() {
342        let text = adapter.partial_text();
343        let event = adapter.stopped(reason, text);
344        emit(event)?;
345    }
346    Ok(())
347}
348
349fn emit_failure(
350    adapter: &mut Option<JsonlAdapter>,
351    reason: TerminalReason,
352    error: &anyhow::Error,
353) -> anyhow::Result<()> {
354    if let Some(adapter) = adapter.as_mut() {
355        let text = adapter.partial_text();
356        let message = terminal_error_message(reason, error);
357        let event = adapter.failed(reason, message, text);
358        emit(event)?;
359    }
360    Ok(())
361}
362
363fn terminal_error_message(reason: TerminalReason, error: &anyhow::Error) -> String {
364    match reason {
365        TerminalReason::Authentication => "authentication failed".to_string(),
366        TerminalReason::ConfigurationError => "configuration failed".to_string(),
367        TerminalReason::OutputError => "output failed".to_string(),
368        TerminalReason::OtherError => "run failed".to_string(),
369        _ => error.to_string(),
370    }
371}
372
373fn classify_error(error: &anyhow::Error) -> (TerminalReason, u8) {
374    if let Some(interrupted) = error.downcast_ref::<AutomationInterrupted>() {
375        return (TerminalReason::Interrupted, interrupted.exit_code());
376    }
377    if let Some(exit) = error.downcast_ref::<AutomationExit>() {
378        return (exit.reason(), exit.exit_code());
379    }
380    for cause in error.chain() {
381        if let Some(error) = cause.downcast_ref::<rho_sdk::Error>() {
382            return match error {
383                rho_sdk::Error::Authentication { .. } => (TerminalReason::Authentication, 1),
384                rho_sdk::Error::Provider(provider)
385                    if provider.kind() == rho_sdk::ProviderErrorKind::Authentication =>
386                {
387                    (TerminalReason::Authentication, 1)
388                }
389                rho_sdk::Error::Provider(_) => (TerminalReason::ProviderError, 1),
390                rho_sdk::Error::Tool(_) => (TerminalReason::ToolHostError, 1),
391                rho_sdk::Error::InvalidConfiguration { .. } => {
392                    (TerminalReason::ConfigurationError, 2)
393                }
394                _ => (TerminalReason::OtherError, 1),
395            };
396        }
397        if let Some(error) = cause.downcast_ref::<rho_providers::model::ModelError>() {
398            use rho_providers::model::ModelError;
399            return match error {
400                ModelError::MissingApiKey
401                | ModelError::MissingCodexAuth
402                | ModelError::MissingAnthropicApiKey
403                | ModelError::MissingGoogleApiKey
404                | ModelError::MissingGithubCopilotAuth
405                | ModelError::MissingMoonshotApiKey
406                | ModelError::MissingPoolsideApiKey
407                | ModelError::MissingOpenRouterApiKey
408                | ModelError::MissingCredentialProfile(_)
409                | ModelError::MissingKimiAuth
410                | ModelError::MissingXaiApiKey
411                | ModelError::MissingXaiAuth
412                | ModelError::Credentials(_) => (TerminalReason::Authentication, 1),
413                ModelError::UnsupportedReasoning { .. } | ModelError::UnsupportedProvider(_) => {
414                    (TerminalReason::ConfigurationError, 2)
415                }
416                _ => (TerminalReason::ProviderError, 1),
417            };
418        }
419    }
420    (TerminalReason::OtherError, 1)
421}
422
423pub(crate) async fn run_session(
424    prompt_text: String,
425    startup: &Startup<'_>,
426    reporter: Option<&mut RunReporter>,
427    cancellation: Option<rho_tools::cancellation::RunCancellation>,
428) -> anyhow::Result<rho_sdk::RunOutcome> {
429    run_session_with_output(prompt_text, startup, reporter, cancellation, None).await
430}
431
432async fn run_session_with_output(
433    prompt_text: String,
434    startup: &Startup<'_>,
435    reporter: Option<&mut RunReporter>,
436    cancellation: Option<rho_tools::cancellation::RunCancellation>,
437    mut jsonl: Option<&mut JsonlAdapter>,
438) -> anyhow::Result<rho_sdk::RunOutcome> {
439    let sdk_options = SdkBootstrapOptions::from_config(startup.config, &startup.cwd)?;
440    let credentials = rho_providers::auth::provider_credentials::ApplicationCredentialSource::new(
441        Arc::new(AppCredentialStore),
442    );
443    let provider = build_automation_provider(sdk_options.provider, &credentials)?;
444    let mut capabilities = startup.agent.capabilities().clone();
445    if startup.no_subagents {
446        capabilities.remove(&ToolCapability::Agent);
447        capabilities.remove(&ToolCapability::Agents);
448    }
449    let launch_delegation_enabled = capabilities.contains(&ToolCapability::Agent);
450    let delegation_enabled =
451        launch_delegation_enabled || capabilities.contains(&ToolCapability::Agents);
452    let tool_set = if startup.no_tools {
453        AppToolSet::disabled()
454    } else {
455        let mut options = ToolSetOptions::new(capabilities);
456        if delegation_enabled {
457            options = options.delegation(DelegationConfig::new(
458                startup.cwd.clone(),
459                startup.config_path.clone(),
460                BackgroundSubagents::Disabled,
461            ));
462        }
463        AppToolSet::new(startup.config, startup.diagnostics.clone(), options)
464    };
465    let tool_specs = tool_set.specs();
466    let system_prompt = if startup.no_system_prompt {
467        startup.diagnostics.update_prompt_sources(Vec::new());
468        SystemPrompt::None
469    } else {
470        let mut text = match startup.agent.prompt() {
471            PromptPolicy::Replace(text) => text.clone(),
472            PromptPolicy::Extend(extra) => {
473                let built = prompt::system_prompt(&tool_specs, &startup.cwd);
474                startup.diagnostics.update_prompt_sources(built.sources);
475                let mut text = built.text;
476                if !launch_delegation_enabled {
477                    prompt::append_subagents_disabled_instruction(&mut text);
478                }
479                if !extra.is_empty() {
480                    text.push_str("\n\n# Agent instructions\n\n");
481                    text.push_str(extra);
482                }
483                text
484            }
485        };
486        if text.is_empty() {
487            text = "You are a coding agent.".into();
488        }
489        SystemPrompt::Custom(text)
490    };
491    startup.diagnostics.update_tools(&tool_specs);
492
493    let workspace_root = sdk_options.workspace.root.clone();
494    let workspace = Workspace::new(&workspace_root)?;
495    let context_window = configured_context_window(startup.config);
496    let compaction = sdk_options.runtime.compaction.clone();
497    startup.diagnostics.update_compaction_config(&compaction);
498    let usage_recording = crate::usage::default_recording().await;
499    let runtime = build_runtime_with_max_steps(
500        RuntimeBuildOptions {
501            provider,
502            tools: tool_set.tools(),
503            workspace,
504            workspace_policy: AppPolicy::for_mode(startup.config.permission_mode),
505            approval_handler: None,
506            system_prompt,
507            reasoning: sdk_options.runtime.reasoning,
508            compaction,
509            context_window,
510            usage_purpose: startup.usage_purpose,
511            usage_parent_session_id: startup.parent_session_id.clone(),
512            usage_recording,
513        },
514        startup.max_steps,
515    )?;
516    let session = runtime.session(SessionOptions::default()).await?;
517    if let Some(adapter) = jsonl.as_deref_mut() {
518        adapter.set_run_context(session.id(), &workspace_root);
519    }
520    if let Some(manager) = tool_set.subagents() {
521        manager.set_session(session.id().to_string());
522    }
523
524    startup
525        .herdr
526        .report_state(HerdrState::Working, None, None)
527        .await;
528    let result = complete_run(&session, prompt_text, reporter, cancellation, jsonl).await;
529
530    runtime.shutdown();
531    tool_set.shutdown().await;
532    startup
533        .herdr
534        .report_state(HerdrState::Idle, None, None)
535        .await;
536    startup.herdr.release().await;
537
538    result
539}
540
541async fn complete_run(
542    session: &rho_sdk::Session,
543    prompt_text: String,
544    reporter: Option<&mut RunReporter>,
545    external_cancellation: Option<rho_tools::cancellation::RunCancellation>,
546    jsonl: Option<&mut JsonlAdapter>,
547) -> anyhow::Result<rho_sdk::RunOutcome> {
548    let mut run = session.start(UserInput::text(prompt_text)).await?;
549    let cancellation = run.cancellation_handle();
550    let external_cancellation = external_cancellation.unwrap_or_default();
551    tokio::select! {
552        outcome = drive_headless_run(&mut run, reporter, jsonl) => outcome,
553        signal = shutdown_signal() => {
554            let signal = signal?;
555            cancellation.cancel();
556            let _ = run.outcome().await;
557            Err(AutomationInterrupted::new(signal).into())
558        }
559        () = external_cancellation.cancelled() => {
560            cancellation.cancel();
561            let _ = run.outcome().await;
562            Err(SubagentCancelled.into())
563        }
564    }
565}
566
567/// Drains run events with no interactive host attached.
568///
569/// Host input requests cannot be answered headlessly; cancel instead of
570/// leaving the requesting tool suspended until a signal arrives.
571async fn drive_headless_run(
572    run: &mut rho_sdk::Run,
573    mut reporter: Option<&mut RunReporter>,
574    mut jsonl: Option<&mut JsonlAdapter>,
575) -> anyhow::Result<rho_sdk::RunOutcome> {
576    let mut heartbeat = tokio::time::interval(REPORT_HEARTBEAT);
577    heartbeat.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
578    loop {
579        let event = tokio::select! {
580            event = run.next_event() => event,
581            _ = heartbeat.tick(), if reporter.is_some() => {
582                if let Some(reporter) = reporter.as_deref_mut() {
583                    reporter.write();
584                }
585                continue;
586            }
587        };
588        let Some(event) = event else {
589            break;
590        };
591        if let Some(reporter) = reporter.as_deref_mut() {
592            reporter.on_event(&event);
593        }
594        if let Some(adapter) = jsonl.as_deref_mut() {
595            if let Some(wire_event) = adapter.event(&event) {
596                if let Err(error) = emit(wire_event) {
597                    run.cancel();
598                    let _ = run.outcome().await;
599                    return Err(error);
600                }
601            }
602        }
603        let request = match event {
604            rho_sdk::RunEvent::HostInputRequested { request }
605            | rho_sdk::RunEvent::ToolHostInputRequested { request, .. } => Some(request),
606            _ => None,
607        };
608        if let Some(request) = request {
609            run.cancel();
610            let _ = run.outcome().await;
611            anyhow::bail!(
612                "rho run cannot answer host input request '{}' ({}); run without tools that require interactive input",
613                request.id(),
614                request.title(),
615            );
616        }
617    }
618    Ok(run.outcome().await?)
619}
620
621pub(crate) struct RunArtifactIdentity {
622    pub(crate) agent_id: String,
623    pub(crate) agent_fingerprint: String,
624    pub(crate) provider: String,
625    pub(crate) model: String,
626}
627
628/// Maintains the `--output-file` status contract for subagent runs and
629/// streams progress to stdout so a watching pane shows live activity.
630pub(crate) struct RunReporter {
631    path: PathBuf,
632    status: RunStatus,
633    attachment: Option<AttachmentWriter>,
634    stream_output: bool,
635    status_tx: Option<tokio::sync::watch::Sender<RunStatus>>,
636    last_write: std::time::Instant,
637}
638
639/// Longest a status-file write is deferred while text streams.
640const REPORT_THROTTLE: std::time::Duration = std::time::Duration::from_secs(2);
641/// Keeps the status file fresh while a provider or tool call emits no events.
642const REPORT_HEARTBEAT: std::time::Duration = std::time::Duration::from_secs(10);
643const LAST_TEXT_BYTES: usize = 400;
644
645impl RunReporter {
646    pub(crate) fn new(
647        path: PathBuf,
648        identity: RunArtifactIdentity,
649        cwd: PathBuf,
650        prompt: &str,
651        stream_output: bool,
652        status_tx: Option<tokio::sync::watch::Sender<RunStatus>>,
653    ) -> anyhow::Result<Self> {
654        let status = RunStatus {
655            state: RunState::Starting,
656            agent_id: Some(identity.agent_id),
657            agent_fingerprint: Some(identity.agent_fingerprint),
658            provider: Some(identity.provider),
659            model: Some(identity.model),
660            ..RunStatus::default()
661        };
662        subagent::write_status(&path, &status)?;
663        let attachment = match AttachmentWriter::new(&path, cwd, prompt) {
664            Ok(attachment) => Some(attachment),
665            Err(error) => {
666                let mut status = status;
667                status.attachment_error = Some(format!("could not record attach output: {error}"));
668                subagent::write_status(&path, &status)?;
669                return Ok(Self {
670                    path,
671                    status,
672                    attachment: None,
673                    stream_output,
674                    status_tx,
675                    last_write: std::time::Instant::now(),
676                });
677            }
678        };
679        Ok(Self {
680            path,
681            status,
682            attachment,
683            stream_output,
684            status_tx,
685            last_write: std::time::Instant::now(),
686        })
687    }
688
689    fn on_event(&mut self, event: &rho_sdk::RunEvent) {
690        use rho_sdk::RunEvent;
691
692        if let Some(attachment) = self.attachment.as_mut() {
693            if let Err(error) = attachment.on_event(event) {
694                self.status.attachment_error =
695                    Some(format!("could not record attach output: {error}"));
696                self.attachment = None;
697                self.write();
698            }
699        }
700        match event {
701            RunEvent::StepStarted { step } => {
702                self.status.state = RunState::Running;
703                self.status.turns = *step as u64;
704                self.write();
705            }
706            RunEvent::ToolStarted { name, .. } => {
707                self.status.last_activity = Some(format!("tool: {name}"));
708                self.stream(&format!("\n[tool] {name}\n"));
709                self.write();
710            }
711            RunEvent::AssistantTextDelta { text } => {
712                self.status.last_activity = Some("assistant text".into());
713                append_tail(
714                    self.status.last_text.get_or_insert_with(String::new),
715                    text,
716                    LAST_TEXT_BYTES,
717                );
718                self.stream(text);
719                self.write_throttled();
720            }
721            RunEvent::ProviderStreamReset { .. } => {
722                self.status.last_activity = Some("retrying provider response".into());
723                self.status.last_text = None;
724                self.stream("\n[provider response discarded; retrying]\n");
725                self.write();
726            }
727            RunEvent::UsageUpdated { usage } => {
728                self.status.input_tokens = usage.total_input_tokens().unwrap_or(0);
729                self.status.output_tokens = usage.output_tokens.unwrap_or(0);
730            }
731            _ => {}
732        }
733    }
734
735    pub(crate) fn finish(&mut self, result: &anyhow::Result<rho_sdk::RunOutcome>) {
736        match result {
737            Ok(outcome) => {
738                self.status.state = RunState::Ok;
739                self.status.result = Some(outcome.text().to_string());
740                let usage = outcome.usage();
741                self.status.input_tokens = usage.total_input_tokens().unwrap_or(0);
742                self.status.output_tokens = usage.output_tokens.unwrap_or(0);
743            }
744            Err(error)
745                if error.is::<AutomationInterrupted>()
746                    || error.downcast_ref::<AutomationExit>().is_some_and(|exit| {
747                        matches!(
748                            exit.reason(),
749                            TerminalReason::MaxSteps | TerminalReason::Timeout
750                        )
751                    })
752                    || error.is::<SubagentCancelled>() =>
753            {
754                self.status.state = RunState::Stopped;
755                self.status.result = self
756                    .status
757                    .last_text
758                    .as_ref()
759                    .map(|text| format!("(partial, stopped before finishing)\n{text}"));
760            }
761            Err(error) => {
762                self.status.state = RunState::Error;
763                self.status.error = Some(format!("{error:#}"));
764            }
765        }
766        self.write();
767    }
768
769    fn stream(&self, text: &str) {
770        if !self.stream_output {
771            return;
772        }
773        let mut stdout = io::stdout().lock();
774        let _ = stdout.write_all(text.as_bytes());
775        let _ = stdout.flush();
776    }
777
778    fn write_throttled(&mut self) {
779        if self.last_write.elapsed() >= REPORT_THROTTLE {
780            self.write();
781        }
782    }
783
784    fn write(&mut self) {
785        self.last_write = std::time::Instant::now();
786        if let Some(status_tx) = &self.status_tx {
787            status_tx.send_replace(self.status.clone());
788        }
789        let _ = subagent::write_status(&self.path, &self.status);
790    }
791}
792
793/// Appends to a rolling tail buffer capped at `max` bytes.
794fn append_tail(buffer: &mut String, text: &str, max: usize) {
795    buffer.push_str(text);
796    if buffer.len() > max {
797        let cut = buffer.len() - max;
798        let boundary = (cut..buffer.len())
799            .find(|index| buffer.is_char_boundary(*index))
800            .unwrap_or(buffer.len());
801        buffer.drain(..boundary);
802    }
803}
804
805#[cfg(unix)]
806async fn shutdown_signal() -> io::Result<ShutdownSignal> {
807    use tokio::signal::unix::{signal, SignalKind};
808
809    let mut interrupt = signal(SignalKind::interrupt())?;
810    let mut terminate = signal(SignalKind::terminate())?;
811    tokio::select! {
812        _ = interrupt.recv() => Ok(ShutdownSignal::Interrupt),
813        _ = terminate.recv() => Ok(ShutdownSignal::Terminate),
814    }
815}
816
817#[cfg(not(unix))]
818async fn shutdown_signal() -> io::Result<ShutdownSignal> {
819    tokio::signal::ctrl_c().await?;
820    Ok(ShutdownSignal::Interrupt)
821}
822
823fn prompt_from_stdin(parts: Vec<String>, read_stdin: bool) -> anyhow::Result<String> {
824    prompt_from_reader(parts, read_stdin, &mut io::stdin())
825}
826
827fn prompt_from_reader(
828    parts: Vec<String>,
829    read_stdin: bool,
830    stdin: &mut impl Read,
831) -> anyhow::Result<String> {
832    let mut chunks = Vec::new();
833    let inline = parts.join(" ").trim().to_string();
834    if !inline.is_empty() {
835        chunks.push(inline);
836    }
837    if read_stdin {
838        let mut buffer = String::new();
839        stdin.read_to_string(&mut buffer)?;
840        let buffer = buffer.trim().to_string();
841        if !buffer.is_empty() {
842            chunks.push(buffer);
843        }
844    }
845
846    let prompt = chunks.join("\n\n");
847    if prompt.is_empty() {
848        anyhow::bail!("rho run requires a prompt argument or --stdin");
849    }
850    Ok(prompt)
851}
852
853#[cfg(test)]
854#[path = "automation_tests.rs"]
855mod tests;
rho_coding_agent/app/automation.rs

rho_coding_agent/app/
automation.rs
