1use crate::protocol::log_template::LogTemplate;
2use crate::{
3 llm::actions::{
4 client_trait::{Client, ClientActionResult},
5 protocol_trait::Protocol,
6 ActionDefinition, Parameter, ParameterDefinition, StartupExamples,
7 },
8 protocol::{ConnectContext, EventType},
9 state::AppState,
10};
11use anyhow::{ensure, Result};
12use serde_json::{json, Value};
13use std::sync::LazyLock;
14fn field(name: &str, hint: &str, description: &str, required: bool) -> Parameter {
15 Parameter {
16 name: name.into(),
17 type_hint: hint.into(),
18 description: description.into(),
19 required,
20 }
21}
22fn event(name: &str, description: &str, parameters: Vec<Parameter>) -> EventType {
23 EventType::new(name, description, json!({"type":"wait_for_more"}))
24 .with_parameters(parameters)
25 .with_actions(GrpcWebClientProtocol.get_sync_actions())
26}
27pub static CONNECTED: LazyLock<EventType> = LazyLock::new(|| {
28 event(
29 "grpc_web_connected",
30 "Binary gRPC-Web HTTP/1.1 connection ready",
31 vec![
32 field("remote_addr", "string", "Connected HTTP peer address", true),
33 field("services", "array", "Internal schema's service names", true),
34 field(
35 "tls_verified",
36 "boolean",
37 "False for this cleartext binding",
38 true,
39 ),
40 ],
41 )
42});
43pub static OPENED: LazyLock<EventType> = LazyLock::new(|| {
44 event(
45 "grpc_web_opened",
46 "RPC response headers received",
47 vec![
48 field("call_id", "integer", "Positive call identifier", true),
49 field("service", "string", "Protobuf service", true),
50 field("method", "string", "Protobuf method", true),
51 field(
52 "server_streaming",
53 "boolean",
54 "Method returns a stream",
55 true,
56 ),
57 ],
58 )
59});
60pub static MESSAGE: LazyLock<EventType> = LazyLock::new(|| {
61 event(
62 "grpc_web_message",
63 "Decoded protobuf response with field-name values",
64 vec![
65 field("call_id", "integer", "Positive call identifier", true),
66 field(
67 "sequence",
68 "integer",
69 "Response sequence, starting at 1",
70 true,
71 ),
72 field("response", "object", "Typed response values", true),
73 ],
74 )
75});
76pub static ENDED: LazyLock<EventType> = LazyLock::new(|| {
77 event(
78 "grpc_web_ended",
79 "Validated final status or local RPC failure",
80 vec![
81 field("call_id", "integer", "Positive call identifier", true),
82 field("code", "integer", "gRPC status 0..16", true),
83 field("message", "string", "Bounded diagnostic", true),
84 field("response_count", "integer", "Decoded response count", true),
85 ],
86 )
87});
88#[derive(Default)]
89pub struct GrpcWebClientProtocol;
90impl GrpcWebClientProtocol {
91 pub fn new() -> Self {
92 Self
93 }
94}
95impl Protocol for GrpcWebClientProtocol {
96 fn protocol_name(&self) -> &'static str {
97 "gRPC-Web"
98 }
99 fn stack_name(&self) -> &'static str {
100 "ETH>IP>TCP>HTTP>GRPC-WEB"
101 }
102 fn group_name(&self) -> &'static str {
103 "Application"
104 }
105 fn description(&self) -> &'static str {
106 "Typed binary HTTP/1.1 gRPC-Web unary and server-streaming client"
107 }
108 fn example_prompt(&self) -> &'static str {
109 "Call a gRPC-Web greeting method and read its typed response"
110 }
111 fn keywords(&self) -> Vec<&'static str> {
112 vec!["grpc-web", "grpcweb", "grpc web"]
113 }
114 fn get_async_actions(&self, _: &AppState) -> Vec<ActionDefinition> {
115 self.get_sync_actions()
116 }
117 fn get_sync_actions(&self) -> Vec<ActionDefinition> {
118 let action = |name: &str, description: &str, parameters: Vec<Parameter>, example: Value| {
119 ActionDefinition {
120 name: name.into(),
121 description: description.into(),
122 parameters,
123 example,
124 log_template: Some(LogTemplate::new().with_info(format!("gRPC-Web {name} queued"))),
125 }
126 };
127 vec![
128 action(
129 "grpc_web_call",
130 "Start one unary or server-streaming RPC; only one active RPC per connection",
131 vec![
132 field(
133 "call_id",
134 "integer",
135 "Fresh positive u32, never reused within this client; at most 256 calls",
136 true,
137 ),
138 field("service", "string", "Fully qualified service name", true),
139 field("method", "string", "Declared protobuf method name", true),
140 field(
141 "request",
142 "object",
143 "Protobuf field-name values; reachable bytes fields are excluded",
144 true,
145 ),
146 field(
147 "metadata",
148 "object",
149 "At most 16 ASCII fields, 8 KiB total; no reserved or binary headers",
150 false,
151 ),
152 field(
153 "gzip",
154 "boolean",
155 "Compress this request; default false",
156 false,
157 ),
158 ],
159 json!({"type":"grpc_web_call","call_id":1,"service":"demo.Greeter","method":"Hello","request":{"name":"Ada"}}),
160 ),
161 action(
162 "grpc_web_cancel",
163 "Cancel the matching active RPC and disconnect its HTTP/1.1 connection",
164 vec![field("call_id", "integer", "Active call identifier", true)],
165 json!({"type":"grpc_web_cancel","call_id":1}),
166 ),
167 action(
168 "disconnect",
169 "Cancel the owned RPC, handlers and transport",
170 vec![],
171 json!({"type":"disconnect"}),
172 ),
173 action(
174 "wait_for_more",
175 "Wait for another event or injected action",
176 vec![],
177 json!({"type":"wait_for_more"}),
178 ),
179 ]
180 }
181 fn get_event_types(&self) -> Vec<EventType> {
182 vec![
183 CONNECTED.clone(),
184 OPENED.clone(),
185 MESSAGE.clone(),
186 ENDED.clone(),
187 ]
188 }
189 fn get_startup_parameters(&self) -> Vec<ParameterDefinition> {
190 let mut schema = crate::client::grpc::GrpcClientProtocol::new()
191 .get_startup_parameters()
192 .into_iter()
193 .find(|parameter| parameter.name == "proto_schema")
194 .unwrap();
195 schema.required = true;
196 schema.description = "Required inline .proto, .proto/.pb path or precompiled descriptor set. No reflection; prefer inline schema".into();
197 let seconds = |name: &str, description: &str, default: u64| ParameterDefinition {
198 name: name.into(),
199 type_hint: "integer".into(),
200 description: description.into(),
201 required: false,
202 example: json!(default),
203 default: Some(json!(default)),
204 };
205 vec![
206 schema,
207 seconds(
208 "connect_timeout_secs",
209 "Whole schema/TCP/HTTP connection deadline, 1..60 seconds",
210 10,
211 ),
212 seconds(
213 "rpc_timeout_secs",
214 "Whole RPC deadline including model backpressure, 1..3600 seconds",
215 300,
216 ),
217 seconds(
218 "idle_timeout_secs",
219 "No RPC or handler activity deadline, 1..3600 seconds",
220 120,
221 ),
222 ]
223 }
224 fn get_dependencies(&self) -> Vec<crate::protocol::dependencies::ProtocolDependency> {
225 crate::server::grpc::actions::GrpcProtocol::new().get_dependencies()
226 }
227 fn startup_dependencies(
228 &self,
229 params: Option<&Value>,
230 ) -> Vec<crate::protocol::dependencies::ProtocolDependency> {
231 crate::server::grpc::actions::GrpcProtocol::new().startup_dependencies(params)
232 }
233 fn metadata(&self) -> crate::protocol::metadata::ProtocolMetadataV2 {
234 use crate::protocol::metadata::{DevelopmentState, ProtocolMetadataV2};
235 ProtocolMetadataV2::builder().state(DevelopmentState::Experimental)
236 .implementation("Owned HTTP/1.1 transport with tonic/prost-reflect typed messages and a bounded binary Web response/trailer adapter before tonic decoding.")
237 .llm_control("Call, observe typed response messages and final status, cancel or disconnect; one RPC per connection, responsive injected controls.")
238 .e2e_testing("Mandatory independent Connect-ES HTTP/1.1 server, paired NetGet binding and malformed framing/deadline/lifecycle tests.")
239 .notes("Binary only, cleartext only. Text, client/bidirectional streaming, reflection, WebSocket, automatic retries/reconnect, TLS, browser execution and pcap/fuzz evidence are excluded. Cancellation disconnects the HTTP/1.1 session.")
240 .max_inbound_bytes(crate::server::grpc_web::wire::MAX_MESSAGE_BYTES).build()
241 }
242 fn get_startup_examples(&self) -> StartupExamples {
243 let schema = "syntax=\"proto3\"; package demo; message Greeting { string name=1; } service Greeter { rpc Hello(Greeting) returns (Greeting); }";
244 let call = json!({"type":"grpc_web_call","call_id":1,"service":"demo.Greeter","method":"Hello","request":{"name":"Ada"}});
245 let base = |handler: Option<Value>, instruction: &str| {
246 let mut value = json!({"type":"open_client","base_stack":"grpc-web","remote_addr":"127.0.0.1:8080","startup_params":{"proto_schema":schema},"instruction":instruction});
247 if let Some(handler) = handler {
248 value["event_handlers"] =
249 json!([{"event_pattern":"grpc_web_connected","handler":handler}]);
250 }
251 value
252 };
253 StartupExamples::new(
254 base(None,"Call demo.Greeter.Hello with call_id1 and request.name Ada, then wait for its messages and final status."),
255 base(Some(json!({"type":"script","language":"python","code":"import json,sys\njson.load(sys.stdin)\nprint(json.dumps({'actions':[{'type':'grpc_web_call','call_id':1,'service':'demo.Greeter','method':'Hello','request':{'name':'Ada'}}]}))"})),""),
256 base(Some(json!({"type":"static","actions":[call]})),""),
257 )
258 }
259}
260impl Client for GrpcWebClientProtocol {
261 fn connect(
262 &self,
263 ctx: ConnectContext,
264 ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::net::SocketAddr>> + Send>>
265 {
266 Box::pin(super::GrpcWebClient::connect(ctx))
267 }
268 fn execute_action(&self, action: Value) -> Result<ClientActionResult> {
269 let kind = action["type"].as_str().unwrap_or("");
270 match kind {
271 "disconnect" => Ok(ClientActionResult::Disconnect),
272 "wait_for_more" => Ok(ClientActionResult::WaitForMore),
273 "grpc_web_call" | "grpc_web_cancel" => {
274 ensure!(
275 action["call_id"]
276 .as_u64()
277 .is_some_and(|id| id > 0 && id <= u64::from(u32::MAX)),
278 "call_id must be a positive u32"
279 );
280 if kind == "grpc_web_call" {
281 for (key, limit) in [("service", 1024), ("method", 256)] {
282 ensure!(
283 action[key]
284 .as_str()
285 .is_some_and(|value| !value.is_empty() && value.len() <= limit),
286 "invalid {key}"
287 );
288 }
289 ensure!(action["request"].is_object(), "request must be an object");
290 ensure!(
291 action.get("gzip").is_none_or(Value::is_boolean),
292 "gzip must be boolean"
293 );
294 }
295 Ok(ClientActionResult::Custom {
296 name: kind.into(),
297 data: action,
298 })
299 }
300 _ => anyhow::bail!("unknown gRPC-Web action"),
301 }
302 }
303}