pub mod actions;
use anyhow::{Context, Result};
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::{mpsc, Mutex};
use tracing::{debug, error, info, warn};
use crate::llm::action_helper::call_llm;
use crate::llm::ollama_client::OllamaClient;
use crate::logging::emit::Log;
use crate::protocol::Event;
use crate::state::app_state::AppState;
use crate::{console_error, console_info, console_trace};
use actions::{
BluetoothBleProtocol, BLUETOOTH_BLE_STARTED_EVENT, BLUETOOTH_READ_REQUEST_EVENT,
BLUETOOTH_STATE_CHANGED_EVENT, BLUETOOTH_SUBSCRIBE_EVENT, BLUETOOTH_WRITE_REQUEST_EVENT,
};
#[cfg(all(unix, feature = "bluetooth"))]
use ble_peripheral_rust::gatt::characteristic::Characteristic;
#[cfg(all(unix, feature = "bluetooth"))]
use ble_peripheral_rust::gatt::peripheral_event::{
PeripheralEvent, ReadRequestResponse, RequestResponse, WriteRequestResponse,
};
#[cfg(all(unix, feature = "bluetooth"))]
use ble_peripheral_rust::gatt::properties::{AttributePermission, CharacteristicProperty};
#[cfg(all(unix, feature = "bluetooth"))]
use ble_peripheral_rust::gatt::service::Service;
#[cfg(all(unix, feature = "bluetooth"))]
use ble_peripheral_rust::{Peripheral, PeripheralImpl};
#[cfg(all(unix, feature = "bluetooth"))]
use uuid::Uuid;
const MAX_LOCAL_NAME_CHARS: usize = 29;
#[derive(Debug)]
struct CharacteristicData {
#[allow(dead_code)]
uuid: String,
#[allow(dead_code)]
properties: Vec<String>,
#[allow(dead_code)]
permissions: Vec<String>,
current_value: Vec<u8>,
}
struct ServerData {
hub: Option<Arc<BleHub>>,
event_tx: Option<mpsc::Sender<PeripheralEvent>>,
memory: String,
characteristics: HashMap<String, CharacteristicData>,
}
#[cfg(all(unix, feature = "bluetooth"))]
pub fn write_response_status(raw_actions: &[serde_json::Value]) -> RequestResponse {
let refused = raw_actions.iter().any(|action| {
matches!(
action.get("type").and_then(|v| v.as_str()),
Some("respond_to_write") | Some("send_write_response")
) && action
.get("status")
.and_then(|v| v.as_str())
.is_some_and(|status| status.eq_ignore_ascii_case("error"))
});
if refused {
RequestResponse::UnlikelyError
} else {
RequestResponse::Success
}
}
#[cfg(all(unix, feature = "bluetooth"))]
#[derive(Debug, PartialEq, Eq)]
pub enum ReadDecision {
Value(Vec<u8>),
UseStored,
Unusable(String),
}
#[cfg(all(unix, feature = "bluetooth"))]
pub fn read_decision(raw_actions: &[serde_json::Value]) -> ReadDecision {
let Some(action) = raw_actions.iter().find(|a| {
matches!(
a.get("type").and_then(|v| v.as_str()),
Some("respond_to_read") | Some("send_read_response")
)
}) else {
return ReadDecision::UseStored;
};
let Some(value) = action.get("value").and_then(|v| v.as_str()) else {
return ReadDecision::Unusable(
"respond_to_read carried no 'value' field (hex-encoded bytes)".to_string(),
);
};
match hex::decode(value.trim_start_matches("0x")) {
Ok(bytes) => ReadDecision::Value(bytes),
Err(e) => ReadDecision::Unusable(format!(
"respond_to_read 'value' is not hex-encoded ({e}); \
expected something like \"0048\" or \"0x0048\""
)),
}
}
#[cfg(all(unix, feature = "bluetooth"))]
pub fn parse_ble_uuid(s: &str) -> Result<Uuid> {
let t = s.trim();
let is_hex = |v: &str| v.chars().all(|c| c.is_ascii_hexdigit());
if (t.len() == 4 || t.len() == 8) && is_hex(t) {
let short = format!("{:0>8}", t.to_ascii_lowercase());
let full = format!("{short}-0000-1000-8000-00805f9b34fb");
return Uuid::parse_str(&full)
.with_context(|| format!("Invalid BLE short UUID {t:?} (expanded to {full})"));
}
Uuid::parse_str(t).with_context(|| {
format!(
"Invalid UUID {t:?}. Use a 16-bit shorthand like \"180D\", a 32-bit one, \
or a full 128-bit UUID."
)
})
}
#[cfg(all(unix, feature = "bluetooth"))]
pub fn characteristic_key(uuid: &str) -> String {
match parse_ble_uuid(uuid) {
Ok(parsed) => parsed.to_string(),
Err(_) => uuid.trim().to_lowercase(),
}
}
#[cfg(all(unix, feature = "bluetooth"))]
struct BleHub {
peripheral: Mutex<Peripheral>,
router: BleRouter,
}
#[cfg(all(unix, feature = "bluetooth"))]
#[derive(Default)]
pub struct BleRouter {
routes: std::sync::Mutex<HashMap<String, mpsc::Sender<PeripheralEvent>>>,
servers: std::sync::Mutex<Vec<mpsc::Sender<PeripheralEvent>>>,
}
#[cfg(all(unix, feature = "bluetooth"))]
static BLE_HUB: tokio::sync::OnceCell<Arc<BleHub>> = tokio::sync::OnceCell::const_new();
#[cfg(all(unix, feature = "bluetooth"))]
impl BleRouter {
pub fn new() -> Self {
Self::default()
}
pub fn register_server(&self, tx: mpsc::Sender<PeripheralEvent>) {
self.servers.lock().unwrap().push(tx);
}
pub fn register_characteristic(&self, char_uuid: &str, tx: mpsc::Sender<PeripheralEvent>) {
self.routes
.lock()
.unwrap()
.insert(characteristic_key(char_uuid), tx);
}
fn route_target(&self, char_uuid_lc: &str) -> Option<mpsc::Sender<PeripheralEvent>> {
if let Some(tx) = self.routes.lock().unwrap().get(char_uuid_lc) {
if !tx.is_closed() {
return Some(tx.clone());
}
}
self.servers
.lock()
.unwrap()
.iter()
.rev()
.find(|tx| !tx.is_closed())
.cloned()
}
fn live_servers(&self) -> Vec<mpsc::Sender<PeripheralEvent>> {
self.servers
.lock()
.unwrap()
.iter()
.filter(|tx| !tx.is_closed())
.cloned()
.collect()
}
pub async fn dispatch(&self, event: PeripheralEvent) {
match event {
PeripheralEvent::StateUpdate { is_powered } => {
for tx in self.live_servers() {
let _ = tx.send(PeripheralEvent::StateUpdate { is_powered }).await;
}
}
ev => {
let char_uuid_lc = match &ev {
PeripheralEvent::ReadRequest { request, .. }
| PeripheralEvent::WriteRequest { request, .. }
| PeripheralEvent::CharacteristicSubscriptionUpdate { request, .. } => {
characteristic_key(&request.characteristic.to_string())
}
PeripheralEvent::StateUpdate { .. } => unreachable!(),
};
match self.route_target(&char_uuid_lc) {
Some(tx) => {
let _ = tx.send(ev).await;
}
None => {
warn!(
"BLE event for characteristic {} has no live server to handle it; \
dropping",
char_uuid_lc
);
}
}
}
}
}
}
pub struct BluetoothBle;
impl BluetoothBle {
#[cfg(all(unix, feature = "bluetooth"))]
pub async fn spawn_with_llm_actions(
device_name: String,
llm_client: OllamaClient,
app_state: Arc<AppState>,
status_tx: mpsc::UnboundedSender<String>,
server_id: crate::state::ServerId,
instruction: String,
) -> Result<std::net::SocketAddr> {
let hub = Self::shared_hub(&status_tx)
.await
.context("Failed to bring up the shared BLE radio")?;
info!("Bluetooth server created on shared radio, adapter powered on");
Log::new(Some(&status_tx)).info(format!(
"Bluetooth server created for device '{}'",
device_name
));
let (event_tx, event_rx) = mpsc::channel::<PeripheralEvent>(256);
hub.router.register_server(event_tx.clone());
let server_data = Arc::new(Mutex::new(ServerData {
hub: Some(hub.clone()),
event_tx: Some(event_tx.clone()),
memory: String::new(),
characteristics: HashMap::new(),
}));
let protocol = Arc::new(BluetoothBleProtocol::new());
let started_event = Event::new(
&BLUETOOTH_BLE_STARTED_EVENT,
serde_json::json!({
"device_name": device_name,
"instruction": instruction,
}),
);
info!("Calling LLM for initial Bluetooth server configuration");
match call_llm(
&llm_client,
&app_state,
server_id,
None, &started_event,
protocol.as_ref(),
)
.await
{
Ok(llm_result) => {
let mut failed = 0usize;
let mut attempted = 0usize;
for action in llm_result.raw_actions {
attempted += 1;
debug!(
"Executing initial Bluetooth action: {:?}",
action.get("type")
);
if let Err(e) =
Self::execute_action(&server_data, &device_name, action, &status_tx).await
{
failed += 1;
Log::new(Some(&status_tx)).error(format!(
"Bluetooth startup for '{device_name}' \
decision=fail_closed_bad_action: initial action failed: {e}"
));
}
}
let log = Log::new(Some(&status_tx));
if attempted == 0 {
log.warn(format!(
"Bluetooth startup for '{device_name}' decision=model_silent: the \
model configured nothing, so the adapter is powered with no services \
and is NOT advertising"
));
} else if failed == attempted {
log.error(format!(
"Bluetooth startup for '{device_name}' \
decision=fail_closed_no_action: every initial action failed; the \
adapter is powered with no services and is NOT advertising"
));
} else {
log.info(format!(
"Bluetooth startup for '{device_name}' decision=model_answer: \
{} of {attempted} initial action(s) applied",
attempted - failed
));
}
}
Err(e) => {
let decision = if crate::utils::WireFailure::classify(&e).is_overloaded() {
"fail_closed_llm_overloaded"
} else {
"fail_closed_llm_error"
};
error!(
"Bluetooth startup configuration failed for '{}' decision={}: {}; the \
adapter is powered but has no services and is NOT advertising",
device_name, decision, e
);
Log::new(Some(&status_tx)).error(format!(
"Bluetooth startup for '{device_name}' decision={decision}: the adapter \
is up but no services were added and it is not advertising."
));
}
}
let llm_client_clone = llm_client.clone();
let app_state_clone = app_state.clone();
let status_tx_clone = status_tx.clone();
let server_data_clone = server_data.clone();
let protocol_clone = protocol.clone();
let task_owner = app_state.clone();
task_owner
.spawn_server_task(server_id, async move {
Self::event_loop(
event_rx,
server_id,
llm_client_clone,
app_state_clone,
status_tx_clone,
server_data_clone,
protocol_clone,
)
.await;
})
.await;
Ok(std::net::SocketAddr::from((
std::net::Ipv4Addr::UNSPECIFIED,
0,
)))
}
#[cfg(all(unix, feature = "bluetooth"))]
async fn shared_hub(status_tx: &mpsc::UnboundedSender<String>) -> Result<Arc<BleHub>> {
let status_tx = status_tx.clone();
BLE_HUB
.get_or_try_init(move || async move {
info!("Creating shared BLE peripheral (first BLE server in this process)");
let (event_tx, mut event_rx) = mpsc::channel::<PeripheralEvent>(256);
let mut peripheral = Peripheral::new(event_tx)
.await
.context("Failed to create BLE peripheral")?;
let mut retries = 0;
while !peripheral.is_powered().await.unwrap_or(false) {
if retries == 0 {
Log::new(Some(&status_tx))
.warn("Bluetooth adapter is not powered on, waiting...");
}
tokio::time::sleep(tokio::time::Duration::from_millis(500)).await;
retries += 1;
if retries > 20 {
anyhow::bail!("Bluetooth adapter failed to power on after 10 seconds");
}
}
info!("Bluetooth adapter powered on (shared radio)");
Log::new(Some(&status_tx)).info("Bluetooth adapter powered on");
let hub = Arc::new(BleHub {
peripheral: Mutex::new(peripheral),
router: BleRouter::new(),
});
let dispatch_hub = hub.clone();
tokio::spawn(async move {
while let Some(event) = event_rx.recv().await {
dispatch_hub.router.dispatch(event).await;
}
warn!("BLE shared radio event stream ended; no further events will be routed");
});
Ok::<Arc<BleHub>, anyhow::Error>(hub)
})
.await
.cloned()
}
#[cfg(all(unix, feature = "bluetooth"))]
async fn execute_action(
server_data: &Arc<Mutex<ServerData>>,
device_name: &str,
action: serde_json::Value,
status_tx: &mpsc::UnboundedSender<String>,
) -> Result<()> {
let action_type = action["type"]
.as_str()
.context("Action must have 'type' field")?;
match action_type {
"add_service" => Self::execute_add_service(server_data, action, status_tx).await,
"start_advertising" => {
Self::execute_start_advertising(server_data, device_name, action, status_tx).await
}
"stop_advertising" => Self::execute_stop_advertising(server_data, status_tx).await,
"send_notification" => {
Self::execute_send_notification(server_data, action, status_tx).await
}
"respond_to_read" | "send_read_response" => {
Ok(())
}
"respond_to_write" | "send_write_response" => {
Ok(())
}
_ => {
Err(anyhow::anyhow!(
"Unknown Bluetooth action type {action_type:?}; nothing was done. Known \
actions: add_service, start_advertising, stop_advertising, \
send_notification, respond_to_read, respond_to_write."
))
}
}
}
#[cfg(all(unix, feature = "bluetooth"))]
async fn execute_add_service(
server_data: &Arc<Mutex<ServerData>>,
action: serde_json::Value,
status_tx: &mpsc::UnboundedSender<String>,
) -> Result<()> {
let uuid_str = action["uuid"]
.as_str()
.context("add_service requires 'uuid' field")?;
let primary = action["primary"].as_bool().unwrap_or(true);
let uuid = parse_ble_uuid(uuid_str).context("Invalid service UUID")?;
let chars_json = action["characteristics"]
.as_array()
.context("add_service requires 'characteristics' array")?;
let mut characteristics = Vec::new();
let mut char_uuids: Vec<String> = Vec::new();
let mut server_data_guard = server_data.lock().await;
for char_json in chars_json {
let char_uuid_str = char_json["uuid"]
.as_str()
.context("characteristic requires 'uuid' field")?;
let char_uuid = parse_ble_uuid(char_uuid_str).context("Invalid characteristic UUID")?;
let props_json = char_json["properties"]
.as_array()
.context("characteristic requires 'properties' array")?;
let mut properties = Vec::new();
for prop in props_json {
let prop_str = prop.as_str().context("property must be string")?;
properties.push(match prop_str.to_lowercase().as_str() {
"read" => CharacteristicProperty::Read,
"write" => CharacteristicProperty::Write,
"notify" => CharacteristicProperty::Notify,
"indicate" => CharacteristicProperty::Indicate,
"write_without_response" => CharacteristicProperty::WriteWithoutResponse,
_ => {
warn!("Unknown property: {}, defaulting to Read", prop_str);
CharacteristicProperty::Read
}
});
}
let empty_perms = Vec::new();
let perms_json = char_json["permissions"].as_array().unwrap_or(&empty_perms);
let mut permissions = Vec::new();
for perm in perms_json {
let perm_str = perm.as_str().context("permission must be string")?;
permissions.push(match perm_str.to_lowercase().as_str() {
"readable" => AttributePermission::Readable,
"writeable" => AttributePermission::Writeable,
_ => {
warn!("Unknown permission: {}", perm_str);
continue;
}
});
}
let initial_value = parse_initial_value(&char_uuid_str, &char_json["initial_value"])?;
server_data_guard.characteristics.insert(
characteristic_key(char_uuid_str),
CharacteristicData {
uuid: char_uuid_str.to_string(),
properties: props_json
.iter()
.filter_map(|v| v.as_str().map(String::from))
.collect(),
permissions: perms_json
.iter()
.filter_map(|v| v.as_str().map(String::from))
.collect(),
current_value: initial_value.clone(),
},
);
#[allow(unused_mut)]
let mut cached_value = if initial_value.is_empty() {
None
} else {
Some(initial_value)
};
#[cfg(target_vendor = "apple")]
if cached_value.is_some() {
let read_only = properties
.iter()
.all(|p| matches!(p, CharacteristicProperty::Read))
&& permissions
.iter()
.all(|p| matches!(p, AttributePermission::Readable));
if !read_only {
warn!(
"CoreBluetooth: ignoring initial_value on characteristic {} - a cached \
value is only legal on a read-only characteristic. Reads are answered \
via bluetooth_read_request instead.",
char_uuid_str
);
Log::new(Some(&status_tx)).warn(format!(
"BLE {}: initial_value ignored (macOS allows a cached value only \
on read-only characteristics); reads are answered by the LLM",
char_uuid_str
));
cached_value = None;
}
}
characteristics.push(Characteristic {
uuid: char_uuid,
properties,
permissions,
value: cached_value,
descriptors: Vec::new(), });
char_uuids.push(char_uuid_str.to_string());
}
let hub = server_data_guard.hub.clone();
let event_tx = server_data_guard.event_tx.clone();
drop(server_data_guard);
if let (Some(hub), Some(tx)) = (hub.as_ref(), event_tx.as_ref()) {
for cu in &char_uuids {
hub.router.register_characteristic(cu, tx.clone());
}
}
let service = Service {
uuid,
primary,
characteristics,
};
if let Some(hub) = hub.as_ref() {
hub.peripheral
.lock()
.await
.add_service(&service)
.await
.context("Failed to add service to peripheral")?;
Log::new(Some(&status_tx)).info(format!(
"Added BLE service {} with {} characteristics",
uuid_str,
chars_json.len()
));
}
Ok(())
}
#[cfg(all(unix, feature = "bluetooth"))]
async fn execute_start_advertising(
server_data: &Arc<Mutex<ServerData>>,
device_name: &str,
action: serde_json::Value,
status_tx: &mpsc::UnboundedSender<String>,
) -> Result<()> {
let name = crate::utils::sanitize::token(
action["device_name"].as_str().unwrap_or(device_name),
MAX_LOCAL_NAME_CHARS,
);
let name = name.as_str();
let service_uuids: Vec<Uuid> = action["service_uuids"]
.as_array()
.map(|arr| {
arr.iter()
.filter_map(|v| v.as_str())
.filter_map(|s| parse_ble_uuid(s).ok())
.collect()
})
.unwrap_or_else(Vec::new);
let hub = server_data.lock().await.hub.clone();
if let Some(hub) = hub.as_ref() {
hub.peripheral
.lock()
.await
.start_advertising(name, &service_uuids)
.await
.context("Failed to start advertising")?;
console_info!(
status_tx,
"Started BLE advertising as '{}' with {} service(s)",
name,
service_uuids.len()
);
} else {
Log::new(Some(status_tx)).warn(format!(
"start_advertising as '{name}' had no BLE radio attached; the device is NOT \
discoverable"
));
}
Ok(())
}
#[cfg(all(unix, feature = "bluetooth"))]
async fn execute_stop_advertising(
server_data: &Arc<Mutex<ServerData>>,
status_tx: &mpsc::UnboundedSender<String>,
) -> Result<()> {
let hub = server_data.lock().await.hub.clone();
if let Some(hub) = hub.as_ref() {
hub.peripheral
.lock()
.await
.stop_advertising()
.await
.context("Failed to stop advertising")?;
Log::new(Some(&status_tx)).info("Stopped BLE advertising");
}
Ok(())
}
#[cfg(all(unix, feature = "bluetooth"))]
async fn execute_send_notification(
server_data: &Arc<Mutex<ServerData>>,
action: serde_json::Value,
status_tx: &mpsc::UnboundedSender<String>,
) -> Result<()> {
let char_uuid_str = action["characteristic_uuid"]
.as_str()
.context("send_notification requires 'characteristic_uuid' field")?;
let value_str = action["value"]
.as_str()
.context("send_notification requires 'value' field (hex-encoded)")?;
let char_uuid = parse_ble_uuid(char_uuid_str).context("Invalid characteristic UUID")?;
let value_str = value_str.trim_start_matches("0x");
let value = hex::decode(value_str).context("Value must be hex-encoded")?;
let hub = {
let mut server_data_guard = server_data.lock().await;
if let Some(char_data) = server_data_guard
.characteristics
.get_mut(&characteristic_key(char_uuid_str))
{
char_data.current_value = value.clone();
}
server_data_guard.hub.clone()
};
if let Some(hub) = hub.as_ref() {
hub.peripheral
.lock()
.await
.update_characteristic(char_uuid, value.clone())
.await
.context("Failed to send notification")?;
Log::new(Some(&status_tx)).debug(format!(
"Sent BLE notification on {} ({} bytes)",
char_uuid_str,
value.len()
));
}
Ok(())
}
#[cfg(all(unix, feature = "bluetooth"))]
pub async fn run_event_loop_without_radio(
event_rx: mpsc::Receiver<PeripheralEvent>,
server_id: crate::state::ServerId,
llm_client: OllamaClient,
app_state: Arc<AppState>,
status_tx: mpsc::UnboundedSender<String>,
seed_characteristics: Vec<(String, Vec<u8>)>,
) {
let characteristics = seed_characteristics
.into_iter()
.map(|(uuid, current_value)| {
(
characteristic_key(&uuid),
CharacteristicData {
uuid: uuid.clone(),
properties: vec!["read".to_string(), "write".to_string()],
permissions: vec!["readable".to_string(), "writable".to_string()],
current_value,
},
)
})
.collect();
let server_data = Arc::new(Mutex::new(ServerData {
hub: None,
event_tx: None,
memory: String::new(),
characteristics,
}));
Self::event_loop(
event_rx,
server_id,
llm_client,
app_state,
status_tx,
server_data,
Arc::new(BluetoothBleProtocol::new()),
)
.await;
}
#[cfg(all(unix, feature = "bluetooth"))]
async fn event_loop(
mut event_rx: mpsc::Receiver<PeripheralEvent>,
server_id: crate::state::ServerId,
llm_client: OllamaClient,
app_state: Arc<AppState>,
status_tx: mpsc::UnboundedSender<String>,
server_data: Arc<Mutex<ServerData>>,
protocol: Arc<BluetoothBleProtocol>,
) {
while let Some(event) = event_rx.recv().await {
match event {
PeripheralEvent::StateUpdate { is_powered, .. } => {
Log::new(Some(&status_tx))
.info(format!("Bluetooth state update: powered = {}", is_powered));
let llm_event = Event::new(
&BLUETOOTH_STATE_CHANGED_EVENT,
serde_json::json!({
"state": if is_powered { "powered_on" } else { "powered_off" },
}),
);
if let Err(e) = Self::call_llm_for_event(
&server_id,
&llm_client,
&app_state,
&status_tx,
&server_data,
&protocol,
llm_event,
)
.await
{
console_error!(
status_tx,
"BLE adapter state change (powered = {}) was not handled: the \
handler failed ({}). Nothing was reconfigured.",
is_powered,
e
);
}
}
PeripheralEvent::ReadRequest {
request,
offset,
responder,
} => {
let char_uuid_str = request.characteristic.to_string();
Log::new(Some(&status_tx)).debug(format!(
"BLE read request on {} (offset: {})",
char_uuid_str, offset
));
let llm_event = Event::new(
&BLUETOOTH_READ_REQUEST_EVENT,
serde_json::json!({
"characteristic_uuid": char_uuid_str,
"offset": offset,
}),
);
match Self::call_llm_for_event(
&server_id,
&llm_client,
&app_state,
&status_tx,
&server_data,
&protocol,
llm_event,
)
.await
{
Ok(llm_result) => {
let reply = match read_decision(&llm_result.raw_actions) {
ReadDecision::Value(value) => {
Log::new(Some(&status_tx)).debug(format!(
"BLE read of {} answered by handler \
(decision=model_value, {} bytes)",
char_uuid_str,
value.len()
));
ReadRequestResponse {
value,
response: RequestResponse::Success,
}
}
ReadDecision::UseStored => {
let stored = {
let guard = server_data.lock().await;
guard
.characteristics
.get(&characteristic_key(&char_uuid_str))
.map(|c| c.current_value.clone())
};
match stored {
Some(value) => {
Log::new(Some(&status_tx)).debug(format!(
"BLE read of {} not answered by the \
handler (decision=model_silent); \
serving the stored value ({} bytes)",
char_uuid_str,
value.len()
));
ReadRequestResponse {
value,
response: RequestResponse::Success,
}
}
None => {
console_error!(
status_tx,
"BLE read of {} could not be answered \
(decision=fail_closed_model_silent_no_\
value): the handler named no value and \
this characteristic has none stored. \
Replying ATT Unlikely Error (0x0E).",
char_uuid_str
);
ReadRequestResponse {
value: Vec::new(),
response: RequestResponse::UnlikelyError,
}
}
}
}
ReadDecision::Unusable(reason) => {
console_error!(
status_tx,
"BLE read of {} could not be answered \
(decision=fail_closed_bad_value): {}. Replying \
ATT Unlikely Error (0x0E); the stored value is \
NOT substituted for an answer that failed to \
decode.",
char_uuid_str,
reason
);
ReadRequestResponse {
value: Vec::new(),
response: RequestResponse::UnlikelyError,
}
}
};
let _ = responder.send(reply);
}
Err(e) => {
console_error!(
status_tx,
"BLE read of {} could not be answered \
(decision=fail_closed_llm_error): the handler failed \
({}). Replying ATT Unlikely Error (0x0E); no \
characteristic value is being invented.",
char_uuid_str,
e
);
let _ = responder.send(ReadRequestResponse {
value: Vec::new(),
response: RequestResponse::UnlikelyError,
});
}
}
}
PeripheralEvent::WriteRequest {
request,
value,
offset,
responder,
} => {
let char_uuid_str = request.characteristic.to_string();
let value_hex = hex::encode(&value);
Log::new(Some(&status_tx)).debug(format!(
"BLE write request on {} ({} bytes)",
char_uuid_str,
value.len()
));
console_trace!(status_tx, "BLE write data (hex): {}", value_hex);
{
let mut guard = server_data.lock().await;
if let Some(char_data) = guard
.characteristics
.get_mut(&characteristic_key(&char_uuid_str))
{
char_data.current_value = value.clone();
}
}
let llm_event = Event::new(
&BLUETOOTH_WRITE_REQUEST_EVENT,
serde_json::json!({
"characteristic_uuid": char_uuid_str,
"value": value_hex,
"offset": offset,
}),
);
match Self::call_llm_for_event(
&server_id,
&llm_client,
&app_state,
&status_tx,
&server_data,
&protocol,
llm_event,
)
.await
{
Ok(llm_result) => {
let response = write_response_status(&llm_result.raw_actions);
if matches!(response, RequestResponse::UnlikelyError) {
Log::new(Some(&status_tx)).info(format!(
"BLE write to {} rejected by handler \
(decision=model_reject); replying ATT Unlikely \
Error (0x0E)",
char_uuid_str
));
} else if !llm_result.raw_actions.iter().any(|a| {
matches!(
a.get("type").and_then(|v| v.as_str()),
Some("respond_to_write") | Some("send_write_response")
)
}) {
Log::new(Some(&status_tx)).debug(format!(
"BLE write to {} not answered by the handler \
(decision=model_silent); the value was stored, so \
replying ATT Success",
char_uuid_str
));
}
let _ = responder.send(WriteRequestResponse { response });
}
Err(e) => {
console_error!(
status_tx,
"BLE write to {} could not be answered: the handler \
failed ({}). Replying ATT Unlikely Error (0x0E); the \
write is NOT acknowledged.",
char_uuid_str,
e
);
let _ = responder.send(WriteRequestResponse {
response: RequestResponse::UnlikelyError,
});
}
}
}
PeripheralEvent::CharacteristicSubscriptionUpdate {
request,
subscribed,
} => {
let char_uuid_str = request.characteristic.to_string();
if subscribed {
console_info!(
status_tx,
"Client subscribed to notifications on {}",
char_uuid_str
);
} else {
console_info!(
status_tx,
"Client unsubscribed from notifications on {}",
char_uuid_str
);
}
let llm_event = Event::new(
&BLUETOOTH_SUBSCRIBE_EVENT,
serde_json::json!({
"characteristic_uuid": char_uuid_str,
"subscribed": subscribed,
}),
);
if let Err(e) = Self::call_llm_for_event(
&server_id,
&llm_client,
&app_state,
&status_tx,
&server_data,
&protocol,
llm_event,
)
.await
{
console_error!(
status_tx,
"BLE subscription change on {} (subscribed = {}) was not handled: \
the handler failed ({}). No notifications will be sent for it.",
char_uuid_str,
subscribed,
e
);
}
}
}
}
}
#[cfg(all(unix, feature = "bluetooth"))]
async fn call_llm_for_event(
server_id: &crate::state::ServerId,
llm_client: &OllamaClient,
app_state: &Arc<AppState>,
status_tx: &mpsc::UnboundedSender<String>,
server_data: &Arc<Mutex<ServerData>>,
protocol: &Arc<BluetoothBleProtocol>,
event: Event,
) -> Result<crate::llm::actions::executor::ExecutionResult> {
let _memory = server_data.lock().await.memory.clone();
let llm_result = call_llm(
llm_client,
app_state,
*server_id,
None, &event,
protocol.as_ref(),
)
.await?;
for action in &llm_result.raw_actions {
let action_type = action.get("type").and_then(|v| v.as_str()).unwrap_or("");
match action_type {
"respond_to_read"
| "send_read_response"
| "respond_to_write"
| "send_write_response" => {
continue;
}
_ => {
if let Err(e) =
Self::execute_action(server_data, "NetGet-BLE", action.clone(), status_tx)
.await
{
error!("Failed to execute action: {}", e);
}
}
}
}
Ok(llm_result)
}
}
#[cfg(any(all(unix, not(feature = "bluetooth")), not(unix)))]
impl BluetoothBle {
pub async fn spawn_with_llm_actions(
_device_name: String,
_llm_client: OllamaClient,
_app_state: Arc<AppState>,
_status_tx: mpsc::UnboundedSender<String>,
_server_id: crate::state::ServerId,
_instruction: String,
) -> Result<std::net::SocketAddr> {
anyhow::bail!(
"Bluetooth server support not enabled - compile with --features bluetooth-ble"
)
}
}
pub fn parse_initial_value(char_uuid: &str, value: &serde_json::Value) -> Result<Vec<u8>> {
let Some(val_str) = value.as_str() else {
return Ok(Vec::new());
};
hex::decode(val_str.trim_start_matches("0x")).with_context(|| {
format!(
"characteristic {char_uuid}: initial_value {val_str:?} is not hex-encoded bytes \
(expected something like \"0048\" or \"0x0048\")"
)
})
}