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

rho_coding_agent/app/
automation.rs
