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

systemprompt_client/
remote_cli.rs
