1use crate::llm::actions::{
2 protocol_trait::{ActionResult, Protocol, Server},
3 ActionDefinition, Parameter, ParameterDefinition, StartupExamples,
4};
5use crate::protocol::log_template::LogTemplate;
6use crate::protocol::{EventType, SpawnContext};
7use crate::state::app_state::AppState;
8use anyhow::{bail, ensure, Context, Result};
9use serde_json::{json, Value};
10use std::sync::LazyLock;
11
12use super::wire;
13
14#[derive(Default)]
15pub struct ConsulProtocol;
16impl ConsulProtocol {
17 pub fn new() -> Self {
18 Self
19 }
20}
21
22pub fn parameter(name: &str, kind: &str, description: &str, required: bool) -> Parameter {
23 Parameter {
24 name: name.into(),
25 type_hint: kind.into(),
26 description: description.into(),
27 required,
28 }
29}
30
31pub fn action(
32 name: &str,
33 description: &str,
34 parameters: Vec<Parameter>,
35 example: Value,
36 log: &str,
37) -> ActionDefinition {
38 ActionDefinition {
39 name: name.into(),
40 description: description.into(),
41 parameters,
42 example,
43 log_template: Some(LogTemplate::new().with_info(log)),
44 }
45}
46
47fn entries_action() -> ActionDefinition {
48 action(
49 "consul_kv_entries",
50 "Answer a KV read with the entries under the key (one for a plain read; every key under the prefix when recurse or keys_only).",
51 vec![parameter("entries", "array", "Each {key, value (text), encoding? utf8|hex, flags?, modify_index?}", true)],
52 json!({"type":"consul_kv_entries","entries":[{"key":"app/config","value":"hello world"}]}),
53 "-> Consul KV {preview(entries,100)}",
54 )
55}
56
57fn not_found_action() -> ActionDefinition {
58 action(
59 "consul_not_found",
60 "The key (or nothing under the prefix) does not exist: 404.",
61 vec![],
62 json!({"type":"consul_not_found"}),
63 "-> Consul 404",
64 )
65}
66
67fn ok_action() -> ActionDefinition {
68 action(
69 "consul_ok",
70 "Accept the write, delete or registration (Consul answers true / 200).",
71 vec![],
72 json!({"type":"consul_ok"}),
73 "-> Consul ok",
74 )
75}
76
77fn refuse_action() -> ActionDefinition {
78 action(
79 "consul_refuse",
80 "Refuse a KV write or delete the way Consul refuses a failed check-and-set: 200 with false.",
81 vec![],
82 json!({"type":"consul_refuse"}),
83 "-> Consul false",
84 )
85}
86
87fn error_action() -> ActionDefinition {
88 action(
89 "consul_error",
90 "Fail the request with an HTTP status and a plain-text message.",
91 vec![
92 parameter("status", "number", "400, 403, 404 or 500", true),
93 parameter(
94 "message",
95 "string",
96 "Plain-text explanation, at most 1024 bytes",
97 true,
98 ),
99 ],
100 json!({"type":"consul_error","status":403,"message":"Permission denied"}),
101 "-> Consul {status} {message}",
102 )
103}
104
105fn services_action() -> ActionDefinition {
106 action(
107 "consul_services",
108 "Answer /v1/catalog/services with every service name and its tags.",
109 vec![parameter(
110 "services",
111 "object",
112 "Service name to its tags, e.g. {\"web\": [\"v1\"]}",
113 true,
114 )],
115 json!({"type":"consul_services","services":{"web":["v1"],"db":[]}}),
116 "-> Consul services {preview(services,100)}",
117 )
118}
119
120fn instances_action() -> ActionDefinition {
121 action(
122 "consul_instances",
123 "Answer a service query with its instances; node, datacenter and health fields are filled in.",
124 vec![parameter("instances", "array", "Each {id?, name, address?, port?, tags?, meta?}", true)],
125 json!({"type":"consul_instances","instances":[{"id":"web1","name":"web","address":"127.0.0.1","port":8080,"tags":["v1"]}]}),
126 "-> Consul instances {preview(instances,100)}",
127 )
128}
129
130pub static KV_READ_EVENT: LazyLock<EventType> = LazyLock::new(|| {
131 EventType::new(
132 "consul_kv_read",
133 "A client read the KV store (GET /v1/kv/<key>). NetGet stores nothing: answer from what you hold.",
134 entries_action().example.clone(),
135 )
136 .with_parameters(vec![
137 parameter("key", "string", "The key, or the prefix with recurse/keys_only", true),
138 parameter("recurse", "boolean", "Every entry under the prefix is wanted", true),
139 parameter("keys_only", "boolean", "Only the keys under the prefix are wanted", true),
140 ])
141 .with_actions(vec![entries_action(), not_found_action(), error_action()])
142});
143
144pub static KV_WRITE_EVENT: LazyLock<EventType> = LazyLock::new(|| {
145 EventType::new(
146 "consul_kv_write",
147 "A client wrote a key (PUT /v1/kv/<key>). Remember it and accept, or refuse.",
148 ok_action().example.clone(),
149 )
150 .with_parameters(vec![
151 parameter("key", "string", "The key written", true),
152 parameter(
153 "value",
154 "string",
155 "The value (text, or hex per value_encoding)",
156 true,
157 ),
158 parameter(
159 "value_encoding",
160 "string",
161 "How value is shown: utf8 or hex",
162 true,
163 ),
164 parameter(
165 "flags",
166 "number",
167 "Opaque client flags stored with the value",
168 true,
169 ),
170 parameter(
171 "cas",
172 "number",
173 "Check-and-set index: write only if the key's modify index is this (0: only if absent)",
174 false,
175 ),
176 ])
177 .with_actions(vec![ok_action(), refuse_action(), error_action()])
178});
179
180pub static KV_DELETE_EVENT: LazyLock<EventType> = LazyLock::new(|| {
181 EventType::new(
182 "consul_kv_delete",
183 "A client deleted a key or prefix (DELETE /v1/kv/<key>).",
184 ok_action().example.clone(),
185 )
186 .with_parameters(vec![
187 parameter("key", "string", "The key or prefix deleted", true),
188 parameter(
189 "recurse",
190 "boolean",
191 "Everything under the prefix goes",
192 true,
193 ),
194 parameter("cas", "number", "Check-and-set index, if given", false),
195 ])
196 .with_actions(vec![ok_action(), refuse_action(), error_action()])
197});
198
199pub static CATALOG_EVENT: LazyLock<EventType> = LazyLock::new(|| {
200 EventType::new(
201 "consul_catalog",
202 "A client asked about services: services (all names), service or health (one service's instances), agent_services (this agent's).",
203 instances_action().example.clone(),
204 )
205 .with_parameters(vec![
206 parameter("endpoint", "string", "services, service, health or agent_services", true),
207 parameter("name", "string", "The service asked about, for service and health", false),
208 ])
209 .with_actions(vec![services_action(), instances_action(), error_action()])
210});
211
212pub static REGISTER_EVENT: LazyLock<EventType> = LazyLock::new(|| {
213 EventType::new(
214 "consul_register",
215 "A client registered (or deregistered) a service with the agent.",
216 ok_action().example.clone(),
217 )
218 .with_parameters(vec![
219 parameter("operation", "string", "register or deregister", true),
220 parameter(
221 "service",
222 "object",
223 "For register: {id, name, address, port, tags, meta}",
224 false,
225 ),
226 parameter("id", "string", "For deregister: the service ID", false),
227 ])
228 .with_actions(vec![ok_action(), error_action()])
229});
230
231pub fn check_error(v: &Value) -> Result<()> {
232 let status = v["status"].as_u64().context("status required")?;
233 ensure!(
234 [400, 403, 404, 500].contains(&status),
235 "status must be 400, 403, 404 or 500"
236 );
237 let m = v["message"].as_str().context("message required")?;
238 ensure!(
239 m.len() <= 1024 && !crate::utils::sanitize::has_controls(m),
240 "message at most 1024 bytes without control characters"
241 );
242 Ok(())
243}
244
245impl Protocol for ConsulProtocol {
246 fn protocol_name(&self) -> &'static str {
247 "Consul"
248 }
249 fn stack_name(&self) -> &'static str {
250 "ETH>IP>TCP>HTTP>Consul"
251 }
252 fn description(&self) -> &'static str {
253 "Consul agent HTTP API: KV store, catalog, health and service registration, answered by the handler"
254 }
255 fn keywords(&self) -> Vec<&'static str> {
256 vec![
257 "consul",
258 "service discovery",
259 "consul kv",
260 "hashicorp consul",
261 ]
262 }
263 fn get_async_actions(&self, _: &AppState) -> Vec<ActionDefinition> {
264 vec![]
265 }
266 fn get_sync_actions(&self) -> Vec<ActionDefinition> {
267 vec![
268 entries_action(),
269 not_found_action(),
270 ok_action(),
271 refuse_action(),
272 error_action(),
273 services_action(),
274 instances_action(),
275 ]
276 }
277 fn get_event_types(&self) -> Vec<EventType> {
278 vec![
279 KV_READ_EVENT.clone(),
280 KV_WRITE_EVENT.clone(),
281 KV_DELETE_EVENT.clone(),
282 CATALOG_EVENT.clone(),
283 REGISTER_EVENT.clone(),
284 ]
285 }
286 fn get_startup_parameters(&self) -> Vec<ParameterDefinition> {
287 vec![]
288 }
289 fn metadata(&self) -> crate::protocol::metadata::ProtocolMetadataV2 {
290 use crate::protocol::metadata::*;
291 ProtocolMetadataV2::builder()
292 .state(DevelopmentState::Experimental)
293 .privilege_requirement(PrivilegeRequirement::None)
294 .well_known_port(8500)
295 .implementation("HTTP/1.1 via hyper: /v1/kv (get, recurse, keys, raw, put with flags and cas, delete), /v1/catalog/services|service|nodes|datacenters, /v1/health/service, /v1/agent/self|services|service/register|service/deregister, /v1/status/leader|peers; Consul's JSON shapes, base64 values and X-Consul-Index")
296 .llm_control("Every KV answer and write decision, every catalog and health answer, and every service registration — NetGet stores nothing")
297 .e2e_testing("tests/server/consul: the official consul CLI (Go api client) and py-consul as independent clients; raw HTTP for bounds and refusals")
298 .notes("No blocking queries (a wait returns at once), no sessions or locks, no ACLs, no Connect, no DNS interface, no watches. Bodies are capped at 512 KiB (Consul's own KV value limit), listings at 10 000 entries. One request per connection. A handler failure is a 500 with a generic message, never a fabricated value.")
299 .request_only("Every response answers an HTTP request")
300 .answers_on_failure()
301 .max_inbound_bytes(wire::MAX_VALUE_BYTES)
302 .build()
303 }
304 fn example_prompt(&self) -> &'static str {
305 "Consul agent on port 8500 holding service registrations and KV config in memory"
306 }
307 fn get_startup_examples(&self) -> StartupExamples {
308 let llm = json!({"type":"open_server","base_stack":"consul","port":8500,"instruction":"Keep KV and services in memory; app/config is 'hello'"});
309 let mut scripted = llm.clone();
310 scripted["event_handlers"] = json!([{"event_pattern":"consul_kv_read","handler":{"type":"script","language":"python","code":"import json,sys\ne=json.load(sys.stdin)['event']\na={'type':'consul_kv_entries','entries':[{'key':e['key'],'value':'hello'}]}\nprint(json.dumps({'actions':[a]}))"}}]);
311 let mut static_example = llm.clone();
312 static_example["event_handlers"] = json!([{"event_pattern":"consul_kv_write","handler":{"type":"static","actions":[{"type":"consul_ok"}]}}]);
313 StartupExamples::new(llm, scripted, static_example)
314 }
315 fn group_name(&self) -> &'static str {
316 "Network Services"
317 }
318}
319
320impl Server for ConsulProtocol {
321 fn spawn(
322 &self,
323 ctx: SpawnContext,
324 ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::net::SocketAddr>> + Send>>
325 {
326 Box::pin(super::spawn(ctx))
327 }
328 fn execute_action(&self, v: Value) -> Result<ActionResult> {
329 let name = v["type"].as_str().unwrap_or_default().to_string();
330 match name.as_str() {
331 "consul_not_found" | "consul_ok" | "consul_refuse" => {}
332 "consul_error" => check_error(&v)?,
333 "consul_kv_entries" => {
334 let entries = v["entries"]
335 .as_array()
336 .context("entries must be an array")?;
337 ensure!(entries.len() <= wire::MAX_ENTRIES, "too many entries");
338 for e in entries {
339 wire::kv_entry(e, 1)?;
340 }
341 }
342 "consul_services" => ensure!(v["services"].is_object(), "services must be an object"),
343 "consul_instances" => {
344 let list = v["instances"]
345 .as_array()
346 .context("instances must be an array")?;
347 ensure!(list.len() <= wire::MAX_ENTRIES, "too many instances");
348 for i in list {
349 wire::instance(i)?;
350 }
351 }
352 _ => bail!("Unknown Consul server action"),
353 }
354 Ok(ActionResult::Custom { name, data: v })
355 }
356}