systemprompt_client/
remote_cli.rs

1//! Remote CLI execution over a deployment's SSE gateway.
2//!
3//! [`RemoteCliExecutor`] POSTs a [`CliExecuteRequest`] to
4//! `/api/v1/admin/cli` and streams the resulting `cli` server-sent events
5//! back through a caller-supplied [`OutputSink`], returning the remote
6//! process's exit code. Transport failures mid-stream are reported through
7//! the sink and surface as exit code `1` rather than an error, matching an
8//! interactive terminal session; only setup failures return [`ClientError`].
9
10use std::io;
11use std::time::Duration;
12
13use futures::StreamExt;
14use reqwest_eventsource::{Event, EventSource};
15use systemprompt_models::api::{CliExecuteRequest, CliOutputEvent};
16
17use crate::error::{ClientError, ClientResult};
18
19pub trait OutputSink: Send {
20    fn stdout_chunk(&mut self, data: &str) -> io::Result<()>;
21    fn stderr_chunk(&mut self, data: &str) -> io::Result<()>;
22    fn error_message(&mut self, message: &str);
23}
24
25#[derive(Debug, Clone, Copy)]
26pub struct RemoteCliRequest<'a> {
27    pub token: &'a str,
28    pub context: &'a str,
29    pub args: &'a [String],
30}
31
32#[derive(Debug, Clone)]
33pub struct RemoteCliExecutor {
34    client: reqwest::Client,
35    execute_url: String,
36    timeout_secs: u64,
37}
38
39impl RemoteCliExecutor {
40    pub fn new(base_url: &str, timeout_secs: u64) -> ClientResult<Self> {
41        let client = reqwest::Client::builder()
42            .timeout(Duration::from_secs(timeout_secs + 30))
43            .build()?;
44        Ok(Self {
45            client,
46            execute_url: format!("{base_url}/api/v1/admin/cli"),
47            timeout_secs,
48        })
49    }
50
51    pub async fn execute(
52        &self,
53        request: RemoteCliRequest<'_>,
54        sink: &mut dyn OutputSink,
55    ) -> ClientResult<i32> {
56        let body = CliExecuteRequest {
57            args: request.args.to_vec(),
58            timeout_secs: self.timeout_secs,
59            context_id: if request.context.is_empty() {
60                None
61            } else {
62                Some(systemprompt_identifiers::ContextId::new(request.context))
63            },
64        };
65
66        let mut builder = self
67            .client
68            .post(&self.execute_url)
69            .header("Authorization", format!("Bearer {}", request.token))
70            .header("Accept", "text/event-stream");
71
72        if !request.context.is_empty() {
73            builder = builder.header("x-context-id", request.context);
74        }
75
76        stream_response(builder.json(&body), sink).await
77    }
78}
79
80async fn stream_response(
81    builder: reqwest::RequestBuilder,
82    sink: &mut dyn OutputSink,
83) -> ClientResult<i32> {
84    let mut es = EventSource::new(builder).map_err(|_e| ClientError::EventStreamSetup)?;
85    let mut exit_code = 0;
86
87    while let Some(event) = es.next().await {
88        match event {
89            Ok(Event::Message(msg)) if msg.event == "cli" => {
90                match serde_json::from_str::<CliOutputEvent>(&msg.data) {
91                    Ok(evt) => {
92                        exit_code = dispatch_event(evt, sink, exit_code)?;
93                    },
94                    Err(e) => {
95                        tracing::warn!(error = %e, data = %msg.data, "Failed to parse CLI event");
96                    },
97                }
98            },
99            Ok(Event::Open | Event::Message(_)) => {},
100            Err(reqwest_eventsource::Error::StreamEnded) => break,
101            Err(e) => {
102                sink.error_message(&format!("Connection error: {e}"));
103                return Ok(1);
104            },
105        }
106    }
107
108    Ok(exit_code)
109}
110
111fn dispatch_event(
112    event: CliOutputEvent,
113    sink: &mut dyn OutputSink,
114    current_exit_code: i32,
115) -> ClientResult<i32> {
116    match event {
117        CliOutputEvent::Stdout { data } => {
118            sink.stdout_chunk(&data)?;
119            Ok(current_exit_code)
120        },
121        CliOutputEvent::Stderr { data } => {
122            sink.stderr_chunk(&data)?;
123            Ok(current_exit_code)
124        },
125        CliOutputEvent::ExitCode { code } => Ok(code),
126        CliOutputEvent::Error { message } => {
127            sink.error_message(&message);
128            Ok(current_exit_code)
129        },
130        CliOutputEvent::Started { pid } => {
131            tracing::debug!(pid = pid, "Remote process started");
132            Ok(current_exit_code)
133        },
134    }
135}
systemprompt_client/remote_cli.rs

systemprompt_client/
remote_cli.rs
