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

horus_cli/frontend/
headless.rs
