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

mobius_cli/frontend/
headless.rs
