1use crate::{
2 llm::actions::{
3 client_trait::{Client, ClientActionResult},
4 protocol_trait::Protocol,
5 ActionDefinition, Parameter, ParameterDefinition,
6 },
7 protocol::{ConnectContext, EventType},
8 state::AppState,
9};
10use anyhow::{bail, Result};
11use serde_json::{json, Value};
12use std::sync::LazyLock;
13#[derive(Default)]
14pub struct BoltClientProtocol;
15impl BoltClientProtocol {
16 pub fn new() -> Self {
17 Self
18 }
19}
20fn p(name: &str, kind: &str, description: &str, required: bool) -> Parameter {
21 Parameter {
22 name: name.into(),
23 type_hint: kind.into(),
24 description: description.into(),
25 required,
26 }
27}
28fn a(
29 name: &str,
30 description: &str,
31 parameters: Vec<Parameter>,
32 example: Value,
33) -> ActionDefinition {
34 let intent = match name {
35 "bolt_login" => "request native authentication; await acknowledgement",
36 "bolt_logoff" => "request native logoff; await acknowledgement",
37 "bolt_run" => "submit one query; await result field metadata",
38 "bolt_pull" => "request a bounded result page; await completion",
39 "bolt_discard" => "discard remaining results; await completion",
40 "bolt_begin" => "request an explicit transaction; await acknowledgement",
41 "bolt_commit" => "request transaction commit; await acknowledgement",
42 "bolt_rollback" => "request transaction rollback; await acknowledgement",
43 "bolt_reset" => "request session recovery; await acknowledgement",
44 "disconnect" => "close the connection and cancel pending I/O",
45 _ => unreachable!("only declared Bolt actions receive a log template"),
46 };
47 ActionDefinition {
48 name: name.into(),
49 description: description.into(),
50 parameters,
51 example,
52 log_template: Some(
53 crate::protocol::log_template::LogTemplate::new()
54 .with_debug(format!("Bolt {name}: {intent}")),
55 ),
56 }
57}
58fn tx_parameters() -> Vec<Parameter> {
59 vec![p("database","string","Optional database; omission leaves native default",false),p("mode","string","read or write; native default write",false),p("transaction_timeout_ms","integer","Optional server transaction timeout1..30000ms, separate from client whole-operation deadline",false),p("bookmarks","array","Up to16 native opaque causal bookmarks; no local causal guarantee",false)]
60}
61pub fn actions() -> Vec<ActionDefinition> {
62 let mut run=vec![p("query","string","Unmodified Cypher text, at most64KiB",true),p("parameters","object","Plain JSON parameter map with signed64-bit integers; graph, temporal, spatial and byte input types excluded",false)];
63 run.extend(tx_parameters());
64 vec![
65 a("bolt_login","Authenticate basic credentials, or none when both fields are omitted. Sends LOGON only in the authentication phase and Bolt5.1+.",vec![p("username","string","Basic principal; username and password must appear together",false),p("password","string","Bounded password, never returned in events or incidental diagnostics",false)],json!({"type":"bolt_login","username":"neo4j","password":"fixture-password"})),
66 a("bolt_logoff","Native LOGOFF from READY, Bolt5.1+; no token file or backend password change",vec![],json!({"type":"bolt_logoff"})),
67 a("bolt_run","RUN one query. A native SUCCESS establishes field order and one cursor; it does not complete or commit the query. Inside an explicit transaction, database/mode/bookmarks/timeout overrides are refused.",run,json!({"type":"bolt_run","query":"RETURN $name AS name","parameters":{"name":"Alice"},"mode":"read"})),
68 a("bolt_pull","PULL1..500 records (default100). Emits one complete page only after native SUCCESS; has_more keeps the cursor. Never requests unbounded PULL.",vec![p("n","integer","Record page count1..500",false)],json!({"type":"bolt_pull","n":100})),
69 a("bolt_discard","DISCARD all remaining records without exposing them; native completion summary is required",vec![],json!({"type":"bolt_discard"})),
70 a("bolt_begin","BEGIN one explicit transaction; one open result at a time",tx_parameters(),json!({"type":"bolt_begin","mode":"read"})),
71 a("bolt_commit","COMMIT only after the result was consumed/discarded; native SUCCESS acknowledges the commit and may carry bookmark",vec![],json!({"type":"bolt_commit"})),
72 a("bolt_rollback","Native ROLLBACK after the result was consumed/discarded",vec![],json!({"type":"bolt_rollback"})),
73 a("bolt_reset","Native RESET between operations, including failure recovery. Ends local transaction/cursor only after SUCCESS; pending-operation cancellation uses disconnect",vec![],json!({"type":"bolt_reset"})),
74 a("disconnect","Send best-effort GOODBYE when idle and close; cancel owned pending I/O immediately",vec![],json!({"type":"disconnect"})),
75 ]
76}
77fn event(id: &str, description: &str, parameters: Vec<Parameter>) -> EventType {
78 let example = match id {
79 "bolt_connected" => json!({"type":"bolt_login"}),
80 "bolt_authentication" => json!({"type":"bolt_run","query":"RETURN 42 AS answer"}),
81 "bolt_query_started" => json!({"type":"bolt_pull","n":100}),
82 "bolt_failure" => json!({"type":"bolt_reset"}),
83 _ => json!({"type":"disconnect"}),
84 };
85 EventType::new(id, description, example)
86 .with_parameters(parameters)
87 .with_actions(actions())
88}
89pub static CONNECTED_EVENT: LazyLock<EventType> = LazyLock::new(|| {
90 event("bolt_connected","Native connection, negotiated Bolt version and validated HELLO; authentication status is separate",vec![p("endpoint","string","Selected direct Bolt endpoint",true),p("version","string","Negotiated5.x version",true),p("server","object","Native HELLO metadata",true),p("authentication_verified","boolean","Only true after native authentication SUCCESS",true),p("authentication_metadata","object|null","Native startup LOGON metadata only when performed;5.0 auth metadata remains in server HELLO metadata",true),p("phase","string","authentication or ready",true)])
91});
92pub static AUTH_EVENT: LazyLock<EventType> = LazyLock::new(|| {
93 event(
94 "bolt_authentication",
95 "Native LOGON/LOGOFF acknowledgement; credentials excluded",
96 vec![
97 p("operation", "string", "login or logoff", true),
98 p(
99 "authentication_verified",
100 "boolean",
101 "Native login SUCCESS; false after LOGOFF",
102 true,
103 ),
104 p(
105 "metadata",
106 "object",
107 "Native advertised address or credentials-expired flag, if provided",
108 true,
109 ),
110 p("phase", "string", "Current local protocol phase", true),
111 ],
112 )
113});
114pub static RUN_EVENT: LazyLock<EventType> = LazyLock::new(|| {
115 event(
116 "bolt_query_started",
117 "RUN SUCCESS established field order and one result cursor; no completion inferred",
118 vec![
119 p(
120 "fields",
121 "array",
122 "Native ordered field names, including repeated/empty names if a peer sends them",
123 true,
124 ),
125 p(
126 "qid",
127 "integer|null",
128 "Native explicit transaction cursor identifier, optional in auto-commit",
129 true,
130 ),
131 p(
132 "metadata",
133 "object",
134 "Native RUN metadata; timing units milliseconds",
135 true,
136 ),
137 p(
138 "in_transaction",
139 "boolean",
140 "Explicit transaction active",
141 true,
142 ),
143 ],
144 )
145});
146pub static PAGE_EVENT: LazyLock<EventType> = LazyLock::new(|| {
147 event("bolt_result_page","Complete native PULL page or DISCARD receipt, emitted only after SUCCESS",vec![p("operation","string","pull or discard",true),p("fields","array","Ordered field names; records are positional arrays",true),p("records","array","Validated values: primitives/list/map, typed graph/temporal/spatial objects. Byte content omitted with length; nonfinite floats named explicitly",true),p("has_more","boolean","Native flag, false when absent",true),p("summary","object","Native metadata including type r/w/rw/s, stats, bookmark, database, timings or statuses when present; no fields invented",true),p("in_transaction","boolean","Explicit transaction remains active",true)])
148});
149pub static TX_EVENT: LazyLock<EventType> = LazyLock::new(|| {
150 event(
151 "bolt_session_result",
152 "Native BEGIN/COMMIT/ROLLBACK/RESET SUCCESS",
153 vec![
154 p(
155 "operation",
156 "string",
157 "begin, commit, rollback or reset",
158 true,
159 ),
160 p(
161 "metadata",
162 "object",
163 "Native acknowledgement metadata, including bookmark only when returned",
164 true,
165 ),
166 p(
167 "in_transaction",
168 "boolean",
169 "Local state after validated success",
170 true,
171 ),
172 p("phase", "string", "Current protocol phase", true),
173 ],
174 )
175});
176pub static FAILURE_EVENT: LazyLock<EventType> = LazyLock::new(|| {
177 event("bolt_failure","Native FAILURE or IGNORED. Tentative records are omitted and query failures require RESET",vec![p("operation","string","Selected operation",true),p("ignored","boolean","True only for native IGNORED",true),p("failure","object|null","Pre5.7 code/message;5.7+ neo4j_code/gql_status/description/diagnostic_record/cause. Credential reflections redacted",true),p("records_discarded","integer","Tentative records withheld before failure",true),p("phase","string","failed, or defunct for authentication/logoff/reset failure",true)])
178});
179pub static ERROR_EVENT: LazyLock<EventType> = LazyLock::new(|| {
180 event(
181 "bolt_request_error",
182 "Local selected action refusal; terminal transport/schema/deadline failures update client status",
183 vec![
184 p(
185 "category",
186 "string",
187 "The action category identifies a local refusal before any wire message",
188 true,
189 ),
190 p(
191 "error",
192 "string",
193 "Fixed bounded category text; no credential or peer payload",
194 true,
195 ),
196 p(
197 "backend_outcome",
198 "string",
199 "not_sent; the refused action produced no wire message",
200 true,
201 ),
202 ],
203 )
204});
205impl Protocol for BoltClientProtocol {
206 fn protocol_name(&self) -> &'static str {
207 "bolt"
208 }
209 fn stack_name(&self) -> &'static str {
210 "ETH>IP>TCP>BOLT"
211 }
212 fn description(&self) -> &'static str {
213 "Selected negotiated Neo4j-compatible queries and bounded result pages"
214 }
215 fn keywords(&self) -> Vec<&'static str> {
216 vec!["bolt", "neo4j", "cypher", "graph"]
217 }
218 fn group_name(&self) -> &'static str {
219 "Core"
220 }
221 fn example_prompt(&self) -> &'static str {
222 "Query a Neo4j-compatible Bolt service and report native results and failures"
223 }
224 fn get_async_actions(&self, _: &AppState) -> Vec<ActionDefinition> {
225 actions()
226 }
227 fn get_sync_actions(&self) -> Vec<ActionDefinition> {
228 vec![]
229 }
230 fn get_event_types(&self) -> Vec<EventType> {
231 vec![
232 CONNECTED_EVENT.clone(),
233 AUTH_EVENT.clone(),
234 RUN_EVENT.clone(),
235 PAGE_EVENT.clone(),
236 TX_EVENT.clone(),
237 FAILURE_EVENT.clone(),
238 ERROR_EVENT.clone(),
239 ]
240 }
241 fn get_startup_parameters(&self) -> Vec<ParameterDefinition> {
242 vec![
243 ParameterDefinition{name:"username".into(),type_hint:"string".into(),description:"Optional basic startup username; requires password. With neither,5.1+ waits for a login action;5.0 uses HELLO none authentication.".into(),required:false,example:json!("neo4j"),default:None},
244 ParameterDefinition{name:"password".into(),type_hint:"string".into(),description:"Optional bounded basic password, used during authentication and omitted from protocol events and incidental diagnostics.".into(),required:false,example:json!("fixture-password"),default:None},
245 ParameterDefinition{name:"request_timeout_secs".into(),type_hint:"integer".into(),description:"Whole connection/handshake/HELLO and each selected operation deadline1..30 seconds; default15. No idle session deadline.".into(),required:false,example:json!(15),default:Some(json!(super::DEFAULT_TIMEOUT_SECS))},
246 ]
247 }
248 fn metadata(&self) -> crate::protocol::metadata::ProtocolMetadataV2 {
249 use crate::protocol::metadata::*;
250 ProtocolMetadataV2::builder().state(DevelopmentState::Experimental).privilege_requirement(PrivilegeRequirement::None).well_known_port(7687).implementation("Direct native Bolt5.0..5.4/5.6..5.8 legacy negotiation, selected authentication, Cypher RUN and bounded PULL/DISCARD, transactions and RESET").llm_control("Typed queries/JSON parameters/results/native errors, common memory, event handlers and injection; no local graph database").e2e_testing("tests/client/bolt: isolated official Neo4j Community peer and native cypher-shell, NetGet pair, handler/model paths, bounds/deadlines and owned cancellation; full existing server suite retained").notes("Native TCP and certificate-verified bolt+s; browser TCP only. One ordered operation/result cursor/explicit transaction, messages1MiB, page500records/4MiB, fields256, query/text64KiB, JSON/wire depth32/nodes65536/retained8MiB, map256/list10000 parameters, event/action queues8, injection queue16/followups4, deadline1..30s. Latest bounded password retained only to redact credential reflections; no credentials file or graph store. Records tentative until PULL SUCCESS. Native metadata remains optional and native failure codes/GQL cause are preserved with reflected credentials omitted. Timeout/transport failure closes with unknown backend outcome. No routing/neo4j URI/pool, handshake manifest, Bolt6/5.5, TLS trust overrides/client certificates, bearer/Kerberos/impersonation, temporal/spatial/graph/byte input values, unbounded PULL, multiple simultaneous result streams, pending RESET interruption, automatic reconnect/retry/replay or conformance claim. Existing server model-as-database behavior remains.").max_inbound_bytes(crate::server::bolt::packstream::MAX_MESSAGE_BYTES).build()
251 }
252 fn get_startup_examples(&self) -> crate::llm::actions::StartupExamples {
253 let llm = json!({"type":"open_client","protocol":"bolt","base_stack":"bolt","remote_addr":"bolt://127.0.0.1:7687","instruction":"Authenticate anonymously if permitted, return42 as answer and consume one page"});
254 let mut fixed = llm.clone();
255 fixed["event_handlers"] = json!([
256 {"event_pattern":"bolt_connected","handler":{"type":"static","actions":[{"type":"bolt_login"}]}},
257 {"event_pattern":"bolt_authentication","handler":{"type":"static","actions":[{"type":"bolt_run","query":"RETURN 42 AS answer"}]}},
258 {"event_pattern":"bolt_query_started","handler":{"type":"static","actions":[{"type":"bolt_pull","n":100}]}},
259 {"event_pattern":"*","handler":{"type":"static","actions":[]}}
260 ]);
261 let mut script = fixed.clone();
262 script["event_handlers"][1]["handler"] = json!({"type":"script","language":"python","code":"import json,sys\njson.dump({'actions':[{'type':'bolt_run','query':'RETURN 42 AS answer'}]},sys.stdout)"});
263 crate::llm::actions::StartupExamples::new(llm, script, fixed)
264 }
265}
266impl Client for BoltClientProtocol {
267 fn connect(
268 &self,
269 ctx: ConnectContext,
270 ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::net::SocketAddr>> + Send>>
271 {
272 Box::pin(super::connect(ctx))
273 }
274 fn execute_action(&self, value: Value) -> Result<ClientActionResult> {
275 if !super::api::within_budget(&value) {
276 crate::utils::json_budget::drop_iteratively(value);
277 bail!("Bolt action depth/node/retained-content limit")
278 }
279 match super::api::action(&value)? {
280 super::api::Action::Disconnect => Ok(ClientActionResult::Disconnect),
281 _ => Ok(ClientActionResult::Custom {
282 name: "bolt".into(),
283 data: value,
284 }),
285 }
286 }
287}