netget/server/connect_rpc/
actions.rs

1use crate::{
2    llm::actions::{
3        protocol_trait::{ActionResult, Protocol, Server},
4        ActionDefinition, ParameterDefinition, StartupExamples,
5    },
6    protocol::{EventType, SpawnContext},
7    state::AppState,
8};
9use anyhow::Result;
10use serde_json::{json, Value};
11pub static OPENED: std::sync::LazyLock<EventType> =
12    std::sync::LazyLock::new(|| connect_event("grpc_stream_opened"));
13pub static TICK: std::sync::LazyLock<EventType> =
14    std::sync::LazyLock::new(|| connect_event("grpc_stream_tick"));
15fn connect_event(id: &str) -> EventType {
16    let mut event = crate::server::grpc::actions::GrpcProtocol::new()
17        .get_event_types()
18        .into_iter()
19        .find(|event| event.id == id)
20        .unwrap();
21    event.parameters.push(crate::llm::actions::Parameter {
22        name: "metadata".into(),
23        type_hint: "object".into(),
24        description: "Bounded ASCII request metadata arrays".into(),
25        required: true,
26    });
27    event.with_actions(ConnectRpcProtocol.get_sync_actions())
28}
29pub(crate) fn event_type(id: &str) -> Option<&'static EventType> {
30    match id {
31        "grpc_stream_opened" => Some(&OPENED),
32        "grpc_stream_tick" => Some(&TICK),
33        _ => None,
34    }
35}
36#[derive(Default)]
37pub struct ConnectRpcProtocol;
38impl ConnectRpcProtocol {
39    pub fn new() -> Self {
40        Self
41    }
42}
43impl Protocol for ConnectRpcProtocol {
44    fn protocol_name(&self) -> &'static str {
45        "ConnectRPC"
46    }
47    fn stack_name(&self) -> &'static str {
48        "ETH>IP>TCP>HTTP>CONNECT-RPC"
49    }
50    fn group_name(&self) -> &'static str {
51        "Application"
52    }
53    fn description(&self) -> &'static str {
54        "Binary protobuf HTTP/1.1 unary and server-streaming binding"
55    }
56    fn example_prompt(&self) -> &'static str {
57        "Serve a protobuf greeting to a ConnectRPC client"
58    }
59    fn keywords(&self) -> Vec<&'static str> {
60        vec!["connect-rpc", "connectrpc", "connect rpc"]
61    }
62    fn get_async_actions(&self, state: &AppState) -> Vec<ActionDefinition> {
63        crate::server::grpc::actions::GrpcProtocol::new().get_async_actions(state)
64    }
65    fn get_sync_actions(&self) -> Vec<ActionDefinition> {
66        crate::server::grpc::actions::GrpcProtocol::new()
67            .get_sync_actions()
68            .into_iter()
69            .filter(|action| action.name != "grpc_unary_response")
70            .chain(std::iter::once(super::wire::metadata_action()))
71            .collect()
72    }
73    fn get_event_types(&self) -> Vec<EventType> {
74        vec![OPENED.clone(), TICK.clone()]
75    }
76    fn get_startup_parameters(&self) -> Vec<ParameterDefinition> {
77        let mut parameters =
78            crate::server::grpc::actions::GrpcProtocol::new().get_startup_parameters();
79        parameters.retain(|parameter| parameter.name == "proto_schema");
80        parameters.extend([
81            ParameterDefinition { name:"rpc_timeout_secs".into(), type_hint:"integer".into(), required:false,
82                description:"Whole RPC deadline 1..3600 seconds, shortened by connect-timeout-ms; includes response backpressure".into(),
83                example:json!(super::DEFAULT_RPC_TIMEOUT_SECS), default:Some(json!(super::DEFAULT_RPC_TIMEOUT_SECS)) },
84        ]);
85        parameters
86    }
87    fn get_dependencies(&self) -> Vec<crate::protocol::dependencies::ProtocolDependency> {
88        crate::server::grpc::actions::GrpcProtocol::new().get_dependencies()
89    }
90    fn startup_dependencies(
91        &self,
92        params: Option<&Value>,
93    ) -> Vec<crate::protocol::dependencies::ProtocolDependency> {
94        crate::server::grpc::actions::GrpcProtocol::new().startup_dependencies(params)
95    }
96    fn metadata(&self) -> crate::protocol::metadata::ProtocolMetadataV2 {
97        use crate::protocol::metadata::{DevelopmentState, ProtocolMetadataV2};
98        ProtocolMetadataV2::builder().state(DevelopmentState::Experimental)
99            .implementation("Separately registered binary protobuf ConnectRPC over HTTP/1.1. Connect bare/envelope framing and errors adapt tonic0.12.3 and shared bounded DynamicCodec decode messages.")
100            .llm_control("Typed stream-open and tick events for unary and server-streaming methods; explicit send/finish/cancel/wait/error. No application storage.")
101            .e2e_testing("Mandatory pinned independent Connect-ES peers in both roles, NetGet pairs and bounded framing/cancellation tests.")
102            .notes("Binary protobuf only: JSON message codecs and GET are rejected. Client streaming, bidi, reflection, WebSocket, receiver TLS, browser execution, pcap and fuzz evidence are outside this Experimental scope. Origin requests are rejected; no browser or CORS claim.")
103            .max_inbound_bytes(super::wire::MAX_MESSAGE_BYTES + 5).build()
104    }
105    fn get_startup_examples(&self) -> StartupExamples {
106        let schema = "syntax=\"proto3\"; package demo; message Greeting { string name=1; } service Greeter { rpc Hello(Greeting) returns (Greeting); }";
107        let base = |handler: Option<Value>, instruction: &str| {
108            let mut value = json!({"type":"open_server","base_stack":"connect-rpc","port":8080,"startup_params":{"proto_schema":schema},"instruction":instruction});
109            if let Some(handler) = handler {
110                value["event_handlers"] =
111                    json!([{"event_pattern":"grpc_stream_opened","handler":handler}]);
112            }
113            value
114        };
115        StartupExamples::new(
116            base(None, "On grpc_stream_opened, greet the request's message.name using grpc_stream_send with message.name, then grpc_stream_finish."),
117            base(Some(json!({"type":"script","language":"python","code":"import json,sys\ne=json.load(sys.stdin)['event']\nprint(json.dumps({'actions':[{'type':'grpc_stream_send','message':{'name':'Hello '+e['message']['name']}},{'type':'grpc_stream_finish'}]}))"})), ""),
118            base(Some(json!({"type":"static","actions":[{"type":"grpc_stream_send","message":{"name":"Hello"}},{"type":"grpc_stream_finish"}]})), ""),
119        )
120    }
121}
122impl Server for ConnectRpcProtocol {
123    fn spawn(
124        &self,
125        ctx: SpawnContext,
126    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::net::SocketAddr>> + Send>>
127    {
128        Box::pin(super::ConnectRpcServer::spawn(ctx))
129    }
130    fn execute_action(&self, action: Value) -> Result<ActionResult> {
131        if action["type"] == "connect_rpc_metadata" {
132            super::wire::validate_metadata_action(&action)?;
133            return Ok(ActionResult::Custom {
134                name: "connect_rpc_metadata".into(),
135                data: action,
136            });
137        }
138        anyhow::ensure!(
139            action["type"] != "grpc_unary_response",
140            "use grpc_stream_send and grpc_stream_finish for ConnectRPC unary replies"
141        );
142        crate::server::grpc::actions::GrpcProtocol::new().execute_action(action)
143    }
144}
netget/server/connect_rpc/actions.rs

netget/server/connect_rpc/
actions.rs
