netget/server/msgpack_rpc/
actions.rs

1use crate::llm::actions::{
2    protocol_trait::{ActionResult, Protocol, Server},
3    ActionDefinition, Parameter, ParameterDefinition,
4};
5use crate::protocol::log_template::LogTemplate;
6use crate::protocol::{EventType, SpawnContext};
7use crate::state::app_state::AppState;
8use anyhow::{bail, ensure, Result};
9use serde_json::{json, Value};
10use std::sync::LazyLock;
11
12#[derive(Default)]
13pub struct MsgpackRpcProtocol;
14impl MsgpackRpcProtocol {
15    pub fn new() -> Self {
16        Self
17    }
18}
19
20pub fn parameter(name: &str, kind: &str, description: &str, required: bool) -> Parameter {
21    Parameter {
22        name: name.into(),
23        type_hint: kind.into(),
24        description: description.into(),
25        required,
26    }
27}
28
29pub fn action(
30    name: &str,
31    description: &str,
32    parameters: Vec<Parameter>,
33    example: Value,
34) -> ActionDefinition {
35    let log_template =
36        match name {
37            "msgpack_result" => {
38                LogTemplate::new().with_info("-> MessagePack-RPC result {preview(result,100)}")
39            }
40            "msgpack_error" => {
41                LogTemplate::new().with_info("-> MessagePack-RPC error {preview(error,100)}")
42            }
43            "msgpack_notify" => LogTemplate::new()
44                .with_info("-> MessagePack-RPC notify {method} {preview(params,80)}"),
45            "msgpack_call" => LogTemplate::new()
46                .with_info("-> MessagePack-RPC call {method} {preview(params,80)}"),
47            _ => LogTemplate::new().with_info(format!("-> MessagePack-RPC {name}")),
48        };
49    ActionDefinition {
50        name: name.into(),
51        description: description.into(),
52        parameters,
53        example,
54        log_template: Some(log_template),
55    }
56}
57
58pub const VALUE_NOTE: &str =
59    "Any JSON; {\"$bin\": hex} is binary and {\"$ext\": type, \"hex\": hex} an extension";
60
61fn result_action() -> ActionDefinition {
62    action(
63        "msgpack_result",
64        "Answer the request successfully with a result value.",
65        vec![parameter("result", "any", VALUE_NOTE, true)],
66        json!({"type":"msgpack_result","result":3}),
67    )
68}
69
70fn error_action() -> ActionDefinition {
71    action(
72        "msgpack_error",
73        "Answer the request with an error value (a string, or e.g. [code, message]); the result is nil.",
74        vec![parameter("error", "any", "The error value; must not be null", true)],
75        json!({"type":"msgpack_error","error":"no such method"}),
76    )
77}
78
79pub fn notify_action() -> ActionDefinition {
80    action(
81        "msgpack_notify",
82        "Send the peer a notification [2, method, params]. May accompany an answer.",
83        vec![
84            parameter("method", "string", "Notification method name", true),
85            parameter("params", "array", "Notification parameters", true),
86        ],
87        json!({"type":"msgpack_notify","method":"progress","params":[50]}),
88    )
89}
90
91fn ignore_action() -> ActionDefinition {
92    action(
93        "msgpack_ignore",
94        "Take no action on a notification (notifications have no reply).",
95        vec![],
96        json!({"type":"msgpack_ignore"}),
97    )
98}
99
100pub static REQUEST_EVENT: LazyLock<EventType> = LazyLock::new(|| {
101    EventType::new(
102        "msgpack_request",
103        "A peer called a method; answer with a result or an error (and optionally notify it).",
104        result_action().example.clone(),
105    )
106    .with_parameters(vec![
107        parameter("method", "string", "The method the peer called", true),
108        parameter("params", "array", "The parameters, as an array", true),
109        parameter(
110            "msgid",
111            "number",
112            "Request id Rust echoes in the response",
113            true,
114        ),
115        parameter("remote_addr", "string", "Peer address and port", true),
116    ])
117    .with_actions(vec![result_action(), error_action(), notify_action()])
118});
119
120pub static NOTIFICATION_EVENT: LazyLock<EventType> = LazyLock::new(|| {
121    EventType::new(
122        "msgpack_notification",
123        "A peer sent a notification, which expects no reply.",
124        ignore_action().example.clone(),
125    )
126    .with_parameters(vec![
127        parameter("method", "string", "Notification method name", true),
128        parameter("params", "array", "The parameters, as an array", true),
129        parameter("remote_addr", "string", "Peer address and port", true),
130    ])
131    .with_actions(vec![ignore_action(), notify_action()])
132});
133
134pub fn check_value(v: &Value) -> Result<()> {
135    super::wire::encode_message(v)?;
136    Ok(())
137}
138
139pub fn check_notify(v: &Value) -> Result<()> {
140    ensure!(
141        v["method"].as_str().is_some_and(|m| !m.is_empty()),
142        "method must be a non-empty string"
143    );
144    ensure!(v["params"].is_array(), "params must be an array");
145    check_value(&v["params"])
146}
147
148impl Protocol for MsgpackRpcProtocol {
149    fn protocol_name(&self) -> &'static str {
150        "MessagePack-RPC"
151    }
152    fn stack_name(&self) -> &'static str {
153        "ETH>IP>TCP>MessagePack-RPC"
154    }
155    fn description(&self) -> &'static str {
156        "MessagePack-RPC server (the protocol Neovim speaks) whose method results the handler decides"
157    }
158    fn keywords(&self) -> Vec<&'static str> {
159        vec![
160            "msgpack-rpc",
161            "messagepack-rpc",
162            "msgpack",
163            "rpc",
164            "neovim rpc",
165        ]
166    }
167    fn get_async_actions(&self, _: &AppState) -> Vec<ActionDefinition> {
168        vec![]
169    }
170    fn get_sync_actions(&self) -> Vec<ActionDefinition> {
171        vec![
172            result_action(),
173            error_action(),
174            notify_action(),
175            ignore_action(),
176        ]
177    }
178    fn get_event_types(&self) -> Vec<EventType> {
179        vec![REQUEST_EVENT.clone(), NOTIFICATION_EVENT.clone()]
180    }
181    fn get_startup_parameters(&self) -> Vec<ParameterDefinition> {
182        vec![ParameterDefinition {
183            name: "idle_timeout_secs".into(),
184            type_hint: "number".into(),
185            description: "Seconds a connection may stay silent between messages (1..=86400)".into(),
186            required: false,
187            example: json!(600),
188            default: Some(json!(super::IDLE_TIMEOUT.as_secs())),
189        }]
190    }
191    fn metadata(&self) -> crate::protocol::metadata::ProtocolMetadataV2 {
192        use crate::protocol::metadata::*;
193        ProtocolMetadataV2::builder()
194            .state(DevelopmentState::Experimental)
195            .privilege_requirement(PrivilegeRequirement::None)
196            .implementation("MessagePack-RPC over Tokio TCP with a hand-written MessagePack codec: requests, responses, notifications; values mapped to JSON with $bin and $ext objects for binary and extensions")
197            .llm_control("The result or error of every request, and any notification sent back")
198            .e2e_testing("tests/server/msgpack_rpc: raw messages and bounds; Neovim (rpcrequest/rpcnotify over sockconnect) and ugorji/go's MsgpackSpecRpc net/rpc codec as independent clients")
199            .notes("Requests on one connection are answered in order. Messages are capped at 1 MiB and decoded only once complete, nesting at 32 levels; a connection may be silent idle_timeout_secs. A handler failure on a request answers an error value naming a generic failure, never a fabricated result.")
200            .request_only("Every response answers a request; notifications go back only as part of an answer")
201            .answers_on_failure()
202            .max_inbound_bytes(super::wire::MAX_MESSAGE)
203            .build()
204    }
205    fn example_prompt(&self) -> &'static str {
206        "MessagePack-RPC server on port 6666 with add and echo methods"
207    }
208    fn get_startup_examples(&self) -> crate::llm::actions::StartupExamples {
209        let llm = json!({"type":"open_server","base_stack":"msgpack-rpc","port":6666,"instruction":"Implement add(a, b) and echo(x); anything else is an error"});
210        let mut scripted = llm.clone();
211        scripted["event_handlers"] = json!([{"event_pattern":"msgpack_request","handler":{"type":"script","language":"python","code":"import json,sys\ne=json.load(sys.stdin)['event']\na={'type':'msgpack_result','result':sum(e['params'])} if e['method']=='add' else {'type':'msgpack_error','error':'no such method'}\nprint(json.dumps({'actions':[a]}))"}}]);
212        let mut static_example = llm.clone();
213        static_example["event_handlers"] = json!([{"event_pattern":"msgpack_request","handler":{"type":"static","actions":[{"type":"msgpack_result","result":"ok"}]}}]);
214        crate::llm::actions::StartupExamples::new(llm, scripted, static_example)
215    }
216    fn group_name(&self) -> &'static str {
217        "Application"
218    }
219}
220
221impl Server for MsgpackRpcProtocol {
222    fn spawn(
223        &self,
224        ctx: SpawnContext,
225    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::net::SocketAddr>> + Send>>
226    {
227        Box::pin(super::spawn(ctx))
228    }
229    fn execute_action(&self, v: Value) -> Result<ActionResult> {
230        let name = v["type"].as_str().unwrap_or_default().to_string();
231        match name.as_str() {
232            "msgpack_result" => check_value(&v["result"])?,
233            "msgpack_error" => {
234                ensure!(!v["error"].is_null(), "error must not be null");
235                check_value(&v["error"])?;
236            }
237            "msgpack_notify" => check_notify(&v)?,
238            "msgpack_ignore" => {}
239            _ => bail!("Unknown MessagePack-RPC server action"),
240        }
241        Ok(ActionResult::Custom { name, data: v })
242    }
243}
netget/server/msgpack_rpc/actions.rs

netget/server/msgpack_rpc/
actions.rs
