netget/server/grpc/
actions.rs

1//! gRPC protocol actions implementation
2
3use crate::llm::actions::{
4    protocol_trait::{ActionResult, Protocol, Server},
5    ActionDefinition, Parameter,
6};
7use crate::protocol::log_template::LogTemplate;
8use crate::protocol::EventType;
9use crate::state::app_state::AppState;
10use anyhow::{Context, Result};
11use serde_json::json;
12use std::sync::LazyLock;
13use tracing::debug;
14
15/// gRPC protocol action handler
16pub struct GrpcProtocol;
17
18impl GrpcProtocol {
19    pub fn new() -> Self {
20        Self
21    }
22
23    fn execute_grpc_unary_response(&self, action: serde_json::Value) -> Result<ActionResult> {
24        let message = action
25            .get("message")
26            .context("Missing 'message' parameter in grpc_unary_response")?;
27
28        debug!("gRPC unary response: {}", serde_json::to_string(message)?);
29
30        // Return as Custom action result so server can encode to protobuf
31        Ok(ActionResult::Custom {
32            name: "grpc_unary_response".to_string(),
33            data: json!({ "message": message }),
34        })
35    }
36
37    fn execute_grpc_error(&self, action: serde_json::Value) -> Result<ActionResult> {
38        let code = action
39            .get("code")
40            .and_then(|v| v.as_str())
41            .unwrap_or("INTERNAL");
42
43        let message = action
44            .get("message")
45            .and_then(|v| v.as_str())
46            .context("Missing 'message' parameter in grpc_error")?;
47
48        debug!("gRPC error response: {} - {}", code, message);
49
50        // Return as Custom action result so server can construct proper gRPC error
51        Ok(ActionResult::Custom {
52            name: "grpc_error".to_string(),
53            data: json!({
54                "code": code,
55                "message": message
56            }),
57        })
58    }
59}
60
61// Implement Protocol trait (common functionality)
62impl Protocol for GrpcProtocol {
63    fn get_startup_parameters(&self) -> Vec<crate::llm::actions::ParameterDefinition> {
64        use crate::llm::actions::ParameterDefinition;
65        vec![
66                ParameterDefinition {
67                    name: "proto_schema".to_string(),
68                    type_hint: "string".to_string(),
69                    description: "Protobuf schema definition. IMPORTANT: For LLM responses, use inline .proto text (proto3 syntax). LLMs should NOT use base64-encoded FileDescriptorSet (truncation issues). Alternatively, provide path to .proto file on disk.".to_string(),
70                    required: true,
71                    example: json!("syntax = \"proto3\"; package test; service UserService { rpc GetUser(UserId) returns (User); } message UserId { int32 id = 1; } message User { int32 id = 1; string name = 2; string email = 3; }"),
72                    default: None,
73                },
74                ParameterDefinition {
75                    name: "enable_reflection".into(), type_hint: "boolean".into(), required: false,
76                    description: "Serve bounded gRPC v1/v1alpha reflection of the startup schema".into(),
77                    example: json!(true), default: Some(json!(super::streaming::DEFAULT_REFLECTION)),
78                },
79                ParameterDefinition {
80                    name: "stream_timeout_secs".into(), type_hint: "integer".into(), required: false,
81                    description: "Whole streaming RPC deadline, 1..3600 seconds, shortened by grpc-timeout".into(),
82                    example: json!(300), default: Some(json!(super::streaming::DEFAULT_TIMEOUT_SECS)),
83                },
84            ]
85    }
86    fn get_async_actions(&self, _state: &AppState) -> Vec<ActionDefinition> {
87        stream_actions()
88            .into_iter()
89            .filter(|action| action.name != "grpc_stream_wait" && action.name != "grpc_error")
90            .collect()
91    }
92    fn get_sync_actions(&self) -> Vec<ActionDefinition> {
93        let mut actions = vec![grpc_unary_response_action()];
94        actions.extend(stream_actions());
95        actions
96    }
97    fn protocol_name(&self) -> &'static str {
98        "gRPC"
99    }
100    fn get_event_types(&self) -> Vec<EventType> {
101        get_grpc_event_types()
102    }
103    fn stack_name(&self) -> &'static str {
104        "ETH>IP>TCP>HTTP2>GRPC"
105    }
106    fn keywords(&self) -> Vec<&'static str> {
107        vec!["grpc", "grpcserver", "protobuf"]
108    }
109
110    /// `protoc` must be on PATH at **runtime**, not merely at build time.
111    ///
112    /// `proto_schema` is a required startup parameter, and the two forms a caller is told to
113    /// use — a `.proto` path on disk and inline proto3 text — are compiled by shelling out to
114    /// `protoc` (`mod.rs`, via `Command::new`). A pre-compiled descriptor set (base64, or a
115    /// `.pb` file) is decoded directly and needs no `protoc`; [`Self::startup_dependencies`]
116    /// drops the dependency for exactly those, so the startup gate refuses only a start that
117    /// would really fail.
118    ///
119    /// Declaring it here means `server_startup` refuses with the installation hint before
120    /// registering the server, and the TUI and the model's protocol list exclude gRPC on a host
121    /// that lacks it, rather than everyone discovering it from a failure part-way through
122    /// startup.
123    ///
124    /// Note this is unrelated to `etcd`/`kubernetes`/`zookeeper`, which need `protoc` to
125    /// *compile* (prost/tonic build scripts) and not to run.
126    fn get_dependencies(&self) -> Vec<crate::protocol::dependencies::ProtocolDependency> {
127        let mut deps =
128            crate::llm::actions::protocol_trait::default_dependencies_from_privilege(self);
129        deps.push(crate::protocol::dependencies::ProtocolDependency::ToolInPath("protoc"));
130        deps
131    }
132
133    fn startup_dependencies(
134        &self,
135        startup_params: Option<&serde_json::Value>,
136    ) -> Vec<crate::protocol::dependencies::ProtocolDependency> {
137        let precompiled = startup_params
138            .and_then(|p| p.get("proto_schema"))
139            .and_then(|s| s.as_str())
140            .map(|schema| schema.trim().ends_with(".pb") || is_base64_descriptor_set(schema.trim()))
141            .unwrap_or(false);
142        let mut deps = self.get_dependencies();
143        if precompiled {
144            deps.retain(|d| {
145                *d != crate::protocol::dependencies::ProtocolDependency::ToolInPath("protoc")
146            });
147        }
148        deps
149    }
150    fn metadata(&self) -> crate::protocol::metadata::ProtocolMetadataV2 {
151        use crate::protocol::metadata::{DevelopmentState, ProtocolMetadataV2};
152        ProtocolMetadataV2::builder()
153            .state(DevelopmentState::Experimental)
154            .implementation("Dynamic protobuf services over hyper HTTP/2; tonic owns streaming and reflection framing, compression and trailers. Immutable bounded startup schema.")
155            .llm_control("Typed field-name JSON for unary requests and stream open/message/input-close/tick events; send, finish, cancel and wait controls. No protocol domain store.")
156            .e2e_testing("Mandatory independent grpcurl1.9.4 and generated grpcio1.75.1 peers exercise unary success/error, all three stream shapes, gzip and v1/v1alpha reflection; paired NetGet client/server and cancellation/bounds regressions in tests/server/grpc and tests/client/grpc. Peers fail when absent.")
157            .notes("Experimental expanded scope. New streams exclude schemas containing bytes fields; legacy unary bytes/base64 behavior remains. Reflection is enabled by default and can be disabled. Receiver TLS, mTLS, streaming retries, load balancing, reflection authentication and pcap/fuzz evidence are outside the validated scope. Legacy unary request compression remains rejected.")
158            .max_inbound_bytes(crate::server::grpc::MAX_REQUEST_BYTES)
159            .build()
160    }
161    fn description(&self) -> &'static str {
162        "gRPC server"
163    }
164    fn example_prompt(&self) -> &'static str {
165        "Start a gRPC server on port 50051 with this schema: service UserService { rpc GetUser(UserId) returns (User); }"
166    }
167    fn group_name(&self) -> &'static str {
168        "AI & API"
169    }
170    fn get_startup_examples(&self) -> crate::llm::actions::StartupExamples {
171        use crate::llm::actions::StartupExamples;
172
173        StartupExamples::new(
174            // LLM mode: instruction-based
175            json!({
176                "type": "open_server",
177                "port": 50051,
178                "base_stack": "grpc",
179                "instruction": "gRPC server with UserService. Respond to GetUser with user details",
180                "startup_params": {
181                    "proto_schema": "syntax = \"proto3\"; package test; service UserService { rpc GetUser(UserId) returns (User); } message UserId { int32 id = 1; } message User { int32 id = 1; string name = 2; string email = 3; }"
182                }
183            }),
184            // Script mode: event_handlers with script handler
185            json!({
186                "type": "open_server",
187                "port": 50051,
188                "base_stack": "grpc",
189                "startup_params": {
190                    "proto_schema": "syntax = \"proto3\"; package test; service Calculator { rpc Add(AddRequest) returns (AddResponse); } message AddRequest { int32 a = 1; int32 b = 2; } message AddResponse { int32 result = 1; }"
191                },
192                "event_handlers": [{
193                    "event_pattern": "grpc_unary_request",
194                    "handler": {
195                        "type": "script",
196                        "language": "python",
197                        "code": "import json,sys\nd=json.load(sys.stdin)\nreq=d['event']['request']\nprint(json.dumps({'actions':[{'type':'grpc_unary_response','message':{'result':req['a']+req['b']}}]}))"
198                    }
199                }]
200            }),
201            // Static mode: event_handlers with static actions
202            json!({
203                "type": "open_server",
204                "port": 50051,
205                "base_stack": "grpc",
206                "startup_params": {
207                    "proto_schema": "syntax = \"proto3\"; package test; service Greeter { rpc SayHello(HelloRequest) returns (HelloReply); } message HelloRequest { string name = 1; } message HelloReply { string message = 1; }"
208                },
209                "event_handlers": [{
210                    "event_pattern": "grpc_unary_request",
211                    "handler": {
212                        "type": "static",
213                        "actions": [{
214                            "type": "grpc_unary_response",
215                            "message": {"message": "Hello, World!"}
216                        }]
217                    }
218                }]
219            }),
220        )
221    }
222}
223
224// Implement Server trait (server-specific functionality)
225impl Server for GrpcProtocol {
226    fn spawn(
227        &self,
228        ctx: crate::protocol::SpawnContext,
229    ) -> std::pin::Pin<
230        Box<dyn std::future::Future<Output = anyhow::Result<std::net::SocketAddr>> + Send>,
231    > {
232        Box::pin(async move {
233            use crate::server::grpc::GrpcServer;
234            GrpcServer::spawn_with_llm_actions(
235                ctx.legacy_listen_addr(),
236                ctx.llm_client,
237                ctx.state,
238                ctx.status_tx,
239                ctx.server_id,
240                ctx.startup_params,
241            )
242            .await
243        })
244    }
245    fn execute_action(&self, action: serde_json::Value) -> Result<ActionResult> {
246        let action_type = action
247            .get("type")
248            .and_then(|v| v.as_str())
249            .context("Missing 'type' field in action")?;
250
251        match action_type {
252            "grpc_unary_response" => self.execute_grpc_unary_response(action),
253            "grpc_error" => self.execute_grpc_error(action),
254            "grpc_stream_send" | "grpc_stream_finish" | "grpc_stream_cancel"
255            | "grpc_stream_wait" => {
256                if action_type == "grpc_stream_send" {
257                    anyhow::ensure!(
258                        action["message"].is_object(),
259                        "message must be a field-name JSON object"
260                    );
261                }
262                if action_type == "grpc_stream_wait" {
263                    anyhow::ensure!(
264                        action["milliseconds"]
265                            .as_u64()
266                            .is_some_and(|n| (1..=1000).contains(&n)),
267                        "milliseconds must be 1..1000"
268                    );
269                }
270                if let Some(id) = action.get("stream_id") {
271                    anyhow::ensure!(
272                        id.as_u64()
273                            .is_some_and(|id| (1..=u64::from(u32::MAX)).contains(&id)),
274                        "stream_id must be 1..4294967295"
275                    );
276                }
277                Ok(ActionResult::Custom {
278                    name: action_type.to_owned(),
279                    data: action,
280                })
281            }
282            _ => Err(anyhow::anyhow!("Unknown gRPC action: {}", action_type)),
283        }
284    }
285}
286
287// ============================================================================
288// Action Definitions
289// ============================================================================
290
291fn grpc_unary_response_action() -> ActionDefinition {
292    ActionDefinition {
293        name: "grpc_unary_response".to_string(),
294        description: "Send gRPC unary response with JSON message".to_string(),
295        parameters: vec![Parameter {
296            name: "message".to_string(),
297            type_hint: "object".to_string(),
298            description: "Response message as JSON object matching protobuf schema".to_string(),
299            required: true,
300        }],
301        example: json!({
302            "type": "grpc_unary_response",
303            "message": {
304                "id": 123,
305                "name": "Alice",
306                "email": "alice@example.com"
307            }
308        }),
309        log_template: Some(
310            LogTemplate::new()
311                .with_info("-> gRPC response")
312                .with_debug("gRPC grpc_unary_response"),
313        ),
314    }
315}
316
317fn grpc_error_action() -> ActionDefinition {
318    ActionDefinition {
319        name: "grpc_error".to_string(),
320        description: "Return gRPC error with status code and message".to_string(),
321        parameters: vec![
322            Parameter {
323                name: "code".to_string(),
324                type_hint: "string".to_string(),
325                description:
326                    "gRPC status code (OK, CANCELLED, INVALID_ARGUMENT, NOT_FOUND, INTERNAL, etc.)"
327                        .to_string(),
328                required: false,
329            },
330            Parameter {
331                name: "message".to_string(),
332                type_hint: "string".to_string(),
333                description: "Terminal gRPC status explanation sent to the peer".to_string(),
334                required: true,
335            },
336        ],
337        example: json!({
338            "type": "grpc_error",
339            "code": "NOT_FOUND",
340            "message": "User not found"
341        }),
342        log_template: Some(
343            LogTemplate::new()
344                .with_info("-> gRPC error {code}")
345                .with_debug("gRPC grpc_error: code={code}, message={message}"),
346        ),
347    }
348}
349
350// ============================================================================
351// gRPC Action Constants
352// ============================================================================
353
354pub static GRPC_UNARY_RESPONSE_ACTION: LazyLock<ActionDefinition> =
355    LazyLock::new(|| grpc_unary_response_action());
356pub static GRPC_ERROR_ACTION: LazyLock<ActionDefinition> = LazyLock::new(|| grpc_error_action());
357
358// ============================================================================
359// gRPC Event Type Constants
360// ============================================================================
361
362/// gRPC unary request event - triggered when client makes a unary RPC call
363pub static GRPC_UNARY_REQUEST_EVENT: LazyLock<EventType> = LazyLock::new(|| {
364    EventType::new(
365        "grpc_unary_request",
366        "gRPC unary RPC request received from client",
367        json!({
368            "type": "grpc_unary_response",
369            "message": {
370                "id": 123,
371                "name": "Alice",
372                "email": "alice@example.com"
373            }
374        }),
375    )
376    .with_parameters(vec![
377        Parameter {
378            name: "service".to_string(),
379            type_hint: "string".to_string(),
380            description: "Service name (e.g., 'UserService')".to_string(),
381            required: true,
382        },
383        Parameter {
384            name: "method".to_string(),
385            type_hint: "string".to_string(),
386            description: "Method name (e.g., 'GetUser')".to_string(),
387            required: true,
388        },
389        Parameter {
390            name: "request".to_string(),
391            type_hint: "object".to_string(),
392            description: "Request message as JSON object".to_string(),
393            required: true,
394        },
395        Parameter {
396            name: "expected_response_schema".to_string(),
397            type_hint: "object".to_string(),
398            description: "Expected response schema as JSON Schema".to_string(),
399            required: false,
400        },
401    ])
402    .with_actions(vec![
403        GRPC_UNARY_RESPONSE_ACTION.clone(),
404        GRPC_ERROR_ACTION.clone(),
405    ])
406    .with_log_template(
407        LogTemplate::new()
408            .with_info("gRPC {client_ip} {service}/{method}")
409            .with_debug("gRPC method {service}/{method} from {client_ip}:{client_port}")
410            .with_trace("gRPC: {json_pretty(.)}"),
411    )
412});
413
414/// Get gRPC event types
415pub fn get_grpc_event_types() -> Vec<EventType> {
416    vec![
417        GRPC_UNARY_REQUEST_EVENT.clone(),
418        GRPC_STREAM_OPENED_EVENT.clone(),
419        GRPC_STREAM_MESSAGE_EVENT.clone(),
420        GRPC_STREAM_INPUT_CLOSED_EVENT.clone(),
421        GRPC_STREAM_TICK_EVENT.clone(),
422    ]
423}
424
425fn stream_actions() -> Vec<ActionDefinition> {
426    let mut actions = Vec::new();
427    for (name, description, extra, example) in [
428        ("grpc_stream_send", "Queue one typed response on the current stream; injected actions require stream_id", Some(("message", "object", "Field-name response JSON matching expected_response_schema")), json!({"type":"grpc_stream_send","stream_id":1,"message":{"name":"update","value":1}})),
429        ("grpc_stream_finish", "Finish this RPC with OK after pending responses; client-streaming requires half-close and one response", None, json!({"type":"grpc_stream_finish","stream_id":1})),
430        ("grpc_stream_cancel", "Cancel an active RPC; injected actions require stream_id", None, json!({"type":"grpc_stream_cancel","stream_id":1})),
431        ("grpc_stream_wait", "Continue reading input and request a subscription tick after 1..1000 milliseconds", Some(("milliseconds", "integer", "Delay before the next handler tick, 1..1000")), json!({"type":"grpc_stream_wait","milliseconds":100})),
432    ] {
433        let mut parameters = vec![Parameter { name:"stream_id".into(), type_hint:"integer".into(), required:false, description:"Active RPC id; required for peer injection, implicit within an event handler".into() }];
434        if let Some((name, hint, description)) = extra { parameters.push(Parameter { name:name.into(), type_hint:hint.into(), required:true, description:description.into() }); }
435        actions.push(ActionDefinition { name:name.into(), description:description.into(), parameters, example, log_template:Some(LogTemplate::new().with_info(format!("gRPC {name} queued for the active RPC"))) });
436    }
437    actions.push(grpc_error_action());
438    actions
439}
440fn stream_event(name: &'static str, description: &'static str) -> EventType {
441    EventType::new(
442        name,
443        description,
444        json!({"type":"grpc_stream_wait","milliseconds":100}),
445    )
446    .with_parameters(vec![
447        Parameter {
448            name: "stream_id".into(),
449            type_hint: "integer".into(),
450            required: true,
451            description: "Active RPC correlation id".into(),
452        },
453        Parameter {
454            name: "service".into(),
455            type_hint: "string".into(),
456            required: true,
457            description: "Fully qualified service name".into(),
458        },
459        Parameter {
460            name: "method".into(),
461            type_hint: "string".into(),
462            required: true,
463            description: "Protobuf RPC method within the declared service".into(),
464        },
465        Parameter {
466            name: "message".into(),
467            type_hint: "object|null".into(),
468            required: false,
469            description: "Decoded typed request message, or null for lifecycle/tick events".into(),
470        },
471        Parameter {
472            name: "expected_response_schema".into(),
473            type_hint: "object".into(),
474            required: true,
475            description: "Response fields and cardinalities; streaming excludes bytes fields"
476                .into(),
477        },
478    ])
479    .with_actions(stream_actions())
480}
481pub static GRPC_STREAM_OPENED_EVENT: LazyLock<EventType> = LazyLock::new(|| {
482    stream_event(
483        "grpc_stream_opened",
484        "Streaming RPC opened; server-streaming carries its one decoded request",
485    )
486});
487pub static GRPC_STREAM_MESSAGE_EVENT: LazyLock<EventType> = LazyLock::new(|| {
488    stream_event(
489        "grpc_stream_message",
490        "A decoded client-streaming or bidirectional message arrived",
491    )
492});
493pub static GRPC_STREAM_INPUT_CLOSED_EVENT: LazyLock<EventType> = LazyLock::new(|| {
494    stream_event(
495        "grpc_stream_input_closed",
496        "Client half-closed its request stream; client-streaming now permits one response",
497    )
498});
499pub static GRPC_STREAM_TICK_EVENT: LazyLock<EventType> = LazyLock::new(|| {
500    stream_event(
501        "grpc_stream_tick",
502        "Requested subscription interval elapsed; send updates, wait or finish",
503    )
504});
505
506/// Whether `schema` is a base64-encoded `FileDescriptorSet` — the form `mod.rs` decodes
507/// directly, without `protoc`.
508fn is_base64_descriptor_set(schema: &str) -> bool {
509    use base64::Engine as _;
510    use prost::Message as _;
511    if schema.len() > (4usize * 1024 * 1024).div_ceil(3) * 4 {
512        return false;
513    }
514    base64::engine::general_purpose::STANDARD
515        .decode(schema)
516        .map(|bytes| prost_types::FileDescriptorSet::decode(bytes.as_slice()).is_ok())
517        .unwrap_or(false)
518}
netget/server/grpc/actions.rs

netget/server/grpc/
actions.rs
