use crate::llm::actions::{
protocol_trait::{ActionResult, Protocol, Server},
ActionDefinition, Parameter, ParameterDefinition,
};
use crate::protocol::log_template::LogTemplate;
use crate::protocol::EventType;
use crate::state::app_state::AppState;
use anyhow::Result;
use serde_json::{json, Value};
use std::sync::LazyLock;
pub struct EtcdProtocol {}
impl EtcdProtocol {
pub fn new() -> Self {
Self {}
}
}
pub static ETCD_RANGE_REQUEST_EVENT: LazyLock<EventType> = LazyLock::new(|| {
EventType::new(
"etcd_range_request",
"Triggered when a client sends a Range (get) request to query keys",
json!({
"type": "etcd_range_response",
"kvs": [
{"key": "foo", "value": "bar", "create_revision": 1, "mod_revision": 1, "version": 1, "lease": 0}
],
"more": false,
"count": 1
}),
)
.with_parameters(vec![
Parameter {
name: "key".to_string(),
type_hint: "string".to_string(),
description: "Key to query".to_string(),
required: true,
},
Parameter {
name: "range_end".to_string(),
type_hint: "string".to_string(),
description: "End of key range (for prefix/range queries)".to_string(),
required: false,
},
Parameter {
name: "limit".to_string(),
type_hint: "number".to_string(),
description: "Maximum number of keys to return".to_string(),
required: false,
},
])
.with_actions(vec![etcd_range_response_action(), etcd_error_action()])
});
pub static ETCD_PUT_REQUEST_EVENT: LazyLock<EventType> = LazyLock::new(|| {
EventType::new(
"etcd_put_request",
"Triggered when a client sends a Put request to store a key-value pair",
json!({
"type": "etcd_put_response",
"revision": 2
}),
)
.with_parameters(vec![
Parameter {
name: "key".to_string(),
type_hint: "string".to_string(),
description: "Key being written".to_string(),
required: true,
},
Parameter {
name: "value".to_string(),
type_hint: "string".to_string(),
description: "Value being written".to_string(),
required: true,
},
Parameter {
name: "lease".to_string(),
type_hint: "number".to_string(),
description: "Lease ID attached to the key (0 = no lease)".to_string(),
required: false,
},
])
.with_actions(vec![etcd_put_response_action(), etcd_error_action()])
});
pub static ETCD_DELETE_REQUEST_EVENT: LazyLock<EventType> = LazyLock::new(|| {
EventType::new(
"etcd_delete_request",
"Triggered when a client sends a DeleteRange request",
json!({
"type": "etcd_delete_range_response",
"deleted": 1
}),
)
.with_parameters(vec![
Parameter {
name: "key".to_string(),
type_hint: "string".to_string(),
description: "Key to delete".to_string(),
required: true,
},
Parameter {
name: "range_end".to_string(),
type_hint: "string".to_string(),
description: "End of key range for prefix/range deletes".to_string(),
required: false,
},
])
.with_actions(vec![
etcd_delete_range_response_action(),
etcd_error_action(),
])
});
pub static ETCD_TXN_REQUEST_EVENT: LazyLock<EventType> = LazyLock::new(|| {
EventType::new(
"etcd_txn_request",
"Triggered when a client sends a transaction request",
json!({
"type": "etcd_txn_response",
"succeeded": true
}),
)
.with_parameters(vec![
Parameter {
name: "compare_count".to_string(),
type_hint: "number".to_string(),
description: "Number of comparison predicates in the transaction".to_string(),
required: true,
},
Parameter {
name: "success_count".to_string(),
type_hint: "number".to_string(),
description: "Number of operations to run if the comparisons hold".to_string(),
required: true,
},
Parameter {
name: "failure_count".to_string(),
type_hint: "number".to_string(),
description: "Number of operations to run if they do not".to_string(),
required: true,
},
Parameter {
name: "compares".to_string(),
type_hint: "array".to_string(),
description: "Comparison predicates as objects with 'key', 'target' \
(VERSION/CREATE/MOD/VALUE), 'result' (EQUAL/GREATER/LESS/NOT_EQUAL) \
and 'value' fields"
.to_string(),
required: true,
},
])
.with_actions(vec![etcd_txn_response_action(), etcd_error_action()])
});
fn etcd_range_response_action() -> ActionDefinition {
ActionDefinition {
name: "etcd_range_response".to_string(),
description: "Return key-value pairs for a Range request".to_string(),
parameters: vec![
Parameter {
name: "kvs".to_string(),
type_hint: "array".to_string(),
description: "Array of key-value objects with 'key', 'value', 'create_revision', 'mod_revision', 'version', 'lease' fields".to_string(),
required: true,
},
Parameter {
name: "more".to_string(),
type_hint: "boolean".to_string(),
description: "Whether there are more keys to return".to_string(),
required: false,
},
Parameter {
name: "count".to_string(),
type_hint: "number".to_string(),
description: "Total count of keys matching the range".to_string(),
required: false,
},
],
example: json!({
"type": "etcd_range_response",
"kvs": [
{"key": "foo", "value": "bar", "create_revision": 1, "mod_revision": 1, "version": 1, "lease": 0}
],
"more": false,
"count": 1
}),
log_template: Some(
LogTemplate::new()
.with_info("-> etcd range response ({kvs_len} keys)")
.with_debug("etcd etcd_range_response: keys={kvs_len} more={more}"),
),
}
}
fn etcd_put_response_action() -> ActionDefinition {
ActionDefinition {
name: "etcd_put_response".to_string(),
description: "Acknowledge a Put. The server increments its own revision counter if you \
omit one."
.to_string(),
parameters: vec![Parameter {
name: "revision".to_string(),
type_hint: "number".to_string(),
description: "Revision this write is recorded at. Omit to let the server increment."
.to_string(),
required: false,
}],
example: json!({
"type": "etcd_put_response",
"revision": 2
}),
log_template: Some(
LogTemplate::new()
.with_info("-> etcd put ok (rev {revision})")
.with_debug("etcd etcd_put_response: revision={revision}"),
),
}
}
fn etcd_delete_range_response_action() -> ActionDefinition {
ActionDefinition {
name: "etcd_delete_range_response".to_string(),
description: "Acknowledge a DeleteRange, reporting how many keys were removed".to_string(),
parameters: vec![Parameter {
name: "deleted".to_string(),
type_hint: "number".to_string(),
description: "Number of keys deleted. etcdctl prints this as 'deleted N'.".to_string(),
required: true,
}],
example: json!({
"type": "etcd_delete_range_response",
"deleted": 1
}),
log_template: Some(
LogTemplate::new()
.with_info("-> etcd deleted {deleted} key(s)")
.with_debug("etcd etcd_delete_range_response: deleted={deleted}"),
),
}
}
fn etcd_txn_response_action() -> ActionDefinition {
ActionDefinition {
name: "etcd_txn_response".to_string(),
description: "Report whether a transaction's comparisons held. The per-operation results \
are returned empty: nested Range/Put/Delete results inside a Txn are not \
implemented."
.to_string(),
parameters: vec![Parameter {
name: "succeeded".to_string(),
type_hint: "boolean".to_string(),
description: "True if the comparisons held and the success branch was taken"
.to_string(),
required: true,
}],
example: json!({
"type": "etcd_txn_response",
"succeeded": true
}),
log_template: Some(
LogTemplate::new()
.with_info("-> etcd txn succeeded={succeeded}")
.with_debug("etcd etcd_txn_response: succeeded={succeeded}"),
),
}
}
fn etcd_error_action() -> ActionDefinition {
ActionDefinition {
name: "etcd_error".to_string(),
description: "Return an error response".to_string(),
parameters: vec![
Parameter {
name: "code".to_string(),
type_hint: "string".to_string(),
description: "Error code (e.g., 'KEY_NOT_FOUND', 'INVALID_ARGUMENT')".to_string(),
required: true,
},
Parameter {
name: "message".to_string(),
type_hint: "string".to_string(),
description: "Error message".to_string(),
required: true,
},
],
example: json!({
"type": "etcd_error",
"code": "KEY_NOT_FOUND",
"message": "etcdserver: key not found"
}),
log_template: Some(
LogTemplate::new()
.with_info("-> etcd error {code}: {message}")
.with_debug("etcd etcd_error: code={code} message={message}"),
),
}
}
impl Protocol for EtcdProtocol {
fn get_startup_parameters(&self) -> Vec<ParameterDefinition> {
vec![ParameterDefinition {
name: "cluster_name".to_string(),
type_hint: "string".to_string(),
description: "Cluster identifier name, used in log lines (default: netget-cluster)"
.to_string(),
required: false,
example: json!("my-cluster"),
default: None,
}]
}
fn get_async_actions(&self, _state: &AppState) -> Vec<ActionDefinition> {
vec![]
}
fn get_sync_actions(&self) -> Vec<ActionDefinition> {
vec![
etcd_range_response_action(),
etcd_put_response_action(),
etcd_delete_range_response_action(),
etcd_txn_response_action(),
etcd_error_action(),
]
}
fn protocol_name(&self) -> &'static str {
"etcd"
}
fn stack_name(&self) -> &'static str {
"ETH>IP>TCP>GRPC>ETCD"
}
fn keywords(&self) -> Vec<&'static str> {
vec!["etcd", "etcd3", "etcdv3", "etcd server"]
}
fn metadata(&self) -> crate::protocol::metadata::ProtocolMetadataV2 {
use crate::protocol::metadata::{DevelopmentState, ProtocolMetadataV2};
ProtocolMetadataV2::builder()
.state(DevelopmentState::Beta)
.well_known_port(2379)
.implementation(
"hyper HTTP/2 + prost, hand-routed gRPC; etcd protobuf schemas compiled by \
build.rs. tonic is a dependency but this server does not use it.",
)
.llm_control(
"Range, Put, DeleteRange and Txn. Compact is answered without consulting the \
handler; a Txn's nested operations are not executed, only its outcome.",
)
.e2e_testing(
"TWO independent clients, neither #[ignore]d and neither skipping when \
anything is missing. etcd-client (the official Rust client, tonic) in \
tests/server/etcd/e2e_test.rs completes put, get, prefix-get and delete and \
asserts on the decoded key/value pairs. The real etcdctl binary (grpc-go, the \
reference implementation) in tests/server/etcd/real_client_test.rs does the \
same four verbs through the CLI and asserts on what it printed; it hard-fails \
when etcdctl is absent rather than skipping. \
The second client is there because the first was not enough: until September \
2026 a successful reply put `grpc-status` in the INITIAL HEADERS and then sent \
a DATA frame, so the stream ended with no trailing HEADERS. tonic tolerated \
that; grpc-go answered `server closed the stream without sending trailers` to \
every RPC carrying a body -- the first Put included. Error replies were \
unaffected because they are Trailers-Only by construction, which is precisely \
why the whole suite passed. The success path now emits real HTTP/2 trailers \
and the error path stays Trailers-Only, and removing either is a red test. \
UNPROVEN: Watch, Lease, Auth, Cluster and Maintenance (none are routed -- they \
return UNIMPLEMENTED), Txn's nested operations, and binary keys or values, \
which cross the action boundary as lossy UTF-8.",
)
.notes(
"KV service only: no Watch, Lease, Auth, Cluster or Maintenance. No storage - \
the handler answers every request; the server keeps only a revision counter. \
Keys and values cross the action boundary as UTF-8 strings, so binary keys \
are lossy.",
)
.max_inbound_bytes(crate::server::etcd::MAX_REQUEST_BYTES)
.build()
}
fn description(&self) -> &'static str {
"etcd v3 distributed key-value store server"
}
fn get_event_types(&self) -> Vec<EventType> {
vec![
(*ETCD_RANGE_REQUEST_EVENT).clone(),
(*ETCD_PUT_REQUEST_EVENT).clone(),
(*ETCD_DELETE_REQUEST_EVENT).clone(),
(*ETCD_TXN_REQUEST_EVENT).clone(),
]
}
fn example_prompt(&self) -> &'static str {
r#"listen on port 2379 via etcd
Store configuration under /config/ prefix.
When clients PUT /config/database with value "localhost:5432", store it (revision 1).
When clients GET /config/database, return "localhost:5432" with revision metadata.
For unknown keys, return empty kvs array.
Examples:
- PUT /config/timeout = "30" → Success, increment revision
- GET /config/timeout → Return "30" with create_revision, mod_revision, version
- DELETE /config/timeout → Remove key, return deleted count
- Range query /config/ → Return all keys with /config/ prefix
Track revisions for MVCC."#
}
fn group_name(&self) -> &'static str {
"Database"
}
fn get_startup_examples(&self) -> crate::llm::actions::StartupExamples {
use crate::llm::actions::StartupExamples;
use serde_json::json;
let script = r#"import json, sys
data = json.load(sys.stdin)
event = data["event"]
if data["event_type_id"] == "etcd_range_request":
actions = [{"type": "etcd_range_response", "kvs": [], "count": 0}]
else:
actions = []
print(json.dumps({"actions": actions}))"#;
StartupExamples::new(
json!({
"type": "open_server",
"port": 2379,
"base_stack": "etcd",
"instruction": "etcd v3 key-value store handling Range, Put, Delete operations"
}),
json!({
"type": "open_server",
"port": 2379,
"base_stack": "etcd",
"event_handlers": [{
"event_pattern": "etcd_range_request",
"handler": {
"type": "script",
"language": "python",
"code": script
}
}]
}),
json!({
"type": "open_server",
"port": 2379,
"base_stack": "etcd",
"event_handlers": [{
"event_pattern": "etcd_range_request",
"handler": {
"type": "static",
"actions": [{
"type": "etcd_range_response",
"kvs": [],
"more": false,
"count": 0
}]
}
}]
}),
)
}
}
impl Server for EtcdProtocol {
fn spawn(
&self,
ctx: crate::protocol::SpawnContext,
) -> std::pin::Pin<
Box<dyn std::future::Future<Output = anyhow::Result<std::net::SocketAddr>> + Send>,
> {
Box::pin(async move {
use crate::server::etcd::EtcdServer;
EtcdServer::spawn_with_llm_actions(
ctx.legacy_listen_addr(),
ctx.llm_client,
ctx.state,
ctx.status_tx,
ctx.server_id,
ctx.startup_params,
)
.await
})
}
fn execute_action(&self, action: Value) -> Result<ActionResult> {
let action_type = action
.get("type")
.and_then(|v| v.as_str())
.ok_or_else(|| anyhow::anyhow!("Missing action type"))?;
match action_type {
"etcd_range_response"
| "etcd_put_response"
| "etcd_delete_range_response"
| "etcd_txn_response"
| "etcd_error" => Ok(ActionResult::Custom {
name: action_type.to_string(),
data: action,
}),
_ => anyhow::bail!("Unknown etcd action type: {}", action_type),
}
}
}