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

rpi_cli/remote/
protocol.rs
