netget/client/grpc/
actions.rs

1//! gRPC client protocol actions implementation
2
3use crate::protocol::log_template::LogTemplate;
4
5use crate::llm::actions::{
6    client_trait::{Client, ClientActionResult},
7    protocol_trait::Protocol,
8    ActionDefinition, Parameter, ParameterDefinition,
9};
10use crate::protocol::EventType;
11use crate::state::app_state::AppState;
12use anyhow::{Context, Result};
13use serde_json::json;
14use std::sync::LazyLock;
15
16/// gRPC client connected event
17pub static GRPC_CLIENT_CONNECTED_EVENT: LazyLock<EventType> = LazyLock::new(|| {
18    EventType::new(
19        "grpc_connected",
20        "gRPC client initialized and ready to call RPC methods",
21        json!({
22            "type": "call_grpc_method",
23            "service": "calculator.Calculator",
24            "method": "Add",
25            "request": {"a": 5, "b": 3}
26        }),
27    )
28    .with_parameters(vec![
29        Parameter {
30            name: "server_addr".to_string(),
31            type_hint: "string".to_string(),
32            description: "gRPC server address".to_string(),
33            required: true,
34        },
35        Parameter {
36            name: "services".to_string(),
37            type_hint: "array".to_string(),
38            description: "Available service names from schema".to_string(),
39            required: true,
40        },
41    ])
42});
43
44/// gRPC client response received event
45pub static GRPC_CLIENT_RESPONSE_RECEIVED_EVENT: LazyLock<EventType> = LazyLock::new(|| {
46    EventType::new(
47        "grpc_response_received",
48        "gRPC response received from server",
49        json!({
50            "type": "call_grpc_method",
51            "service": "calculator.Calculator",
52            "method": "Multiply",
53            "request": {"a": 2, "b": 3}
54        }),
55    )
56    .with_parameters(vec![
57        Parameter {
58            name: "service".to_string(),
59            type_hint: "string".to_string(),
60            description: "Fully qualified protobuf service name".to_string(),
61            required: true,
62        },
63        Parameter {
64            name: "method".to_string(),
65            type_hint: "string".to_string(),
66            description: "Protobuf RPC method within the declared service".to_string(),
67            required: true,
68        },
69        Parameter {
70            name: "response".to_string(),
71            type_hint: "object".to_string(),
72            description: "Response message as JSON".to_string(),
73            required: true,
74        },
75    ])
76});
77
78/// gRPC client error event
79pub static GRPC_CLIENT_ERROR_EVENT: LazyLock<EventType> = LazyLock::new(|| {
80    EventType::new(
81        "grpc_error",
82        "gRPC error received from server",
83        json!({
84            "type": "wait_for_more"
85        }),
86    )
87    .with_parameters(vec![
88        Parameter {
89            name: "service".to_string(),
90            type_hint: "string".to_string(),
91            description: "Fully qualified protobuf service name".to_string(),
92            required: true,
93        },
94        Parameter {
95            name: "method".to_string(),
96            type_hint: "string".to_string(),
97            description: "Protobuf RPC method within the declared service".to_string(),
98            required: true,
99        },
100        Parameter {
101            name: "code".to_string(),
102            type_hint: "string".to_string(),
103            description: "gRPC status code".to_string(),
104            required: true,
105        },
106        Parameter {
107            name: "message".to_string(),
108            type_hint: "string".to_string(),
109            description: "Terminal gRPC status explanation from the peer".to_string(),
110            required: true,
111        },
112    ])
113});
114
115/// gRPC client protocol action handler
116pub struct GrpcClientProtocol;
117
118impl GrpcClientProtocol {
119    pub fn new() -> Self {
120        Self
121    }
122}
123
124// Implement Protocol trait (common functionality)
125impl Protocol for GrpcClientProtocol {
126    fn get_startup_parameters(&self) -> Vec<ParameterDefinition> {
127        vec![
128                ParameterDefinition {
129                    name: "proto_schema".to_string(),
130                    description: "Optional inline proto3 text or .proto/.pb path; omit to discover the bounded schema through server reflection. Models should use inline text, not base64 descriptors.".to_string(),
131                    type_hint: "string".to_string(),
132                    required: false,
133                    example: json!("CpUCCg9jYWxjdWxhdG9yLnByb3RvEgpjYWxjdWxhdG9yIikKCkFkZFJlcXVlc3QSCwoDYQgBIAEoBVIBYRILCgNiCAIgASgFUgFiIiIKC0FkZFJlc3BvbnNlEhMKBnJlc3VsdBgBIAEoBVIGcmVzdWx0MkIKCkNhbGN1bGF0b3ISNAoDQWRkEhYuY2FsY3VsYXRvci5BZGRSZXF1ZXN0Gh0uY2FsY3VsYXRvci5BZGRSZXNwb25zZSIAYgZwcm90bzM="),
134                    default: None,
135                },
136                ParameterDefinition {
137                    name: "use_tls".to_string(),
138                    description: "Whether to use TLS for connection (default: false)".to_string(),
139                    type_hint: "boolean".to_string(),
140                    required: false,
141                    example: json!(false),
142                    default: Some(json!(super::DEFAULT_TLS)),
143                },
144                ParameterDefinition {
145                    name:"connect_timeout_secs".into(),type_hint:"integer".into(),required:false,description:"Whole schema/connection deadline, 1..60 seconds".into(),example:json!(10),default:Some(json!(super::CONNECT_TIMEOUT_SECS)),
146                },
147                ParameterDefinition {
148                    name:"stream_timeout_secs".into(),type_hint:"integer".into(),required:false,description:"Whole streaming RPC deadline, 1..3600 seconds".into(),example:json!(300),default:Some(json!(super::streaming::DEFAULT_STREAM_TIMEOUT)),
149                },
150                ParameterDefinition {
151                    name:"idle_timeout_secs".into(),type_hint:"integer".into(),required:false,description:"Idle with no active operations or handlers, 1..3600 seconds".into(),example:json!(120),default:Some(json!(super::streaming::DEFAULT_IDLE_TIMEOUT)),
152                },
153                ParameterDefinition {
154                    name:"server_name".into(),type_hint:"string".into(),required:false,description:"Verified TLS certificate hostname, default endpoint hostname; requires use_tls".into(),example:json!("localhost"),default:None,
155                },
156                ParameterDefinition {
157                    name:"ca_file".into(),type_hint:"string".into(),required:false,description:"Optional trusted PEM CA file, regular file at most 1 MiB; requires use_tls".into(),example:json!("/path/to/ca.pem"),default:None,
158                },
159            ]
160    }
161    fn get_async_actions(&self, _state: &AppState) -> Vec<ActionDefinition> {
162        let mut actions = vec![
163            ActionDefinition {
164                name: "call_grpc_method".to_string(),
165                description: "Call a gRPC method with the given request".to_string(),
166                parameters: vec![
167                    Parameter {
168                        name: "service".to_string(),
169                        type_hint: "string".to_string(),
170                        description: "Fully qualified service name (e.g., 'calculator.Calculator')"
171                            .to_string(),
172                        required: true,
173                    },
174                    Parameter {
175                        name: "method".to_string(),
176                        type_hint: "string".to_string(),
177                        description: "Method name (e.g., 'Add')".to_string(),
178                        required: true,
179                    },
180                    Parameter {
181                        name: "request".to_string(),
182                        type_hint: "object".to_string(),
183                        description: "Request message as JSON object".to_string(),
184                        required: true,
185                    },
186                    Parameter {
187                        name: "metadata".to_string(),
188                        type_hint: "object".to_string(),
189                        description: "Optional gRPC metadata (headers)".to_string(),
190                        required: false,
191                    },
192                ],
193                example: json!({
194                    "type": "call_grpc_method",
195                    "service": "calculator.Calculator",
196                    "method": "Add",
197                    "request": {"a": 5, "b": 3},
198                    "metadata": {"auth-token": "secret"}
199                }),
200                log_template: Some(
201                    LogTemplate::new().with_info("gRPC {service}/{method} call queued"),
202                ),
203            },
204            ActionDefinition {
205                name: "disconnect".to_string(),
206                description: "Disconnect from the gRPC server".to_string(),
207                parameters: vec![],
208                example: json!({
209                    "type": "disconnect"
210                }),
211                log_template: Some(LogTemplate::new().with_info("gRPC client disconnected")),
212            },
213        ];
214        actions.extend(stream_actions());
215        actions
216    }
217    fn get_sync_actions(&self) -> Vec<ActionDefinition> {
218        let mut actions = vec![
219            ActionDefinition {
220                name: "call_grpc_method".to_string(),
221                description: "Call another gRPC method in response to received data".to_string(),
222                parameters: vec![
223                    Parameter {
224                        name: "service".to_string(),
225                        type_hint: "string".to_string(),
226                        description: "Fully qualified service name".to_string(),
227                        required: true,
228                    },
229                    Parameter {
230                        name: "method".to_string(),
231                        type_hint: "string".to_string(),
232                        description: "Protobuf RPC method within the declared service".to_string(),
233                        required: true,
234                    },
235                    Parameter {
236                        name: "request".to_string(),
237                        type_hint: "object".to_string(),
238                        description: "Request message as JSON object".to_string(),
239                        required: true,
240                    },
241                    Parameter {
242                        name: "metadata".to_string(),
243                        type_hint: "object".to_string(),
244                        description: "Optional gRPC metadata (headers)".to_string(),
245                        required: false,
246                    },
247                ],
248                example: json!({
249                    "type": "call_grpc_method",
250                    "service": "calculator.Calculator",
251                    "method": "Multiply",
252                    "request": {"a": 2, "b": 3}
253                }),
254                log_template: Some(
255                    LogTemplate::new().with_info("gRPC {service}/{method} follow-up call queued"),
256                ),
257            },
258            ActionDefinition {
259                name: "wait_for_more".to_string(),
260                description: "Wait without making another call".to_string(),
261                parameters: vec![],
262                example: json!({
263                    "type": "wait_for_more"
264                }),
265                log_template: Some(
266                    LogTemplate::new().with_info("Waiting for another gRPC response"),
267                ),
268            },
269        ];
270        actions.extend(stream_actions());
271        actions
272    }
273    fn protocol_name(&self) -> &'static str {
274        "gRPC"
275    }
276    fn get_event_types(&self) -> Vec<EventType> {
277        vec![
278            GRPC_CLIENT_CONNECTED_EVENT.clone(),
279            GRPC_CLIENT_RESPONSE_RECEIVED_EVENT.clone(),
280            GRPC_CLIENT_ERROR_EVENT.clone(),
281            GRPC_CLIENT_STREAM_OPENED_EVENT.clone(),
282            GRPC_CLIENT_STREAM_MESSAGE_EVENT.clone(),
283            GRPC_CLIENT_STREAM_INPUT_READY_EVENT.clone(),
284            GRPC_CLIENT_STREAM_ENDED_EVENT.clone(),
285        ]
286    }
287    fn stack_name(&self) -> &'static str {
288        "ETH>IP>TCP>HTTP/2>gRPC"
289    }
290    fn keywords(&self) -> Vec<&'static str> {
291        vec!["grpc", "grpc client", "connect to grpc", "rpc"]
292    }
293    /// The gRPC client shells out to `protoc`, exactly as the server does.
294    ///
295    /// `src/client/grpc/mod.rs` runs `protoc --descriptor_set_out=/dev/stdout` to compile the
296    /// inline `.proto` text a caller supplies, so without the binary on PATH that schema form
297    /// cannot load at all. The *server* has declared this since the dependency mechanism was
298    /// adopted; the client shells out to the same binary and did not, so a host without
299    /// `protoc` was told about one half of the pair and discovered the other from a failure.
300    ///
301    /// Note this is a genuine **runtime** dependency, unlike `etcd`/`kubernetes`/`zookeeper`,
302    /// whose `protoc` use is in a build script and is finished before the binary exists.
303    fn get_dependencies(&self) -> Vec<crate::protocol::dependencies::ProtocolDependency> {
304        let mut deps =
305            crate::llm::actions::protocol_trait::default_dependencies_from_privilege(self);
306        deps.push(crate::protocol::dependencies::ProtocolDependency::ToolInPath("protoc"));
307        deps
308    }
309
310    /// Reflection and precompiled descriptors need no protoc; proto text/path does.
311    fn startup_dependencies(
312        &self,
313        startup_params: Option<&serde_json::Value>,
314    ) -> Vec<crate::protocol::dependencies::ProtocolDependency> {
315        let precompiled = startup_params
316            .and_then(|p| p.get("proto_schema"))
317            .and_then(|s| s.as_str())
318            .map(|schema| schema.trim().ends_with(".pb") || is_base64_descriptor_set(schema))
319            .unwrap_or(true);
320        let mut deps = self.get_dependencies();
321        if precompiled {
322            deps.retain(|d| {
323                *d != crate::protocol::dependencies::ProtocolDependency::ToolInPath("protoc")
324            });
325        }
326        deps
327    }
328    fn metadata(&self) -> crate::protocol::metadata::ProtocolMetadataV2 {
329        use crate::protocol::metadata::{DevelopmentState, ProtocolMetadataV2};
330
331        ProtocolMetadataV2::builder()
332            .state(DevelopmentState::Experimental)
333            .implementation("Bounded dynamic tonic gRPC client with unary and all three streaming forms, v1/v1alpha reflection discovery and verified TLS")
334            .llm_control("Typed request/response fields and owned start/send/input-half-close/cancel controls; bounded input-ready events report queue availability")
335            .e2e_testing("Mandatory generated grpcio1.75.1 peer and NetGet pair in tests/client/grpc/streaming_test.rs; reflection, stream shapes, gzip, TLS verification, cancellation, deadlines and bounded parked handlers. Legacy unary checks are retained.")
336            .notes("Experimental expanded scope. Streaming bytes fields and binary metadata are excluded; legacy unary bytes behavior is preserved. One endpoint, no automatic reconnect or streaming retry, no mTLS.")
337            .build()
338    }
339    fn description(&self) -> &'static str {
340        "gRPC client for calling RPC services"
341    }
342    fn example_prompt(&self) -> &'static str {
343        "Connect to gRPC server at localhost:50051 and call Calculator.Add with a=5, b=3"
344    }
345    fn group_name(&self) -> &'static str {
346        "RPC & API"
347    }
348    fn get_startup_examples(&self) -> crate::llm::actions::StartupExamples {
349        use crate::llm::actions::StartupExamples;
350        let schema = "syntax = \"proto3\"; package calculator; service Calculator { rpc Add(Input) returns (Output); } message Input { int32 a = 1; int32 b = 2; } message Output { int32 result = 1; }";
351        let call = json!({"type":"call_grpc_method","service":"calculator.Calculator","method":"Add","request":{"a":5,"b":3}});
352        StartupExamples::new(
353            json!({"type":"open_client","remote_addr":"localhost:50051","base_stack":"grpc",
354                "instruction":"Call calculator.Calculator/Add with a=5, b=3. Report the decoded result and disconnect.","startup_params":{"proto_schema":schema}}),
355            json!({"type":"open_client","remote_addr":"localhost:50051","base_stack":"grpc","startup_params":{"proto_schema":schema},
356                "event_handlers":[{"event_pattern":"*","handler":{"type":"script","language":"python",
357                    "code":"import json,sys\nd=json.load(sys.stdin)\na=[{'type':'wait_for_more'}]\nif d['event_type_id']=='grpc_connected': a=[{'type':'call_grpc_method','service':'calculator.Calculator','method':'Add','request':{'a':5,'b':3}}]\nelif d['event_type_id'] in ('grpc_response_received','grpc_error'): a=[{'type':'disconnect'}]\nprint(json.dumps({'actions':a}))"}}]}),
358            json!({"type":"open_client","remote_addr":"localhost:50051","base_stack":"grpc","startup_params":{"proto_schema":schema},
359                "event_handlers":[{"event_pattern":"grpc_connected","handler":{"type":"static","actions":[call]}},
360                    {"event_pattern":"grpc_response_received","handler":{"type":"static","actions":[{"type":"disconnect"}]}},
361                    {"event_pattern":"*","handler":{"type":"static","actions":[{"type":"wait_for_more"}]}}]}),
362        )
363    }
364}
365
366// Implement Client trait (client-specific functionality)
367impl Client for GrpcClientProtocol {
368    fn connect(
369        &self,
370        ctx: crate::protocol::ConnectContext,
371    ) -> std::pin::Pin<
372        Box<dyn std::future::Future<Output = anyhow::Result<std::net::SocketAddr>> + Send>,
373    > {
374        Box::pin(async move {
375            use crate::client::grpc::GrpcClient;
376            GrpcClient::connect_with_llm_actions(
377                ctx.remote_addr,
378                ctx.llm_client,
379                ctx.state,
380                ctx.status_tx,
381                ctx.client_id,
382                ctx.startup_params,
383            )
384            .await
385        })
386    }
387    fn execute_action(&self, action: serde_json::Value) -> Result<ClientActionResult> {
388        let action_type = action
389            .get("type")
390            .and_then(|v| v.as_str())
391            .context("Missing 'type' field in action")?;
392
393        match action_type {
394            "call_grpc_method" => {
395                let service = action
396                    .get("service")
397                    .and_then(|v| v.as_str())
398                    .context("Missing 'service' field")?
399                    .to_string();
400
401                let method = action
402                    .get("method")
403                    .and_then(|v| v.as_str())
404                    .context("Missing 'method' field")?
405                    .to_string();
406
407                let request = action
408                    .get("request")
409                    .context("Missing 'request' field")?
410                    .clone();
411
412                let metadata = action.get("metadata").and_then(|v| v.as_object()).cloned();
413
414                // Return custom result with RPC call data
415                Ok(ClientActionResult::Custom {
416                    name: "grpc_call".to_string(),
417                    data: json!({
418                        "service": service,
419                        "method": method,
420                        "request": request,
421                        "metadata": metadata,
422                    }),
423                })
424            }
425            "disconnect" => Ok(ClientActionResult::Disconnect),
426            "wait_for_more" => Ok(ClientActionResult::WaitForMore),
427            "grpc_stream_start" | "grpc_stream_send" | "grpc_stream_finish"
428            | "grpc_stream_cancel" => {
429                anyhow::ensure!(
430                    action["stream_id"]
431                        .as_u64()
432                        .is_some_and(|id| (1..=u64::from(u32::MAX)).contains(&id)),
433                    "stream_id must be 1..4294967295"
434                );
435                if action_type == "grpc_stream_start" {
436                    anyhow::ensure!(
437                        action["service"].is_string() && action["method"].is_string(),
438                        "service and method must be strings"
439                    );
440                }
441                if action_type == "grpc_stream_send" {
442                    anyhow::ensure!(
443                        action["message"].is_object(),
444                        "message must be field-name JSON"
445                    );
446                }
447                Ok(ClientActionResult::Custom {
448                    name: action_type.to_owned(),
449                    data: action,
450                })
451            }
452            _ => Err(anyhow::anyhow!(
453                "Unknown gRPC client action: {}",
454                action_type
455            )),
456        }
457    }
458}
459
460fn stream_actions() -> Vec<ActionDefinition> {
461    let mut actions = Vec::new();
462    for (name, description, example) in [
463        ("grpc_stream_start", "Start a real streaming method with a fresh stream_id; server-streaming requires request, client/bidi may provide an initial request", json!({"type":"grpc_stream_start","stream_id":1,"service":"streams.Session","method":"Watch","request":{"name":"watch"}})),
464        ("grpc_stream_send", "Queue one typed message on client/bidirectional input; rejects a full queue or closed input", json!({"type":"grpc_stream_send","stream_id":1,"message":{"name":"update","value":2}})),
465        ("grpc_stream_finish", "Half-close client/bidirectional input while continuing to receive replies", json!({"type":"grpc_stream_finish","stream_id":1})),
466        ("grpc_stream_cancel", "Cancel an active RPC and its pending reads/writes", json!({"type":"grpc_stream_cancel","stream_id":1})),
467    ] {
468        let mut parameters = vec![Parameter {name:"stream_id".into(),type_hint:"integer".into(),required:true,description:"Positive u32 id, unique for this client session".into()}];
469        if name == "grpc_stream_start" {
470            for (name,hint,required,description) in [
471                ("service","string",true,"Fully qualified service name"), ("method","string",true,"Streaming method name"),
472                ("request","object",false,"Typed initial request; required for server-streaming"),
473                ("metadata","object",false,"At most 16 lowercase ASCII metadata fields; reserved/binary headers excluded"),
474                ("gzip","boolean",false,"Compress outgoing stream messages with gzip"),
475            ] {parameters.push(Parameter {name:name.into(),type_hint:hint.into(),required,description:description.into()});}
476        } else if name == "grpc_stream_send" {parameters.push(Parameter {name:"message".into(),type_hint:"object".into(),required:true,description:"Field-name JSON matching the request schema, excluding bytes fields".into()});}
477        actions.push(ActionDefinition {name:name.into(),description:description.into(),parameters,example,log_template:Some(LogTemplate::new().with_info(format!("gRPC {name} queued for stream {{stream_id}}")))});
478    }
479    actions
480}
481fn stream_event(
482    name: &'static str,
483    description: &'static str,
484    fields: &[(&str, &str)],
485) -> EventType {
486    let mut parameters = vec![Parameter {
487        name: "stream_id".into(),
488        type_hint: "integer".into(),
489        required: true,
490        description: "RPC correlation id".into(),
491    }];
492    parameters.extend(fields.iter().map(|(name, hint)| {
493        Parameter {
494            name: (*name).into(),
495            type_hint: (*hint).into(),
496            required: true,
497            description: match *name {
498                "service" => "Fully qualified protobuf service name",
499                "method" => "Streaming RPC method within the declared service",
500                "sequence" => "Zero-based sequence of this decoded response",
501                "response" => "Decoded response fields matching the protobuf output schema",
502                "code" => "Terminal numeric gRPC status code for this RPC",
503                "message" => "Terminal gRPC status explanation from the peer",
504                "response_count" => "Total decoded responses received before this RPC ended",
505                "input_sequence" => "Sequence of the input consumed by the request encoder",
506                "queue_capacity" => "Available input queue slot, without a wire delivery receipt",
507                _ => "Typed field supplied by the streaming RPC event",
508            }
509            .into(),
510        }
511    }));
512    EventType::new(name, description, json!({"type":"wait_for_more"}))
513        .with_parameters(parameters)
514        .with_actions(stream_actions())
515}
516pub static GRPC_CLIENT_STREAM_OPENED_EVENT: LazyLock<EventType> = LazyLock::new(|| {
517    stream_event(
518        "grpc_stream_opened",
519        "Server accepted response headers for this RPC",
520        &[("service", "string"), ("method", "string")],
521    )
522});
523pub static GRPC_CLIENT_STREAM_MESSAGE_EVENT: LazyLock<EventType> = LazyLock::new(|| {
524    stream_event(
525        "grpc_stream_message_received",
526        "A bounded, decoded typed response arrived",
527        &[("sequence", "integer"), ("response", "object")],
528    )
529});
530pub static GRPC_CLIENT_STREAM_ENDED_EVENT: LazyLock<EventType> = LazyLock::new(|| {
531    stream_event(
532        "grpc_stream_ended",
533        "RPC completed, failed, reached its deadline or was cancelled",
534        &[
535            ("code", "integer"),
536            ("message", "string"),
537            ("response_count", "integer"),
538        ],
539    )
540});
541pub static GRPC_CLIENT_STREAM_INPUT_READY_EVENT: LazyLock<EventType> = LazyLock::new(|| {
542    stream_event("grpc_stream_input_ready","The request encoder consumed a queued input item; one queue slot is available, without claiming wire delivery",&[("input_sequence","integer"),("queue_capacity","integer")])
543});
544
545/// Whether `schema` is a base64-encoded `FileDescriptorSet`, the one form the client loads
546/// without `protoc`.
547fn is_base64_descriptor_set(schema: &str) -> bool {
548    use base64::Engine as _;
549    use prost::Message as _;
550    if schema.len() > (4usize * 1024 * 1024).div_ceil(3) * 4 {
551        return false;
552    }
553    base64::engine::general_purpose::STANDARD
554        .decode(schema)
555        .map(|bytes| prost_types::FileDescriptorSet::decode(bytes.as_slice()).is_ok())
556        .unwrap_or(false)
557}
netget/client/grpc/actions.rs

netget/client/grpc/
actions.rs
