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

mobius_cli/frontend/
headless.rs
