horus_cli/frontend/
headless.rs

1use horus::protocol::{EventMsg, Op, ReviewDecision, Submission};
2use horus::{Error, Result};
3use horus_gateway::client::{GatewayEvents, GatewaySender};
4use horus_gateway::wire::{ClientMessage, ServerMessage};
5use uuid::Uuid;
6
7pub async fn run(
8    sender: GatewaySender,
9    mut events: GatewayEvents,
10    session_id: String,
11    task: String,
12) -> Result<Option<String>> {
13    let submission_id = Uuid::new_v4().to_string();
14    sender
15        .send(ClientMessage::Submit {
16            session_id: session_id.clone(),
17            submission: Submission {
18                id: submission_id.clone(),
19                op: Op::UserInput { text: task },
20            },
21        })
22        .await
23        .map_err(gateway_error)?;
24
25    let mut turn_id = None;
26    let mut first_error = None;
27    let mut approval_error = None;
28    loop {
29        let frame =
30            events.next().await.map_err(gateway_error)?.ok_or_else(|| {
31                Error::Stopped("gateway disconnected before turn completion".into())
32            })?;
33        let event = match frame.message {
34            ServerMessage::AgentEvent {
35                session_id: actual,
36                event,
37                ..
38            } if actual == session_id => event,
39            ServerMessage::Rejected {
40                request_id,
41                message,
42                ..
43            } if request_id == submission_id => return Err(Error::Stopped(message)),
44            ServerMessage::Error { message, .. } => return Err(Error::Stopped(message)),
45            _ => continue,
46        };
47        match event.msg {
48            EventMsg::TurnStarted(turn)
49                if event.submission_id.as_deref() == Some(&submission_id) =>
50            {
51                turn_id = Some(turn.turn_id);
52            }
53            EventMsg::ExecApprovalRequest(request)
54                if turn_id.as_deref() == Some(request.turn_id.as_str()) =>
55            {
56                approval_error = Some(Error::Config(
57                    "headless run requested tool approval; configure a no-prompt gateway sandbox policy"
58                        .into(),
59                ));
60                sender
61                    .send(ClientMessage::Submit {
62                        session_id: session_id.clone(),
63                        submission: Submission {
64                            id: Uuid::new_v4().to_string(),
65                            op: Op::ExecApproval {
66                                id: request.id,
67                                decision: ReviewDecision::Abort,
68                            },
69                        },
70                    })
71                    .await
72                    .map_err(gateway_error)?;
73            }
74            EventMsg::Error(error)
75                if event.submission_id.as_deref() == Some(submission_id.as_str()) =>
76            {
77                first_error.get_or_insert(error.message);
78            }
79            EventMsg::TurnComplete(turn) if turn_id.as_deref() == Some(turn.turn_id.as_str()) => {
80                if let Some(error) = approval_error {
81                    return Err(error);
82                }
83                if let Some(error) = first_error {
84                    return Err(Error::Stopped(error));
85                }
86                return Ok(turn.last_agent_message);
87            }
88            EventMsg::TurnAborted(turn) if turn_id.as_deref() == Some(turn.turn_id.as_str()) => {
89                return Err(approval_error
90                    .unwrap_or_else(|| Error::Stopped(first_error.unwrap_or(turn.reason))));
91            }
92            _ => {}
93        }
94    }
95}
96
97fn gateway_error(error: horus_gateway::Error) -> Error {
98    Error::Stopped(error.to_string())
99}
horus_cli/frontend/headless.rs

horus_cli/frontend/
headless.rs
