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(request.context))
66            },
67        };
68
69        let mut builder = self
70            .client
71            .post(&self.execute_url)
72            .header("Authorization", format!("Bearer {}", request.token))
73            .header("Accept", "text/event-stream");
74
75        if !request.context.is_empty() {
76            builder = builder.header("x-context-id", request.context);
77        }
78
79        stream_response(builder.json(&body), sink).await
80    }
81}
82
83async fn stream_response(
84    builder: reqwest::RequestBuilder,
85    sink: &mut dyn OutputSink,
86) -> ClientResult<i32> {
87    let mut es = EventSource::new(builder).map_err(|_e| ClientError::EventStreamSetup)?;
88    let mut exit_code = 0;
89
90    while let Some(event) = es.next().await {
91        match event {
92            Ok(Event::Message(msg)) if msg.event == "cli" => {
93                match serde_json::from_str::<CliOutputEvent>(&msg.data) {
94                    Ok(evt) => {
95                        exit_code = dispatch_event(evt, sink, exit_code)?;
96                    },
97                    Err(e) => {
98                        tracing::warn!(error = %e, data = %msg.data, "Failed to parse CLI event");
99                    },
100                }
101            },
102            Ok(Event::Open | Event::Message(_)) => {},
103            Err(reqwest_eventsource::Error::StreamEnded) => break,
104            Err(e) => {
105                sink.error_message(&format!("Connection error: {e}"));
106                return Ok(1);
107            },
108        }
109    }
110
111    Ok(exit_code)
112}
113
114fn dispatch_event(
115    event: CliOutputEvent,
116    sink: &mut dyn OutputSink,
117    current_exit_code: i32,
118) -> ClientResult<i32> {
119    match event {
120        CliOutputEvent::Stdout { data } => {
121            sink.stdout_chunk(&data)?;
122            Ok(current_exit_code)
123        },
124        CliOutputEvent::Stderr { data } => {
125            sink.stderr_chunk(&data)?;
126            Ok(current_exit_code)
127        },
128        CliOutputEvent::ExitCode { code } => Ok(code),
129        CliOutputEvent::Error { message } => {
130            sink.error_message(&message);
131            Ok(current_exit_code)
132        },
133        CliOutputEvent::Started { pid } => {
134            tracing::debug!(pid = pid, "Remote process started");
135            Ok(current_exit_code)
136        },
137    }
138}
systemprompt_client/remote_cli.rs

systemprompt_client/
remote_cli.rs
