use crate::llm::actions::{
client_trait::{Client, ClientActionResult},
protocol_trait::Protocol,
ActionDefinition, Parameter, ParameterDefinition,
};
use crate::protocol::log_template::LogTemplate;
use crate::protocol::EventType;
use crate::server::coap::codec::{content_format_id, MAX_REPRESENTATION_LEN};
use crate::state::app_state::AppState;
use anyhow::{anyhow, Context, Result};
use serde_json::{json, Value};
use std::sync::LazyLock;
pub const REQUEST_RESULT: &str = "coap_request";
pub const DEFAULT_ACK_TIMEOUT_MS: u64 = 2000;
pub const DEFAULT_MAX_RETRANSMIT: u64 = 4;
pub const DEFAULT_BLOCK_SIZE: u64 = 1024;
pub const MAX_URI_OPTION_LEN: usize = 255;
fn param(name: &str, type_hint: &str, description: &str, required: bool) -> Parameter {
Parameter {
name: name.to_string(),
type_hint: type_hint.to_string(),
description: description.to_string(),
required,
}
}
fn response_params() -> Vec<Parameter> {
vec![
param(
"method",
"string",
"GET, POST, PUT or DELETE — the request answered",
true,
),
param(
"path",
"string",
"The request's path, e.g. /sensors/temp",
true,
),
param(
"code",
"string",
"The response code in c.dd form, e.g. 2.05",
true,
),
param(
"status",
"string",
"The code's name, e.g. Content, Changed, Created, Not Found",
true,
),
param(
"content_format",
"string",
"The payload's media type (text/plain, application/json, …), if the server named \
one",
false,
),
param(
"payload",
"string",
"The payload as text. Absent when empty or not UTF-8 (payload_size still says how \
long it was)",
false,
),
param(
"payload_json",
"any",
"The payload parsed, when the content format is application/json and it parses",
false,
),
param("payload_size", "number", "Payload length in bytes", true),
param(
"options",
"object",
"Other response options: max_age, etag (hex), location_path, observe, size2",
true,
),
]
}
pub static COAP_CONNECTED_EVENT: LazyLock<EventType> = LazyLock::new(|| {
EventType::new(
"coap_connected",
"Ready to send CoAP requests to the server (UDP has no handshake; nothing was sent)",
json!({"type": "coap_get", "path": "/.well-known/core"}),
)
.with_parameters(vec![param(
"remote_addr",
"string",
"The CoAP server requests go to",
true,
)])
});
pub static COAP_RESPONSE_EVENT: LazyLock<EventType> = LazyLock::new(|| {
let mut params = response_params();
params.push(param(
"observing",
"boolean",
"Present on responses to coap_observe / coap_observe_cancel: true while the \
observation is registered, false once it is cancelled",
false,
));
params.push(param(
"blocks",
"number",
"How many Block2 blocks the payload was reassembled from (1 when it fit in one)",
true,
));
EventType::new(
"coap_response",
"The server answered a request (after reassembling any Block2 transfer)",
json!({"type": "coap_put", "path": "/example_data", "payload": "hello"}),
)
.with_parameters(params)
});
pub static COAP_NOTIFICATION_EVENT: LazyLock<EventType> = LazyLock::new(|| {
let mut params = response_params();
params.push(param(
"sequence",
"number",
"The notification's Observe sequence number",
true,
));
EventType::new(
"coap_notification",
"An observed resource changed and the server sent its new state (RFC 7641)",
json!({"type": "coap_observe_cancel", "path": "/time"}),
)
.with_parameters(params)
});
pub static COAP_ERROR_EVENT: LazyLock<EventType> = LazyLock::new(|| {
EventType::new(
"coap_error",
"A request got no usable answer",
json!({"type": "coap_get", "path": "/"}),
)
.with_parameters(vec![
param(
"kind",
"string",
"timeout (no answer after every retransmission), reset (the server answered RST), \
body_too_large (a Block2 transfer passed the reassembly limit), bad_block (a \
Block2 block out of order), or not_observing (cancel for a path with no \
observation)",
true,
),
param("message", "string", "What went wrong", true),
param("method", "string", "The request's method", true),
param("path", "string", "The request's path", true),
])
});
pub struct CoapClientProtocol;
impl CoapClientProtocol {
pub fn new() -> Self {
Self
}
}
impl Default for CoapClientProtocol {
fn default() -> Self {
Self::new()
}
}
fn common_params(with_payload: bool) -> Vec<Parameter> {
let mut params = vec![
param("path", "string", "Resource path, e.g. /sensors/temp", true),
param(
"query",
"string",
"Query string without '?', items separated by '&', e.g. unit=c&n=3",
false,
),
param(
"confirmable",
"boolean",
"Send as Confirmable (retransmitted until acknowledged; default true) or \
Non-confirmable",
false,
),
];
if with_payload {
params.push(param(
"payload",
"any",
"The body: a string is sent as text, an object or array as JSON. Up to 64 KiB; \
over one block (block_size, default 1024 bytes) it is sent in Block1 blocks",
true,
));
params.push(param(
"content_format",
"string",
"text/plain, application/json, application/link-format, … (default: \
application/json for an object or array, text/plain for a string)",
false,
));
} else {
params.push(param(
"accept",
"string",
"The media type wanted back, e.g. application/json",
false,
));
}
params
}
fn all_actions() -> Vec<ActionDefinition> {
let def = |name: &str, description: &str, params: Vec<Parameter>, example: Value| {
let info = match name {
"coap_get" => "-> CoAP GET {path} {query}",
"coap_post" => "-> CoAP POST {path} {query} {preview(payload,60)}",
"coap_put" => "-> CoAP PUT {path} {query} {preview(payload,60)}",
"coap_delete" => "-> CoAP DELETE {path} {query}",
"coap_observe" => "-> CoAP observe {path}",
"coap_observe_cancel" => "-> CoAP cancel observation of {path}",
_ => "-> CoAP stop",
};
ActionDefinition {
name: name.to_string(),
description: description.to_string(),
parameters: params,
example,
log_template: Some(LogTemplate::new().with_info(info)),
}
};
vec![
def(
"coap_get",
"Read a resource",
common_params(false),
json!({"type": "coap_get", "path": "/sensors/temp"}),
),
def(
"coap_post",
"Send data to a resource (often creates one)",
common_params(true),
json!({"type": "coap_post", "path": "/logs", "payload": "door opened"}),
),
def(
"coap_put",
"Replace a resource's state",
common_params(true),
json!({"type": "coap_put", "path": "/example_data", "payload": "hello"}),
),
def(
"coap_delete",
"Delete a resource",
common_params(false),
json!({"type": "coap_delete", "path": "/example_data"}),
),
def(
"coap_observe",
"Register to be notified whenever the resource changes (RFC 7641). The first \
answer arrives as coap_response with observing true; every later change as \
coap_notification",
common_params(false),
json!({"type": "coap_observe", "path": "/time"}),
),
def(
"coap_observe_cancel",
"Cancel an observation made with coap_observe",
vec![param("path", "string", "The observed path", true)],
json!({"type": "coap_observe_cancel", "path": "/time"}),
),
def(
"disconnect",
"Stop: forget every exchange and observation and close the socket",
vec![],
json!({"type": "disconnect"}),
),
]
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CoapRequest {
pub method: &'static str,
pub path: Vec<String>,
pub query: Vec<String>,
pub confirmable: bool,
pub content_format: Option<u16>,
pub accept: Option<u16>,
pub payload: Vec<u8>,
pub observe: ObserveAction,
}
impl CoapRequest {
pub fn path_string(&self) -> String {
format!("/{}", self.path.join("/"))
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ObserveAction {
None,
Register,
Cancel,
}
fn split_path(path: &str) -> Result<Vec<String>> {
let segments: Vec<String> = path
.split('/')
.filter(|s| !s.is_empty())
.map(str::to_string)
.collect();
if let Some(long) = segments.iter().find(|s| s.len() > MAX_URI_OPTION_LEN) {
return Err(anyhow!(
"path segment of {} bytes; a Uri-Path option carries at most {MAX_URI_OPTION_LEN}",
long.len()
));
}
Ok(segments)
}
fn media_type(action: &Value, field: &str) -> Result<Option<u16>> {
match action.get(field) {
None | Some(Value::Null) => Ok(None),
Some(Value::String(name)) => content_format_id(name).map(Some).with_context(|| {
format!("unknown {field} {name:?}; use e.g. text/plain or application/json")
}),
Some(v) => v
.as_u64()
.and_then(|n| u16::try_from(n).ok())
.map(Some)
.with_context(|| format!("{field} must be a media type name or a number 0-65535")),
}
}
pub fn request_from_action(action: &Value) -> Result<Option<CoapRequest>> {
let action_type = action
.get("type")
.and_then(Value::as_str)
.context("missing 'type'")?;
let (method, observe, with_payload) = match action_type {
"coap_get" => ("GET", ObserveAction::None, false),
"coap_post" => ("POST", ObserveAction::None, true),
"coap_put" => ("PUT", ObserveAction::None, true),
"coap_delete" => ("DELETE", ObserveAction::None, false),
"coap_observe" => ("GET", ObserveAction::Register, false),
"coap_observe_cancel" => ("GET", ObserveAction::Cancel, false),
"disconnect" => return Ok(None),
other => return Err(anyhow!("Unknown CoAP client action: {other}")),
};
let path = split_path(
action
.get("path")
.and_then(Value::as_str)
.context("missing string field 'path'")?,
)?;
let query: Vec<String> = match action.get("query") {
None | Some(Value::Null) => Vec::new(),
Some(Value::String(q)) => q
.trim_start_matches('?')
.split('&')
.filter(|s| !s.is_empty())
.map(str::to_string)
.collect(),
Some(v) => return Err(anyhow!("'query' must be a string, got {v}")),
};
if let Some(long) = query.iter().find(|s| s.len() > MAX_URI_OPTION_LEN) {
return Err(anyhow!(
"query item of {} bytes; a Uri-Query option carries at most {MAX_URI_OPTION_LEN}",
long.len()
));
}
let confirmable = match action.get("confirmable") {
None | Some(Value::Null) => true,
Some(v) => v.as_bool().context("'confirmable' must be true or false")?,
};
let (payload, content_format) = if with_payload {
let explicit = media_type(action, "content_format")?;
match action.get("payload") {
Some(Value::String(text)) => (text.as_bytes().to_vec(), explicit.or(Some(0))),
Some(v @ (Value::Object(_) | Value::Array(_))) => {
(serde_json::to_vec(v)?, explicit.or(Some(50)))
}
Some(Value::Null) | None => {
return Err(anyhow!("{action_type} needs a 'payload' (text or JSON)"))
}
Some(other) => (other.to_string().into_bytes(), explicit.or(Some(50))),
}
} else {
(Vec::new(), None)
};
if payload.len() > MAX_REPRESENTATION_LEN {
return Err(anyhow!(
"payload is {} bytes; this client uploads at most {MAX_REPRESENTATION_LEN} (a body \
over one block is sent in Block1 blocks, RFC 7959)",
payload.len()
));
}
Ok(Some(CoapRequest {
method,
path,
query,
confirmable,
content_format,
accept: if with_payload {
None
} else {
media_type(action, "accept")?
},
payload,
observe,
}))
}
impl Protocol for CoapClientProtocol {
fn get_startup_parameters(&self) -> Vec<ParameterDefinition> {
vec![
ParameterDefinition {
name: "ack_timeout_ms".to_string(),
type_hint: "integer".to_string(),
description: "RFC 7252 ACK_TIMEOUT: milliseconds before the first \
retransmission of a Confirmable request (randomised up to 1.5x, \
doubling each time). Default 2000"
.to_string(),
required: false,
example: json!(2000),
default: Some(json!(DEFAULT_ACK_TIMEOUT_MS)),
},
ParameterDefinition {
name: "max_retransmit".to_string(),
type_hint: "integer".to_string(),
description: "RFC 7252 MAX_RETRANSMIT: how many times a Confirmable request is \
re-sent before it is reported as a timeout. Default 4, at most 8"
.to_string(),
required: false,
example: json!(4),
default: Some(json!(DEFAULT_MAX_RETRANSMIT)),
},
ParameterDefinition {
name: "block_size".to_string(),
type_hint: "integer".to_string(),
description: "Block1 block size in bytes: a request body larger than this is \
uploaded in blocks of this size (RFC 7959). A power of two from \
16 to 1024. A server that asks for smaller blocks gets them"
.to_string(),
required: false,
example: json!(1024),
default: Some(json!(DEFAULT_BLOCK_SIZE)),
},
]
}
fn get_async_actions(&self, _state: &AppState) -> Vec<ActionDefinition> {
all_actions()
}
fn get_sync_actions(&self) -> Vec<ActionDefinition> {
all_actions()
}
fn protocol_name(&self) -> &'static str {
"CoAP"
}
fn get_event_types(&self) -> Vec<EventType> {
vec![
COAP_CONNECTED_EVENT.clone(),
COAP_RESPONSE_EVENT.clone(),
COAP_NOTIFICATION_EVENT.clone(),
COAP_ERROR_EVENT.clone(),
]
}
fn stack_name(&self) -> &'static str {
"ETH>IP>UDP>CoAP"
}
fn keywords(&self) -> Vec<&'static str> {
vec![
"coap",
"coap client",
"constrained application protocol",
"iot client",
]
}
fn metadata(&self) -> crate::protocol::metadata::ProtocolMetadataV2 {
use crate::protocol::metadata::{
DevelopmentState, PrivilegeRequirement, ProtocolMetadataV2,
};
ProtocolMetadataV2::builder()
.state(DevelopmentState::Beta)
.privilege_requirement(PrivilegeRequirement::None)
.implementation(
"CoAP (RFC 7252) client over a connected UDP socket, encoding and decoding with \
the server's own codec (src/server/coap/codec.rs). A transport task owns the \
socket: message ids and tokens, Confirmable retransmission with RFC 7252 \
back-off and a bounded attempt count, empty ACKs for separate responses and \
notifications, RST for tokens it does not know, Block2 reassembly (RFC 7959) \
up to 64 KiB, and Observe registration and cancellation (RFC 7641). The model \
is asked from a separate turn task.",
)
.llm_control(
"Which requests to send: GET/POST/PUT/DELETE with path, query, content format \
and a text or JSON payload (Block1 over block_size), Confirmable or not, and \
Observe register/cancel. \
Every answer arrives as coap_response, coap_notification or coap_error.",
)
.e2e_testing(
"tests/client/coap/real_server_test.rs, 7+ LLM calls, against libcoap's \
coap-server (C), written and read back with libcoap's coap-client. coap-client \
stores 3000 bytes; the model GETs them and is shown one coap_response \
reassembled from three Block2 blocks, PUTs a sentence built from that body \
(checked byte for byte), which coap-client reads back, then observes /time: \
registration, notifications, cancellation on the same token. A second test \
injects a PUT through the command channel that coap-client reads back, and a \
third PUTs 3000 bytes the transport uploads as Block1 blocks, read back whole \
by coap-client; blockwise_upload_test.rs does the same against aiocoap's \
server and against a hand-written server that lowers the block size. Not \
#[ignore]d; a missing coap-server or coap-client fails the test. \
transport_test.rs drives retransmission, the Block2 cap and order, the \
oversize drop, deduplication, RST and the exchange cap against hand-written \
servers.",
)
.notes(
"No DTLS (coaps), no CoAP over TCP/WebSocket, no multicast. A request body is \
at most 64 KiB, uploaded in Block1 blocks over block_size. A notification \
over one block is reported truncated rather than fetched. Payloads reach the \
model as text or parsed JSON; a non-UTF-8 payload is reported by size only, \
never re-encoded.",
)
.build()
}
fn description(&self) -> &'static str {
"CoAP client for reading, writing and observing resources on constrained devices"
}
fn example_prompt(&self) -> &'static str {
"Connect to the CoAP server at localhost:5683, list its resources and observe /time"
}
fn group_name(&self) -> &'static str {
"Application"
}
fn get_startup_examples(&self) -> crate::llm::actions::StartupExamples {
use crate::llm::actions::StartupExamples;
StartupExamples::new(
json!({
"type": "open_client",
"remote_addr": "localhost:5683",
"base_stack": "coap",
"instruction": "List the server's resources and read each one"
}),
json!({
"type": "open_client",
"remote_addr": "localhost:5683",
"base_stack": "coap",
"event_handlers": [{
"event_pattern": "coap_notification",
"handler": {
"type": "script",
"language": "python",
"code": "<coap_client_handler>"
}
}]
}),
json!({
"type": "open_client",
"remote_addr": "localhost:5683",
"base_stack": "coap",
"event_handlers": [
{
"event_pattern": "coap_connected",
"handler": {
"type": "static",
"actions": [{"type": "coap_get", "path": "/.well-known/core"}]
}
},
{
"event_pattern": "coap_response",
"handler": {"type": "static", "actions": [{"type": "disconnect"}]}
}
]
}),
)
}
}
impl Client for CoapClientProtocol {
fn connect(
&self,
ctx: crate::protocol::ConnectContext,
) -> std::pin::Pin<
Box<dyn std::future::Future<Output = anyhow::Result<std::net::SocketAddr>> + Send>,
> {
Box::pin(async move {
let (ack_timeout_ms, max_retransmit, block_size) = match ctx.startup_params.as_ref() {
Some(p) => (
p.get_optional_u64("ack_timeout_ms")?
.unwrap_or(DEFAULT_ACK_TIMEOUT_MS),
p.get_optional_u64("max_retransmit")?
.unwrap_or(DEFAULT_MAX_RETRANSMIT),
p.get_optional_u64("block_size")?
.unwrap_or(DEFAULT_BLOCK_SIZE),
),
None => (
DEFAULT_ACK_TIMEOUT_MS,
DEFAULT_MAX_RETRANSMIT,
DEFAULT_BLOCK_SIZE,
),
};
let block_szx = (0..=crate::server::coap::codec::MAX_SZX)
.find(|szx| crate::server::coap::codec::szx_size(*szx) as u64 == block_size)
.ok_or_else(|| {
anyhow!("block_size {block_size} is not a power of two from 16 to 1024")
})?;
if !(10..=60_000).contains(&ack_timeout_ms) {
return Err(anyhow!(
"ack_timeout_ms {ack_timeout_ms} is outside 10-60000"
));
}
if max_retransmit > 8 {
return Err(anyhow!("max_retransmit {max_retransmit} is over 8"));
}
crate::client::coap::CoapClient::connect_with_llm_actions(
ctx.remote_addr,
crate::client::coap::Reliability {
ack_timeout: std::time::Duration::from_millis(ack_timeout_ms),
max_retransmit: u32::try_from(max_retransmit).unwrap_or(4),
block_szx,
},
ctx.llm_client,
ctx.state,
ctx.status_tx,
ctx.client_id,
)
.await
})
}
fn execute_action(&self, action: Value) -> Result<ClientActionResult> {
match request_from_action(&action)? {
None => Ok(ClientActionResult::Disconnect),
Some(_) => Ok(ClientActionResult::Custom {
name: REQUEST_RESULT.to_string(),
data: action,
}),
}
}
}