mobius_cli/frontend/
headless.rs

1use mobius::protocol::{
2    EventMsg, MessageAuthor, MessageSubmission, ModelStepContentPhase, Op, ReviewDecision,
3    Submission,
4};
5use mobius::{Error, Result};
6use mobius_gateway::client::{GatewayEvents, GatewaySender};
7use mobius_gateway::wire::{ClientMessage, ServerMessage};
8use uuid::Uuid;
9
10use crate::gateway_error;
11
12/// Runs one task through the headless gateway client.
13/// # Errors
14///
15/// Returns an error if validation or an operation required to complete the request fails.
16pub async fn run(
17    sender: GatewaySender,
18    mut events: GatewayEvents,
19    session_id: String,
20    task: String,
21) -> Result<Option<String>> {
22    let submission_id = Uuid::new_v4().to_string();
23    sender
24        .send(ClientMessage::Submit {
25            session_id: session_id.clone(),
26            submission: Submission {
27                id: submission_id.clone(),
28                op: Op::Message {
29                    message: MessageSubmission {
30                        author: MessageAuthor::User,
31                        text: task,
32                        attachments: Vec::new(),
33                        reply: None,
34                        requested_delivery: None,
35                        target_turn_id: None,
36                    },
37                },
38            },
39        })
40        .await
41        .map_err(gateway_error)?;
42
43    let mut turn_id = None;
44    let mut first_error = None;
45    let mut approval_error = None;
46    let mut last_assistant_message = None;
47    loop {
48        let frame =
49            events.next().await.map_err(gateway_error)?.ok_or_else(|| {
50                Error::Stopped("gateway disconnected before turn completion".into())
51            })?;
52        if let Some(error) = frame.message.response_error(Some(&submission_id)) {
53            return Err(Error::Stopped(error.message.into()));
54        }
55        let event = match frame.message {
56            ServerMessage::AgentEvent {
57                session_id: actual,
58                record,
59                ..
60            } if actual == session_id => record.event,
61            _ => continue,
62        };
63        match event.msg {
64            EventMsg::TurnStarted(turn)
65                if event.submission_id.as_deref() == Some(&submission_id) =>
66            {
67                turn_id = Some(turn.turn_id);
68            }
69            EventMsg::ExecApprovalRequest(request)
70                if turn_id.as_deref() == Some(request.turn_id.as_str()) =>
71            {
72                approval_error = Some(Error::Config(
73                    "headless run requested tool approval; configure a no-prompt gateway sandbox policy"
74                        .into(),
75                ));
76                sender
77                    .send(ClientMessage::Submit {
78                        session_id: session_id.clone(),
79                        submission: Submission {
80                            id: Uuid::new_v4().to_string(),
81                            op: Op::ExecApproval {
82                                id: request.id,
83                                decision: ReviewDecision::Abort,
84                            },
85                        },
86                    })
87                    .await
88                    .map_err(gateway_error)?;
89            }
90            EventMsg::Error(error)
91                if event.submission_id.as_deref() == Some(submission_id.as_str()) =>
92            {
93                first_error.get_or_insert(error.message);
94            }
95            EventMsg::AssistantMessage(message)
96                if turn_id.as_deref() == Some(message.turn_id.as_str()) =>
97            {
98                let final_answer = message
99                    .content
100                    .into_iter()
101                    .filter(|item| item.phase == ModelStepContentPhase::FinalAnswer)
102                    .map(|item| item.text)
103                    .collect::<String>();
104                if !final_answer.is_empty() {
105                    last_assistant_message = Some(final_answer);
106                }
107            }
108            EventMsg::TurnComplete(turn) if turn_id.as_deref() == Some(turn.turn_id.as_str()) => {
109                if let Some(error) = approval_error {
110                    return Err(error);
111                }
112                if let Some(error) = first_error {
113                    return Err(Error::Stopped(error));
114                }
115                return Ok(last_assistant_message);
116            }
117            EventMsg::TurnAborted(turn) if turn_id.as_deref() == Some(turn.turn_id.as_str()) => {
118                return Err(approval_error
119                    .unwrap_or_else(|| Error::Stopped(first_error.unwrap_or(turn.reason))));
120            }
121            _ => {}
122        }
123    }
124}
mobius_cli/frontend/headless.rs

mobius_cli/frontend/
headless.rs
