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

rpi_cli/remote/
protocol.rs
