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