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                    // `assistant_json` keeps this identical to the
188                    // `AgentMessage` encoding without copying the message.
189                    message: rpi_agent::message::assistant_json(message),
190                    assistant_message_event: serde_json::to_value(assistant_message_event)
191                        .unwrap_or(Value::Null),
192                    event_type: assistant_message_event.type_tag().to_string(),
193                    content_index,
194                    delta,
195                }
196            }
197            AgentEvent::ToolExecutionStart {
198                tool_call_id,
199                tool_name,
200                args,
201            } => RemoteEvent::ToolExecutionStart {
202                tool_call_id: tool_call_id.clone(),
203                tool_name: tool_name.clone(),
204                args: args.clone(),
205            },
206            AgentEvent::ToolExecutionUpdate {
207                tool_call_id,
208                tool_name,
209                args,
210                partial_result,
211            } => RemoteEvent::ToolExecutionUpdate {
212                tool_call_id: tool_call_id.clone(),
213                tool_name: tool_name.clone(),
214                args: args.clone(),
215                partial_result: tool_result_json(partial_result),
216            },
217            AgentEvent::ToolExecutionEnd {
218                tool_call_id,
219                tool_name,
220                result,
221                is_error,
222            } => RemoteEvent::ToolExecutionEnd {
223                tool_call_id: tool_call_id.clone(),
224                tool_name: tool_name.clone(),
225                is_error: *is_error,
226                result: tool_result_json(result),
227            },
228        }
229    }
230}
231
232/// Serialize a tool result into the stable wire shape (content blocks, details,
233/// usage, and the batch hints).
234pub fn tool_result_json(result: &AgentToolResult) -> Value {
235    let content: Vec<Value> = result
236        .content
237        .iter()
238        .map(|item| match item {
239            rpi_agent::types::TextContentOrImage::Text(text) => serde_json::json!({
240                "type": "text",
241                "text": text.text,
242            }),
243            rpi_agent::types::TextContentOrImage::Image(image) => {
244                serde_json::to_value(image).unwrap_or(Value::Null)
245            }
246        })
247        .collect();
248    serde_json::json!({
249        "content": content,
250        "details": result.details,
251        "usage": result.usage.as_ref().and_then(|usage| serde_json::to_value(usage).ok()),
252        "addedToolNames": result.added_tool_names,
253        "terminate": result.terminate,
254    })
255}
256
257/// A command sent from the client to the server (`{"id":…,"type":…,…}`).
258///
259/// The `id` is attached by the client when sending (see
260/// [`with_id`](RemoteCommand::with_id)) so responses can be correlated.
261#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
262#[serde(tag = "type", rename_all = "snake_case")]
263pub enum RemoteCommand {
264    /// Run a prompt on the remote `main` lane; events stream until `agent_end`.
265    Prompt { content: String },
266    /// Abort the in-flight run.
267    Abort,
268    /// Fetch the current session state (model, thinking, tools, ids).
269    GetState,
270    /// Switch the active model by id.
271    SetModel { model: String },
272    /// Set the thinking level (`off`…`max`).
273    SetThinkingLevel { level: String },
274    /// Replace the active tool set.
275    SetActiveTools { tools: Vec<String> },
276    /// Liveness probe.
277    Ping,
278    /// Ask the server to end the session loop.
279    Stop,
280}
281
282impl RemoteCommand {
283    /// Serialize the command with an `id` field for request/response correlation.
284    pub fn with_id(&self, id: impl Into<Value>) -> Value {
285        let mut value = serde_json::to_value(self).unwrap_or(Value::Null);
286        if let Some(object) = value.as_object_mut() {
287            object.insert("id".to_string(), id.into());
288        }
289        value
290    }
291}
292
293/// Response status for a [`RemoteResponse`].
294#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
295#[serde(rename_all = "lowercase")]
296pub enum ResponseStatus {
297    Ok,
298    Error,
299}
300
301/// The server's answer to a single [`RemoteCommand`].
302#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
303pub struct RemoteResponse {
304    #[serde(default)]
305    pub id: Value,
306    pub status: ResponseStatus,
307    #[serde(default, skip_serializing_if = "Option::is_none")]
308    pub result: Option<Value>,
309    #[serde(default, skip_serializing_if = "Option::is_none")]
310    pub error: Option<String>,
311}
312
313impl RemoteResponse {
314    /// A successful response carrying an optional result payload.
315    pub fn ok(id: Value, result: Value) -> Self {
316        Self {
317            id,
318            status: ResponseStatus::Ok,
319            result: Some(result),
320            error: None,
321        }
322    }
323
324    /// A failed response carrying an error message.
325    pub fn error(id: Value, error: impl Into<String>) -> Self {
326        Self {
327            id,
328            status: ResponseStatus::Error,
329            result: None,
330            error: Some(error.into()),
331        }
332    }
333
334    /// True when the request succeeded.
335    pub fn is_ok(&self) -> bool {
336        matches!(self.status, ResponseStatus::Ok)
337    }
338
339    /// Serialize to the wire shape, adding the `"type":"response"` tag.
340    pub fn to_json(&self) -> Value {
341        let mut value = serde_json::to_value(self).unwrap_or(Value::Null);
342        if let Some(object) = value.as_object_mut() {
343            object.insert("type".to_string(), Value::String("response".to_string()));
344        }
345        value
346    }
347}
348
349/// The `get_state` result payload, as a typed view over the response.
350#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
351pub struct SessionState {
352    #[serde(default)]
353    pub model: Option<String>,
354    #[serde(rename = "thinkingLevel", default)]
355    pub thinking_level: Option<Value>,
356    #[serde(rename = "activeTools", default)]
357    pub active_tools: Vec<String>,
358    #[serde(rename = "leafId", default)]
359    pub leaf_id: Option<String>,
360    #[serde(rename = "sessionId", default)]
361    pub session_id: Option<String>,
362}
363
364/// The `prompt` result payload (`{outcome, finalText}`).
365#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
366pub struct PromptOutcome {
367    #[serde(default)]
368    pub outcome: Option<String>,
369    #[serde(rename = "finalText", default)]
370    pub final_text: Option<String>,
371}
372
373#[cfg(test)]
374mod tests {
375    use super::*;
376    use rpi_agent::types::AgentToolResult;
377
378    #[test]
379    fn event_projection_matches_wire_shape() {
380        let end = serde_json::to_value(RemoteEvent::from_agent_event(&AgentEvent::AgentEnd {
381            messages: vec![],
382        }))
383        .unwrap();
384        assert_eq!(end["type"], "agent_end");
385        assert_eq!(end["messages"], serde_json::json!([]));
386        assert_eq!(end["messageCount"], 0);
387
388        let tool = serde_json::to_value(RemoteEvent::from_agent_event(
389            &AgentEvent::ToolExecutionEnd {
390                tool_call_id: "call-1".into(),
391                tool_name: "read".into(),
392                result: AgentToolResult::text("hello"),
393                is_error: false,
394            },
395        ))
396        .unwrap();
397        assert_eq!(tool["type"], "tool_execution_end");
398        assert_eq!(tool["result"]["content"][0]["text"], "hello");
399        assert_eq!(tool["result"]["terminate"], false);
400
401        let retry = serde_json::to_value(RemoteEvent::from_agent_event(
402            &AgentEvent::RetryScheduled {
403                attempt: 3,
404                max_retries: 10,
405                delay_ms: 8_000,
406                error: "503 service unavailable".into(),
407            },
408        ))
409        .unwrap();
410        assert_eq!(retry["type"], "retry_scheduled");
411        assert_eq!(retry["attempt"], 3);
412        assert_eq!(retry["maxRetries"], 10);
413        assert_eq!(retry["delayMs"], 8_000);
414    }
415
416    #[test]
417    fn events_round_trip_over_the_wire() {
418        for event in [
419            RemoteEvent::AgentStart,
420            RemoteEvent::MessageUpdate {
421                message: serde_json::json!({"role": "assistant"}),
422                assistant_message_event: serde_json::json!({"type": "text_delta"}),
423                event_type: "text_delta".into(),
424                content_index: Some(0),
425                delta: Some("hi".into()),
426            },
427            RemoteEvent::ToolExecutionEnd {
428                tool_call_id: "c1".into(),
429                tool_name: "read".into(),
430                is_error: true,
431                result: tool_result_json(&AgentToolResult::text("nope")),
432            },
433        ] {
434            let json = serde_json::to_value(&event).unwrap();
435            let back: RemoteEvent = serde_json::from_value(json).unwrap();
436            assert_eq!(event, back);
437        }
438    }
439
440    #[test]
441    fn command_and_response_round_trip() {
442        let cmd = RemoteCommand::Prompt {
443            content: "hi".into(),
444        };
445        let value = cmd.with_id("r1");
446        assert_eq!(value["type"], "prompt");
447        assert_eq!(value["content"], "hi");
448        assert_eq!(value["id"], "r1");
449
450        let response = RemoteResponse::ok(serde_json::json!("r1"), serde_json::json!({"pong": true}));
451        let json = response.to_json();
452        assert_eq!(json["type"], "response");
453        assert_eq!(json["id"], "r1");
454        assert_eq!(json["status"], "ok");
455        let parsed: RemoteResponse = serde_json::from_value(json).unwrap();
456        assert!(parsed.is_ok());
457
458        let err = RemoteResponse::error(Value::Null, "boom").to_json();
459        assert_eq!(err["status"], "error");
460        assert_eq!(err["error"], "boom");
461    }
462}
rpi_cli/remote/protocol.rs

rpi_cli/remote/
protocol.rs
