rpi_cli/remote/
protocol.rs

1//! Shared wire protocol for rpi's headless RPC mode (`--mode rpc`, the server)
2//! and the remote client (`--connect`).
3//!
4//! Single source of truth: the server serializes these types and the client
5//! deserializes them, so the contract cannot drift between the two ends.
6//!
7//! Wire shape (one JSON object per line):
8//! - the server opens with `{"type":"ready"}`, then
9//! - streams [`RemoteEvent`]s (`{"type":"agent_start"}`, `{"type":"message_update",…}`, …), and
10//! - answers each [`RemoteCommand`] with a [`RemoteResponse`] (`{"type":"response",…}`).
11//!
12//! These are intentionally **not** the core `pi-agent`/`pi-ai` types: `AgentEvent`
13//! is documented as process-local (see `pi-agent/src/events.rs`) and deliberately
14//! has no serde impl. This module projects the process-local events into a
15//! stable, wire-safe shape, and translates back only the subset the client needs.
16//! That keeps the core crates untouched and the client free of local resources.
17
18use serde::{Deserialize, Serialize};
19use serde_json::Value;
20
21use rpi_agent::events::AgentEvent;
22use rpi_agent::types::AgentToolResult;
23
24/// A fine-grained lifecycle event streamed from the server to the client.
25///
26/// Mirrors the shape produced by the previous hand-rolled projection in
27/// `modes.rs`; kept field-for-field compatible so existing clients keep working.
28#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
29#[serde(tag = "type", rename_all = "snake_case")]
30pub enum RemoteEvent {
31    AgentStart,
32    AgentEnd {
33        #[serde(rename = "messageCount")]
34        message_count: usize,
35        messages: Value,
36    },
37    RetryScheduled {
38        attempt: u32,
39        #[serde(rename = "maxRetries")]
40        max_retries: u32,
41        #[serde(rename = "delayMs")]
42        delay_ms: u64,
43        error: String,
44    },
45    TurnStart,
46    TurnEnd {
47        message: Value,
48        #[serde(rename = "toolResultCount")]
49        tool_result_count: usize,
50        #[serde(rename = "toolResults")]
51        tool_results: Value,
52    },
53    MessageStart {
54        #[serde(default)]
55        message: Value,
56    },
57    MessageEnd {
58        #[serde(default)]
59        message: Value,
60    },
61    MessageUpdate {
62        #[serde(default)]
63        message: Value,
64        #[serde(rename = "assistantMessageEvent", default)]
65        assistant_message_event: Value,
66        #[serde(rename = "eventType", default)]
67        event_type: String,
68        /// Convenience mirror of the delta's content index (text/thinking/tool-call
69        /// deltas only), so the client can apply deltas without re-parsing the
70        /// nested event.
71        #[serde(rename = "contentIndex", default, skip_serializing_if = "Option::is_none")]
72        content_index: Option<u64>,
73        /// Convenience mirror of the appended text for text/thinking deltas.
74        #[serde(default, skip_serializing_if = "Option::is_none")]
75        delta: Option<String>,
76    },
77    ToolExecutionStart {
78        #[serde(rename = "toolCallId")]
79        tool_call_id: String,
80        #[serde(rename = "toolName")]
81        tool_name: String,
82        #[serde(default)]
83        args: Value,
84    },
85    ToolExecutionUpdate {
86        #[serde(rename = "toolCallId")]
87        tool_call_id: String,
88        #[serde(rename = "toolName")]
89        tool_name: String,
90        #[serde(default)]
91        args: Value,
92        #[serde(rename = "partialResult")]
93        partial_result: Value,
94    },
95    ToolExecutionEnd {
96        #[serde(rename = "toolCallId")]
97        tool_call_id: String,
98        #[serde(rename = "toolName")]
99        tool_name: String,
100        #[serde(rename = "isError")]
101        is_error: bool,
102        result: Value,
103    },
104}
105
106impl RemoteEvent {
107    /// The wire `type` tag (kept in sync with the serde `rename_all`).
108    pub fn type_tag(&self) -> &'static str {
109        match self {
110            RemoteEvent::AgentStart => "agent_start",
111            RemoteEvent::AgentEnd { .. } => "agent_end",
112            RemoteEvent::RetryScheduled { .. } => "retry_scheduled",
113            RemoteEvent::TurnStart => "turn_start",
114            RemoteEvent::TurnEnd { .. } => "turn_end",
115            RemoteEvent::MessageStart { .. } => "message_start",
116            RemoteEvent::MessageUpdate { .. } => "message_update",
117            RemoteEvent::MessageEnd { .. } => "message_end",
118            RemoteEvent::ToolExecutionStart { .. } => "tool_execution_start",
119            RemoteEvent::ToolExecutionUpdate { .. } => "tool_execution_update",
120            RemoteEvent::ToolExecutionEnd { .. } => "tool_execution_end",
121        }
122    }
123
124    /// True for the terminal `AgentEnd` event.
125    pub fn is_terminal(&self) -> bool {
126        matches!(self, RemoteEvent::AgentEnd { .. })
127    }
128
129    /// Project a process-local [`AgentEvent`] onto the wire shape.
130    pub fn from_agent_event(event: &AgentEvent) -> Self {
131        use rpi_ai::types::AssistantMessageEvent;
132        match event {
133            AgentEvent::AgentStart => RemoteEvent::AgentStart,
134            AgentEvent::AgentEnd { messages } => RemoteEvent::AgentEnd {
135                message_count: messages.len(),
136                messages: serde_json::to_value(messages).unwrap_or(Value::Null),
137            },
138            AgentEvent::RetryScheduled {
139                attempt,
140                max_retries,
141                delay_ms,
142                error,
143            } => RemoteEvent::RetryScheduled {
144                attempt: *attempt,
145                max_retries: *max_retries,
146                delay_ms: *delay_ms,
147                error: error.clone(),
148            },
149            AgentEvent::TurnStart => RemoteEvent::TurnStart,
150            AgentEvent::TurnEnd {
151                message,
152                tool_results,
153            } => RemoteEvent::TurnEnd {
154                message: serde_json::to_value(message).unwrap_or(Value::Null),
155                tool_result_count: tool_results.len(),
156                tool_results: serde_json::to_value(tool_results).unwrap_or(Value::Null),
157            },
158            AgentEvent::MessageStart { message } => RemoteEvent::MessageStart {
159                message: serde_json::to_value(message).unwrap_or(Value::Null),
160            },
161            AgentEvent::MessageEnd { message } => RemoteEvent::MessageEnd {
162                message: serde_json::to_value(message).unwrap_or(Value::Null),
163            },
164            AgentEvent::MessageUpdate {
165                message,
166                assistant_message_event,
167            } => {
168                let (content_index, delta) = match assistant_message_event {
169                    AssistantMessageEvent::TextDelta {
170                        content_index,
171                        delta,
172                        ..
173                    }
174                    | AssistantMessageEvent::ThinkingDelta {
175                        content_index,
176                        delta,
177                        ..
178                    }
179                    | AssistantMessageEvent::ToolCallDelta {
180                        content_index,
181                        delta,
182                        ..
183                    } => (Some(*content_index as u64), Some(delta.clone())),
184                    _ => (None, None),
185                };
186                RemoteEvent::MessageUpdate {
187                    message: serde_json::to_value(message).unwrap_or(Value::Null),
188                    assistant_message_event: serde_json::to_value(assistant_message_event)
189                        .unwrap_or(Value::Null),
190                    event_type: assistant_message_event.type_tag().to_string(),
191                    content_index,
192                    delta,
193                }
194            }
195            AgentEvent::ToolExecutionStart {
196                tool_call_id,
197                tool_name,
198                args,
199            } => RemoteEvent::ToolExecutionStart {
200                tool_call_id: tool_call_id.clone(),
201                tool_name: tool_name.clone(),
202                args: args.clone(),
203            },
204            AgentEvent::ToolExecutionUpdate {
205                tool_call_id,
206                tool_name,
207                args,
208                partial_result,
209            } => RemoteEvent::ToolExecutionUpdate {
210                tool_call_id: tool_call_id.clone(),
211                tool_name: tool_name.clone(),
212                args: args.clone(),
213                partial_result: tool_result_json(partial_result),
214            },
215            AgentEvent::ToolExecutionEnd {
216                tool_call_id,
217                tool_name,
218                result,
219                is_error,
220            } => RemoteEvent::ToolExecutionEnd {
221                tool_call_id: tool_call_id.clone(),
222                tool_name: tool_name.clone(),
223                is_error: *is_error,
224                result: tool_result_json(result),
225            },
226        }
227    }
228}
229
230/// Serialize a tool result into the stable wire shape (content blocks, details,
231/// usage, and the batch hints).
232pub fn tool_result_json(result: &AgentToolResult) -> Value {
233    let content: Vec<Value> = result
234        .content
235        .iter()
236        .map(|item| match item {
237            rpi_agent::types::TextContentOrImage::Text(text) => serde_json::json!({
238                "type": "text",
239                "text": text.text,
240            }),
241            rpi_agent::types::TextContentOrImage::Image(image) => {
242                serde_json::to_value(image).unwrap_or(Value::Null)
243            }
244        })
245        .collect();
246    serde_json::json!({
247        "content": content,
248        "details": result.details,
249        "usage": result.usage.as_ref().and_then(|usage| serde_json::to_value(usage).ok()),
250        "addedToolNames": result.added_tool_names,
251        "terminate": result.terminate,
252    })
253}
254
255/// A command sent from the client to the server (`{"id":…,"type":…,…}`).
256///
257/// The `id` is attached by the client when sending (see
258/// [`with_id`](RemoteCommand::with_id)) so responses can be correlated.
259#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
260#[serde(tag = "type", rename_all = "snake_case")]
261pub enum RemoteCommand {
262    /// Run a prompt on the remote `main` lane; events stream until `agent_end`.
263    Prompt { content: String },
264    /// Abort the in-flight run.
265    Abort,
266    /// Fetch the current session state (model, thinking, tools, ids).
267    GetState,
268    /// Switch the active model by id.
269    SetModel { model: String },
270    /// Set the thinking level (`off`…`max`).
271    SetThinkingLevel { level: String },
272    /// Replace the active tool set.
273    SetActiveTools { tools: Vec<String> },
274    /// Liveness probe.
275    Ping,
276    /// Ask the server to end the session loop.
277    Stop,
278}
279
280impl RemoteCommand {
281    /// Serialize the command with an `id` field for request/response correlation.
282    pub fn with_id(&self, id: impl Into<Value>) -> Value {
283        let mut value = serde_json::to_value(self).unwrap_or(Value::Null);
284        if let Some(object) = value.as_object_mut() {
285            object.insert("id".to_string(), id.into());
286        }
287        value
288    }
289}
290
291/// Response status for a [`RemoteResponse`].
292#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
293#[serde(rename_all = "lowercase")]
294pub enum ResponseStatus {
295    Ok,
296    Error,
297}
298
299/// The server's answer to a single [`RemoteCommand`].
300#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
301pub struct RemoteResponse {
302    #[serde(default)]
303    pub id: Value,
304    pub status: ResponseStatus,
305    #[serde(default, skip_serializing_if = "Option::is_none")]
306    pub result: Option<Value>,
307    #[serde(default, skip_serializing_if = "Option::is_none")]
308    pub error: Option<String>,
309}
310
311impl RemoteResponse {
312    /// A successful response carrying an optional result payload.
313    pub fn ok(id: Value, result: Value) -> Self {
314        Self {
315            id,
316            status: ResponseStatus::Ok,
317            result: Some(result),
318            error: None,
319        }
320    }
321
322    /// A failed response carrying an error message.
323    pub fn error(id: Value, error: impl Into<String>) -> Self {
324        Self {
325            id,
326            status: ResponseStatus::Error,
327            result: None,
328            error: Some(error.into()),
329        }
330    }
331
332    /// True when the request succeeded.
333    pub fn is_ok(&self) -> bool {
334        matches!(self.status, ResponseStatus::Ok)
335    }
336
337    /// Serialize to the wire shape, adding the `"type":"response"` tag.
338    pub fn to_json(&self) -> Value {
339        let mut value = serde_json::to_value(self).unwrap_or(Value::Null);
340        if let Some(object) = value.as_object_mut() {
341            object.insert("type".to_string(), Value::String("response".to_string()));
342        }
343        value
344    }
345}
346
347/// The `get_state` result payload, as a typed view over the response.
348#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
349pub struct SessionState {
350    #[serde(default)]
351    pub model: Option<String>,
352    #[serde(rename = "thinkingLevel", default)]
353    pub thinking_level: Option<Value>,
354    #[serde(rename = "activeTools", default)]
355    pub active_tools: Vec<String>,
356    #[serde(rename = "leafId", default)]
357    pub leaf_id: Option<String>,
358    #[serde(rename = "sessionId", default)]
359    pub session_id: Option<String>,
360}
361
362/// The `prompt` result payload (`{outcome, finalText}`).
363#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
364pub struct PromptOutcome {
365    #[serde(default)]
366    pub outcome: Option<String>,
367    #[serde(rename = "finalText", default)]
368    pub final_text: Option<String>,
369}
370
371#[cfg(test)]
372mod tests {
373    use super::*;
374    use rpi_agent::types::AgentToolResult;
375
376    #[test]
377    fn event_projection_matches_wire_shape() {
378        let end = serde_json::to_value(RemoteEvent::from_agent_event(&AgentEvent::AgentEnd {
379            messages: vec![],
380        }))
381        .unwrap();
382        assert_eq!(end["type"], "agent_end");
383        assert_eq!(end["messages"], serde_json::json!([]));
384        assert_eq!(end["messageCount"], 0);
385
386        let tool = serde_json::to_value(RemoteEvent::from_agent_event(
387            &AgentEvent::ToolExecutionEnd {
388                tool_call_id: "call-1".into(),
389                tool_name: "read".into(),
390                result: AgentToolResult::text("hello"),
391                is_error: false,
392            },
393        ))
394        .unwrap();
395        assert_eq!(tool["type"], "tool_execution_end");
396        assert_eq!(tool["result"]["content"][0]["text"], "hello");
397        assert_eq!(tool["result"]["terminate"], false);
398
399        let retry = serde_json::to_value(RemoteEvent::from_agent_event(
400            &AgentEvent::RetryScheduled {
401                attempt: 3,
402                max_retries: 10,
403                delay_ms: 8_000,
404                error: "503 service unavailable".into(),
405            },
406        ))
407        .unwrap();
408        assert_eq!(retry["type"], "retry_scheduled");
409        assert_eq!(retry["attempt"], 3);
410        assert_eq!(retry["maxRetries"], 10);
411        assert_eq!(retry["delayMs"], 8_000);
412    }
413
414    #[test]
415    fn events_round_trip_over_the_wire() {
416        for event in [
417            RemoteEvent::AgentStart,
418            RemoteEvent::MessageUpdate {
419                message: serde_json::json!({"role": "assistant"}),
420                assistant_message_event: serde_json::json!({"type": "text_delta"}),
421                event_type: "text_delta".into(),
422                content_index: Some(0),
423                delta: Some("hi".into()),
424            },
425            RemoteEvent::ToolExecutionEnd {
426                tool_call_id: "c1".into(),
427                tool_name: "read".into(),
428                is_error: true,
429                result: tool_result_json(&AgentToolResult::text("nope")),
430            },
431        ] {
432            let json = serde_json::to_value(&event).unwrap();
433            let back: RemoteEvent = serde_json::from_value(json).unwrap();
434            assert_eq!(event, back);
435        }
436    }
437
438    #[test]
439    fn command_and_response_round_trip() {
440        let cmd = RemoteCommand::Prompt {
441            content: "hi".into(),
442        };
443        let value = cmd.with_id("r1");
444        assert_eq!(value["type"], "prompt");
445        assert_eq!(value["content"], "hi");
446        assert_eq!(value["id"], "r1");
447
448        let response = RemoteResponse::ok(serde_json::json!("r1"), serde_json::json!({"pong": true}));
449        let json = response.to_json();
450        assert_eq!(json["type"], "response");
451        assert_eq!(json["id"], "r1");
452        assert_eq!(json["status"], "ok");
453        let parsed: RemoteResponse = serde_json::from_value(json).unwrap();
454        assert!(parsed.is_ok());
455
456        let err = RemoteResponse::error(Value::Null, "boom").to_json();
457        assert_eq!(err["status"], "error");
458        assert_eq!(err["error"], "boom");
459    }
460}
rpi_cli/remote/protocol.rs

rpi_cli/remote/
protocol.rs
