netget/server/capnp_rpc/
actions.rs

1use crate::llm::actions::{
2    protocol_trait::{ActionResult, Protocol, Server},
3    ActionDefinition, Parameter, ParameterDefinition, StartupExamples,
4};
5use crate::protocol::log_template::LogTemplate;
6use crate::protocol::{EventType, SpawnContext};
7use crate::state::app_state::AppState;
8use anyhow::{bail, ensure, Context, Result};
9use serde_json::{json, Value};
10use std::sync::LazyLock;
11
12#[derive(Default)]
13pub struct CapnpRpcProtocol;
14impl CapnpRpcProtocol {
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    log: &str,
35) -> ActionDefinition {
36    ActionDefinition {
37        name: name.into(),
38        description: description.into(),
39        parameters,
40        example,
41        log_template: Some(LogTemplate::new().with_info(log)),
42    }
43}
44
45pub const VALUE_NOTE: &str = "Fields by name as the schema declares them: numbers, booleans, strings, enums by enumerant name, lists as arrays, structs and groups as objects, a union as its one member, Data as {\"$hex\": \"…\"}";
46
47fn return_action() -> ActionDefinition {
48    action(
49        "capnp_return",
50        "Answer the call with its results struct; the event's results_shape lists the fields.",
51        vec![parameter("results", "object", VALUE_NOTE, true)],
52        json!({"type":"capnp_return","results":{"sum":5}}),
53        "-> Cap'n Proto return {preview(results,100)}",
54    )
55}
56
57pub fn exception_action() -> ActionDefinition {
58    action(
59        "capnp_exception",
60        "Answer the call with an exception instead of results.",
61        vec![
62            parameter(
63                "reason",
64                "string",
65                "Human-readable reason, at most 1024 bytes",
66                true,
67            ),
68            parameter(
69                "kind",
70                "string",
71                "failed (default), overloaded or unimplemented",
72                false,
73            ),
74        ],
75        json!({"type":"capnp_exception","reason":"no such entry","kind":"failed"}),
76        "-> Cap'n Proto exception {reason}",
77    )
78}
79
80pub static CALL_EVENT: LazyLock<EventType> = LazyLock::new(|| {
81    EventType::new(
82        "capnp_call",
83        "A peer called a method on the bootstrap capability. Answer with its results or an exception.",
84        return_action().example.clone(),
85    )
86    .with_parameters(vec![
87        parameter("interface", "string", "The interface the method belongs to", true),
88        parameter("method", "string", "The method's name", true),
89        parameter("params", "object", VALUE_NOTE, true),
90        parameter(
91            "results_shape",
92            "string",
93            "The fields capnp_return's results may set, with their types",
94            true,
95        ),
96        parameter("remote_addr", "string", "Peer address and port", true),
97    ])
98    .with_actions(vec![return_action(), exception_action()])
99});
100
101/// Check an exception action's fields; returns (reason, type code).
102pub fn check_exception(v: &Value) -> Result<(String, u16)> {
103    let reason = v["reason"].as_str().context("reason required")?;
104    ensure!(
105        !reason.is_empty() && reason.len() <= 1024 && !crate::utils::sanitize::has_controls(reason),
106        "reason must be 1..=1024 bytes without control characters"
107    );
108    let kind = match v.get("kind").and_then(Value::as_str) {
109        None => super::rpc::EXC_FAILED,
110        Some(name) => super::rpc::exception_type(name)
111            .filter(|t| *t != 2)
112            .context("kind must be failed, overloaded or unimplemented")?,
113    };
114    Ok((reason.to_string(), kind))
115}
116
117fn param(
118    name: &str,
119    kind: &str,
120    description: &str,
121    required: bool,
122    example: Value,
123    default: Option<Value>,
124) -> ParameterDefinition {
125    ParameterDefinition {
126        name: name.into(),
127        type_hint: kind.into(),
128        description: description.into(),
129        required,
130        example,
131        default,
132    }
133}
134
135/// A small inline schema for the examples.
136pub const EXAMPLE_SCHEMA: &str = "interface Directory {\n  lookup @0 (name :Text) -> (size :UInt64, kind :Text);\n  add @1 (a :Int32, b :Int32) -> (sum :Int64);\n}";
137
138pub fn schema_parameters() -> Vec<ParameterDefinition> {
139    vec![
140        param(
141            "schema",
142            "string",
143            "The interface's Cap'n Proto schema: inline source (a file id is added if absent), a path to a .capnp file (both compiled at startup with the capnp tool), or a path to its compiled form (`capnp compile -o- file.capnp > file.bin`, which needs no tool)",
144            true,
145            json!(EXAMPLE_SCHEMA),
146            None,
147        ),
148        param(
149            "interface",
150            "string",
151            "Name of the interface the bootstrap capability implements, e.g. Directory",
152            true,
153            json!("Directory"),
154            None,
155        ),
156    ]
157}
158
159impl Protocol for CapnpRpcProtocol {
160    fn protocol_name(&self) -> &'static str {
161        "Cap'n Proto RPC"
162    }
163    fn stack_name(&self) -> &'static str {
164        "ETH>IP>TCP>CapnProtoRPC"
165    }
166    fn description(&self) -> &'static str {
167        "Cap'n Proto RPC server: exports one capability whose method results the handler decides, typed by a startup schema"
168    }
169    fn keywords(&self) -> Vec<&'static str> {
170        vec![
171            "capnp",
172            "cap'n proto",
173            "capnproto",
174            "capnp-rpc",
175            "capability rpc",
176        ]
177    }
178    fn get_async_actions(&self, _: &AppState) -> Vec<ActionDefinition> {
179        vec![]
180    }
181    fn get_sync_actions(&self) -> Vec<ActionDefinition> {
182        vec![return_action(), exception_action()]
183    }
184    fn get_event_types(&self) -> Vec<EventType> {
185        vec![CALL_EVENT.clone()]
186    }
187    fn get_startup_parameters(&self) -> Vec<ParameterDefinition> {
188        let mut p = schema_parameters();
189        p.push(param(
190            "idle_timeout_secs",
191            "number",
192            "Seconds a connection may stay silent between messages (1..=86400)",
193            false,
194            json!(600),
195            Some(json!(super::IDLE_TIMEOUT.as_secs())),
196        ));
197        p
198    }
199    fn get_dependencies(&self) -> Vec<crate::protocol::dependencies::ProtocolDependency> {
200        let mut deps =
201            crate::llm::actions::protocol_trait::default_dependencies_from_privilege(self);
202        deps.push(crate::protocol::dependencies::ProtocolDependency::ToolInPath("capnp"));
203        deps
204    }
205    fn startup_dependencies(
206        &self,
207        startup_params: Option<&Value>,
208    ) -> Vec<crate::protocol::dependencies::ProtocolDependency> {
209        let mut deps = self.get_dependencies();
210        if !super::needs_compiler(startup_params) {
211            deps.retain(|d| {
212                *d != crate::protocol::dependencies::ProtocolDependency::ToolInPath("capnp")
213            });
214        }
215        deps
216    }
217    fn metadata(&self) -> crate::protocol::metadata::ProtocolMetadataV2 {
218        use crate::protocol::metadata::*;
219        ProtocolMetadataV2::builder()
220            .state(DevelopmentState::Experimental)
221            .privilege_requirement(PrivilegeRequirement::None)
222            .implementation("Hand-written Cap'n Proto encoding (bounded pointer-following reader, single-segment builder) and RPC level 1 over Tokio TCP: Bootstrap, Call (on the bootstrap capability or its promised answer), Return, Finish, Release, Abort, Unimplemented; params and results mapped to JSON by a schema loaded at startup")
223            .llm_control("The results or exception of every call")
224            .e2e_testing("tests/server/capnp_rpc: raw messages, the codec and bounds; pycapnp (the C++ Cap'n Proto runtime) and capnproto.org/go/capnp v3 as independent clients")
225            .notes("One capability is exported (the bootstrap); results carry no capabilities, so promise pipelining reaches only the bootstrap answer. Interface and AnyPointer fields are not mapped. Messages are capped at 4 MiB and 64 segments, nesting at 64, traversal at four times the message size. A handler failure answers an exception, never fabricated results.")
226            .request_only("Every Return answers a Bootstrap or a Call")
227            .answers_on_failure()
228            .max_inbound_bytes(super::layout::MAX_MESSAGE_BYTES)
229            .build()
230    }
231    fn example_prompt(&self) -> &'static str {
232        "Cap'n Proto RPC server on port 5923 exporting a Directory interface with lookup and add methods"
233    }
234    fn get_startup_examples(&self) -> StartupExamples {
235        let llm = json!({"type":"open_server","base_stack":"capnp-rpc","port":5923,"startup_params":{"schema":EXAMPLE_SCHEMA,"interface":"Directory"},"instruction":"Answer lookups for a small fictional directory"});
236        let mut scripted = llm.clone();
237        scripted["event_handlers"] = json!([{"event_pattern":"capnp_call","handler":{"type":"script","language":"python","code":"import json,sys\ne=json.load(sys.stdin)['event']\nif e['method']=='add':\n    a={'type':'capnp_return','results':{'sum':e['params']['a']+e['params']['b']}}\nelse:\n    a={'type':'capnp_exception','reason':'not implemented: '+e['method']}\nprint(json.dumps({'actions':[a]}))"}}]);
238        let mut static_example = llm.clone();
239        static_example["event_handlers"] = json!([{"event_pattern":"capnp_call","handler":{"type":"static","actions":[{"type":"capnp_exception","reason":"maintenance"}]}}]);
240        StartupExamples::new(llm, scripted, static_example)
241    }
242    fn group_name(&self) -> &'static str {
243        "Application"
244    }
245}
246
247impl Server for CapnpRpcProtocol {
248    fn spawn(
249        &self,
250        ctx: SpawnContext,
251    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::net::SocketAddr>> + Send>>
252    {
253        Box::pin(super::spawn(ctx))
254    }
255    fn execute_action(&self, v: Value) -> Result<ActionResult> {
256        let name = v["type"].as_str().unwrap_or_default().to_string();
257        match name.as_str() {
258            "capnp_return" => ensure!(
259                v["results"].is_object(),
260                "results must be an object of the result fields"
261            ),
262            "capnp_exception" => {
263                check_exception(&v)?;
264            }
265            _ => bail!("Unknown Cap'n Proto RPC server action"),
266        }
267        Ok(ActionResult::Custom { name, data: v })
268    }
269}
netget/server/capnp_rpc/actions.rs

netget/server/capnp_rpc/
actions.rs
