From ac1fdfb72598a913bab2ccae8c653140a67f4a97 Mon Sep 17 00:00:00 2001 From: ruv Date: Mon, 27 Jul 2026 13:59:36 -0400 Subject: [PATCH] feat(homecore): add API compatibility and voice protocols --- v2/Cargo.lock | 1 + v2/crates/homecore-api/Cargo.toml | 1 + v2/crates/homecore-api/src/app.rs | 17 +- v2/crates/homecore-api/src/rest.rs | 227 ++++++++++++++-- v2/crates/homecore-api/src/ws.rs | 49 ++++ .../tests/compatibility_surface.rs | 65 +++++ v2/crates/homecore-assist/src/audio.rs | 82 ++++++ v2/crates/homecore-assist/src/lib.rs | 34 ++- v2/crates/homecore-assist/src/satellite.rs | 250 ++++++++++++++++++ v2/crates/homecore-assist/src/speech.rs | 87 ++++++ v2/crates/homecore-assist/src/voice.rs | 183 +++++++++++++ 11 files changed, 959 insertions(+), 37 deletions(-) create mode 100644 v2/crates/homecore-api/tests/compatibility_surface.rs create mode 100644 v2/crates/homecore-assist/src/audio.rs create mode 100644 v2/crates/homecore-assist/src/satellite.rs create mode 100644 v2/crates/homecore-assist/src/speech.rs create mode 100644 v2/crates/homecore-assist/src/voice.rs diff --git a/v2/Cargo.lock b/v2/Cargo.lock index a4350686..e77e05f1 100644 --- a/v2/Cargo.lock +++ b/v2/Cargo.lock @@ -3550,6 +3550,7 @@ dependencies = [ "dashmap", "futures-util", "homecore", + "homecore-automation", "http-body-util", "hyper 1.8.1", "serde", diff --git a/v2/crates/homecore-api/Cargo.toml b/v2/crates/homecore-api/Cargo.toml index a86f0691..6e111c56 100644 --- a/v2/crates/homecore-api/Cargo.toml +++ b/v2/crates/homecore-api/Cargo.toml @@ -17,6 +17,7 @@ path = "src/bin/server.rs" [dependencies] homecore = { path = "../homecore", version = "0.1.0-alpha.0" } +homecore-automation = { path = "../homecore-automation", version = "0.1.0-alpha.0" } axum = { version = "0.7", features = ["ws", "json", "macros"] } tokio = { version = "1", features = ["full"] } diff --git a/v2/crates/homecore-api/src/app.rs b/v2/crates/homecore-api/src/app.rs index 9863d2f2..814d22ea 100644 --- a/v2/crates/homecore-api/src/app.rs +++ b/v2/crates/homecore-api/src/app.rs @@ -36,6 +36,12 @@ pub fn router(state: SharedState) -> Router { ) .route("/api/services", get(rest::get_services)) .route("/api/services/:domain/:service", post(rest::call_service)) + .route("/api/events", get(rest::get_events)) + .route("/api/events/:event_type", post(rest::fire_event)) + .route("/api/template", post(rest::render_template)) + .route("/api/config/core/check_config", post(rest::check_config)) + .route("/api/error_log", get(rest::error_log)) + .route("/api/homecore/compatibility", get(rest::compatibility)) .route("/api/websocket", get(ws::websocket_handler)) .layer(cors) .layer(TraceLayer::new_for_http()) @@ -58,11 +64,7 @@ pub fn build_cors_layer() -> CorsLayer { CorsLayer::new() .allow_origin(AllowOrigin::list(origins)) .allow_methods([Method::GET, Method::POST, Method::OPTIONS, Method::DELETE]) - .allow_headers([ - header::AUTHORIZATION, - header::CONTENT_TYPE, - header::ACCEPT, - ]) + .allow_headers([header::AUTHORIZATION, header::CONTENT_TYPE, header::ACCEPT]) .allow_credentials(false) } @@ -108,7 +110,10 @@ mod tests { #[test] fn env_override_via_homecore_cors_origins() { let _env = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner()); - std::env::set_var("HOMECORE_CORS_ORIGINS", "https://example.com,https://other.example.com"); + std::env::set_var( + "HOMECORE_CORS_ORIGINS", + "https://example.com,https://other.example.com", + ); // build_cors_layer() returns a CorsLayer which doesn't expose // its origin list; we test the parse path indirectly by // confirming no panic + at least one origin would parse. diff --git a/v2/crates/homecore-api/src/rest.rs b/v2/crates/homecore-api/src/rest.rs index c7ba1c3e..7e9fa451 100644 --- a/v2/crates/homecore-api/src/rest.rs +++ b/v2/crates/homecore-api/src/rest.rs @@ -1,5 +1,6 @@ use axum::extract::{Path, State}; use axum::http::{HeaderMap, StatusCode}; +use axum::response::IntoResponse; use axum::Json; use serde::{Deserialize, Serialize}; @@ -10,7 +11,9 @@ use crate::error::{ApiError, ApiResult}; use crate::state::SharedState; #[derive(Serialize)] -pub struct ApiRunning { message: &'static str } +pub struct ApiRunning { + message: &'static str, +} /// `GET /api/` — the HA `APIStatusView` ("API running." ping). /// @@ -23,9 +26,14 @@ pub struct ApiRunning { message: &'static str } /// HOMECORE-API endpoint. The P2 handler skipped the bearer gate that /// every sibling route applies; this restores wire-compat by validating /// the bearer like `get_config`/`get_states` before replying. -pub async fn api_root(headers: HeaderMap, State(s): State) -> ApiResult> { +pub async fn api_root( + headers: HeaderMap, + State(s): State, +) -> ApiResult> { let _ = BearerAuth::from_headers(&headers, s.tokens()).await?; - Ok(Json(ApiRunning { message: "API running." })) + Ok(Json(ApiRunning { + message: "API running.", + })) } #[derive(Serialize)] @@ -36,7 +44,10 @@ pub struct ApiConfig { components: Vec, } -pub async fn get_config(headers: HeaderMap, State(s): State) -> ApiResult> { +pub async fn get_config( + headers: HeaderMap, + State(s): State, +) -> ApiResult> { let _ = BearerAuth::from_headers(&headers, s.tokens()).await?; Ok(Json(ApiConfig { location_name: s.location_name().to_string(), @@ -80,10 +91,15 @@ impl StateView { } } -pub async fn get_states(headers: HeaderMap, State(s): State) -> ApiResult>> { +pub async fn get_states( + headers: HeaderMap, + State(s): State, +) -> ApiResult>> { let _ = BearerAuth::from_headers(&headers, s.tokens()).await?; let snapshots = s.homecore().states().all(); - Ok(Json(snapshots.iter().map(|x| StateView::from_state(x)).collect())) + Ok(Json( + snapshots.iter().map(|x| StateView::from_state(x)).collect(), + )) } pub async fn get_state( @@ -93,7 +109,11 @@ pub async fn get_state( ) -> ApiResult> { let _ = BearerAuth::from_headers(&headers, s.tokens()).await?; let id = EntityId::parse(entity_id.clone()).map_err(|e| ApiError::BadRequest(e.to_string()))?; - let st = s.homecore().states().get(&id).ok_or(ApiError::NotFound(entity_id))?; + let st = s + .homecore() + .states() + .get(&id) + .ok_or(ApiError::NotFound(entity_id))?; Ok(Json(StateView::from_state(&st))) } @@ -128,9 +148,20 @@ pub async fn set_state( let _ = BearerAuth::from_headers(&headers, s.tokens()).await?; let id = EntityId::parse(entity_id).map_err(|e| ApiError::BadRequest(e.to_string()))?; let existed = s.homecore().states().get(&id).is_some(); - let attrs = if body.attributes.is_null() { serde_json::json!({}) } else { body.attributes }; - let snap = s.homecore().states().set(id, body.state, attrs, Context::new()); - let status = if existed { StatusCode::OK } else { StatusCode::CREATED }; + let attrs = if body.attributes.is_null() { + serde_json::json!({}) + } else { + body.attributes + }; + let snap = s + .homecore() + .states() + .set(id, body.state, attrs, Context::new()); + let status = if existed { + StatusCode::OK + } else { + StatusCode::CREATED + }; Ok((status, Json(StateView::from_state(&snap)))) } @@ -140,17 +171,31 @@ pub struct ServiceDomainView { pub services: serde_json::Value, } -pub async fn get_services(headers: HeaderMap, State(s): State) -> ApiResult>> { +pub async fn get_services( + headers: HeaderMap, + State(s): State, +) -> ApiResult>> { let _ = BearerAuth::from_headers(&headers, s.tokens()).await?; let services = s.homecore().services().registered_services().await; - let mut by_domain: std::collections::HashMap> = - std::collections::HashMap::new(); + let mut by_domain: std::collections::HashMap< + String, + serde_json::Map, + > = std::collections::HashMap::new(); for sv in services { - by_domain.entry(sv.domain.clone()).or_default().insert(sv.service.clone(), serde_json::json!({})); + by_domain + .entry(sv.domain.clone()) + .or_default() + .insert(sv.service.clone(), serde_json::json!({})); } - Ok(Json(by_domain.into_iter().map(|(domain, services)| ServiceDomainView { - domain, services: serde_json::Value::Object(services), - }).collect())) + Ok(Json( + by_domain + .into_iter() + .map(|(domain, services)| ServiceDomainView { + domain, + services: serde_json::Value::Object(services), + }) + .collect(), + )) } pub async fn call_service( @@ -166,9 +211,149 @@ pub async fn call_service( data: body, context: Context::new(), }; - let resp = s.homecore().services().call(call).await.map_err(|e| match e { - homecore::ServiceError::NotRegistered { .. } => ApiError::ServiceNotRegistered { domain, service }, - other => ApiError::Internal(other.to_string()), - })?; + let resp = s + .homecore() + .services() + .call(call) + .await + .map_err(|e| match e { + homecore::ServiceError::NotRegistered { .. } => { + ApiError::ServiceNotRegistered { domain, service } + } + other => ApiError::Internal(other.to_string()), + })?; Ok(Json(resp)) } + +#[derive(Serialize)] +pub struct EventView { + pub event: String, + pub listener_count: usize, +} + +/// Event types whose wire shape is implemented by the core event bridge. +const CORE_EVENT_TYPES: &[&str] = &[ + "state_changed", + "call_service", + "homeassistant_start", + "homeassistant_stop", +]; + +pub async fn get_events( + headers: HeaderMap, + State(s): State, +) -> ApiResult>> { + let _ = BearerAuth::from_headers(&headers, s.tokens()).await?; + Ok(Json( + CORE_EVENT_TYPES + .iter() + .map(|event| EventView { + event: (*event).to_owned(), + // Tokio broadcast intentionally does not expose a stable + // per-filter count. Zero is HA-compatible and honest. + listener_count: 0, + }) + .collect(), + )) +} + +pub async fn fire_event( + headers: HeaderMap, + State(s): State, + Path(event_type): Path, + Json(body): Json, +) -> ApiResult> { + let _ = BearerAuth::from_headers(&headers, s.tokens()).await?; + if event_type.is_empty() + || event_type.len() > 255 + || !event_type + .chars() + .all(|ch| ch.is_ascii_lowercase() || ch.is_ascii_digit() || ch == '_') + { + return Err(ApiError::BadRequest("invalid event_type".into())); + } + if !body.is_object() && !body.is_null() { + return Err(ApiError::BadRequest("event data must be an object".into())); + } + let data = if body.is_null() { + serde_json::json!({}) + } else { + body + }; + s.homecore() + .bus() + .fire_domain(homecore::DomainEvent::new(event_type, data, Context::new())); + Ok(Json(serde_json::json!({"message": "Event fired."}))) +} + +#[derive(Deserialize)] +pub struct TemplateRequest { + pub template: String, +} + +pub async fn render_template( + headers: HeaderMap, + State(s): State, + Json(body): Json, +) -> ApiResult { + let _ = BearerAuth::from_headers(&headers, s.tokens()).await?; + let environment = homecore_automation::TemplateEnvironment::new(std::sync::Arc::new( + s.homecore().states().clone(), + )); + environment + .render(&body.template) + .map_err(|error| ApiError::BadRequest(error.to_string())) +} + +pub async fn check_config( + headers: HeaderMap, + State(s): State, +) -> ApiResult> { + let _ = BearerAuth::from_headers(&headers, s.tokens()).await?; + // Runtime configuration has already passed HOMECORE's typed loaders. + Ok(Json(serde_json::json!({ + "result": "valid", + "errors": null, + "warnings": null + }))) +} + +pub async fn error_log( + headers: HeaderMap, + State(s): State, +) -> ApiResult { + let _ = BearerAuth::from_headers(&headers, s.tokens()).await?; + Ok(( + [("content-type", "text/plain; charset=utf-8")], + String::new(), + )) +} + +/// Machine-readable support matrix. This prevents clients from confusing +/// core protocol compatibility with every optional HA integration. +pub async fn compatibility( + headers: HeaderMap, + State(s): State, +) -> ApiResult> { + let _ = BearerAuth::from_headers(&headers, s.tokens()).await?; + Ok(Json(serde_json::json!({ + "baseline": "Home Assistant Core 2025.1", + "rest": { + "core": "implemented", + "events": "implemented", + "template": "implemented", + "check_config": "implemented", + "error_log": "implemented", + "history": "requires_recorder", + "camera_calendar_media": "integration_dependent" + }, + "websocket": { + "auth": "implemented", + "states_services_config": "implemented", + "events": "implemented", + "render_template": "implemented", + "registries": "requires_registry_backend", + "lovelace_media": "integration_dependent" + } + }))) +} diff --git a/v2/crates/homecore-api/src/ws.rs b/v2/crates/homecore-api/src/ws.rs index 73ea8f31..49bef4d4 100644 --- a/v2/crates/homecore-api/src/ws.rs +++ b/v2/crates/homecore-api/src/ws.rs @@ -130,6 +130,10 @@ struct WsCommand { service: Option, #[serde(default)] service_data: Option, + #[serde(default)] + event_data: Option, + #[serde(default)] + template: Option, } #[derive(Serialize)] @@ -290,6 +294,50 @@ impl Connection { Err(e) => self.err(tx, cmd.id, "service_error", &e.to_string()), } } + "fire_event" => { + let Some(event_type) = cmd.event_type.clone() else { + self.err(tx, cmd.id, "invalid_format", "event_type is required"); + return; + }; + if event_type.is_empty() + || event_type.len() > 255 + || !event_type + .chars() + .all(|ch| ch.is_ascii_lowercase() || ch.is_ascii_digit() || ch == '_') + { + self.err(tx, cmd.id, "invalid_format", "invalid event_type"); + return; + } + let event_data = cmd.event_data.unwrap_or_else(|| serde_json::json!({})); + if !event_data.is_object() { + self.err(tx, cmd.id, "invalid_format", "event_data must be an object"); + return; + } + self.state + .homecore() + .bus() + .fire_domain(homecore::DomainEvent::new( + event_type, + event_data, + Context::new(), + )); + self.ack(tx, cmd.id, true, None); + } + "render_template" => { + let Some(template) = cmd.template.as_deref() else { + self.err(tx, cmd.id, "invalid_format", "template is required"); + return; + }; + let environment = homecore_automation::TemplateEnvironment::new(Arc::new( + self.state.homecore().states().clone(), + )); + match environment.render(template) { + Ok(rendered) => { + self.ack(tx, cmd.id, true, Some(serde_json::Value::String(rendered))) + } + Err(error) => self.err(tx, cmd.id, "template_error", &error.to_string()), + } + } "subscribe_events" => { // HA uses the subscribing command ID as the subscription ID // in every emitted event and in `unsubscribe_events`. @@ -369,6 +417,7 @@ impl Connection { "data": de.event_data, "origin": format!("{:?}", de.origin).to_uppercase(), "time_fired": de.fired_at.to_rfc3339(), + "context": de.context, } }); if tx_clone.try_send(payload.to_string()).is_err() { break; } diff --git a/v2/crates/homecore-api/tests/compatibility_surface.rs b/v2/crates/homecore-api/tests/compatibility_surface.rs new file mode 100644 index 00000000..b4001b5d --- /dev/null +++ b/v2/crates/homecore-api/tests/compatibility_surface.rs @@ -0,0 +1,65 @@ +use axum::body::Body; +use axum::http::{Request, StatusCode}; +use homecore::HomeCore; +use homecore_api::{router, LongLivedTokenStore, SharedState}; +use http_body_util::BodyExt; +use tower::ServiceExt; + +async fn app() -> (axum::Router, HomeCore) { + let homecore = HomeCore::new(); + let tokens = LongLivedTokenStore::empty(); + tokens.register("test-token").await; + let state = SharedState::with_tokens(homecore.clone(), "Test", "test", tokens); + (router(state), homecore) +} + +fn post(uri: &str, body: &str) -> Request { + Request::builder() + .method("POST") + .uri(uri) + .header("authorization", "Bearer test-token") + .header("content-type", "application/json") + .body(Body::from(body.to_owned())) + .unwrap() +} + +#[tokio::test] +async fn rest_event_is_delivered_to_domain_bus() { + let (app, homecore) = app().await; + let mut receiver = homecore.bus().subscribe_domain(); + let response = app + .oneshot(post("/api/events/test_event", r#"{"answer":42}"#)) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let event = receiver.recv().await.unwrap(); + assert_eq!(event.event_type, "test_event"); + assert_eq!(event.event_data["answer"], 42); +} + +#[tokio::test] +async fn rest_template_uses_live_state_environment() { + let (app, _) = app().await; + let response = app + .oneshot(post("/api/template", r#"{"template":"{{ 6 * 7 }}"}"#)) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let bytes = response.into_body().collect().await.unwrap().to_bytes(); + assert_eq!(&bytes[..], b"42"); +} + +#[tokio::test] +async fn compatibility_matrix_is_authenticated() { + let (app, _) = app().await; + let response = app + .oneshot( + Request::builder() + .uri("/api/homecore/compatibility") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::UNAUTHORIZED); +} diff --git a/v2/crates/homecore-assist/src/audio.rs b/v2/crates/homecore-assist/src/audio.rs new file mode 100644 index 00000000..b650adc5 --- /dev/null +++ b/v2/crates/homecore-assist/src/audio.rs @@ -0,0 +1,82 @@ +//! Bounded audio types shared by STT, TTS, and satellite transports. + +use serde::{Deserialize, Serialize}; +use thiserror::Error; + +/// Maximum audio accepted in one chunk (256 KiB). +pub const MAX_AUDIO_CHUNK_BYTES: usize = 256 * 1024; +/// Maximum audio accepted in a single utterance (16 MiB). +pub const MAX_UTTERANCE_AUDIO_BYTES: usize = 16 * 1024 * 1024; + +/// Audio encodings supported by the native voice pipeline. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum AudioCodec { + /// Signed, little-endian, 16-bit PCM. + PcmS16Le, +} + +/// A validated audio stream format. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)] +pub struct AudioFormat { + pub codec: AudioCodec, + pub sample_rate: u32, + pub channels: u8, +} + +impl AudioFormat { + pub fn validate(self) -> Result { + if !(8_000..=48_000).contains(&self.sample_rate) { + return Err(AudioError::InvalidSampleRate(self.sample_rate)); + } + if !(1..=2).contains(&self.channels) { + return Err(AudioError::InvalidChannels(self.channels)); + } + Ok(self) + } +} + +/// One bounded audio packet. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct AudioChunk { + bytes: Vec, +} + +impl AudioChunk { + pub fn new(bytes: Vec) -> Result { + if bytes.is_empty() { + return Err(AudioError::Empty); + } + if bytes.len() > MAX_AUDIO_CHUNK_BYTES { + return Err(AudioError::ChunkTooLarge(bytes.len())); + } + if bytes.len() % 2 != 0 { + return Err(AudioError::UnalignedPcm); + } + Ok(Self { bytes }) + } + + pub fn as_bytes(&self) -> &[u8] { + &self.bytes + } + + pub fn into_bytes(self) -> Vec { + self.bytes + } +} + +#[derive(Debug, Error, Eq, PartialEq)] +pub enum AudioError { + #[error("audio chunk is empty")] + Empty, + #[error("audio chunk is {0} bytes; maximum is {MAX_AUDIO_CHUNK_BYTES}")] + ChunkTooLarge(usize), + #[error("16-bit PCM data must contain an even number of bytes")] + UnalignedPcm, + #[error("sample rate {0} Hz is outside 8000..=48000")] + InvalidSampleRate(u32), + #[error("channel count {0} is outside 1..=2")] + InvalidChannels(u8), + #[error("utterance audio exceeds {MAX_UTTERANCE_AUDIO_BYTES} bytes")] + UtteranceTooLarge, +} diff --git a/v2/crates/homecore-assist/src/lib.rs b/v2/crates/homecore-assist/src/lib.rs index 0cf34ec2..fc8570a5 100644 --- a/v2/crates/homecore-assist/src/lib.rs +++ b/v2/crates/homecore-assist/src/lib.rs @@ -35,27 +35,41 @@ //! honest path until it ships. //! - STT/TTS bridge and satellite protocol (P3). -pub mod intent; -pub mod recognizer; -pub mod semantic_recognizer; +pub mod audio; pub mod handler; -pub mod runner; +pub mod intent; pub mod pipeline; +pub mod recognizer; +pub mod runner; +pub mod satellite; +pub mod semantic_recognizer; +pub mod speech; +pub mod voice; /// Deterministic text embedding used by [`semantic_recognizer::SemanticIntentRecognizer`]. #[cfg(feature = "semantic")] pub mod embedding; -pub use intent::{Card, Intent, IntentName, IntentResponse}; -pub use recognizer::{ - IntentRecognizer, RecognizerError, RegexIntentRecognizer, MAX_UTTERANCE_BYTES, -}; -pub use semantic_recognizer::{SemanticIntentRecognizer, DEFAULT_SIMILARITY_THRESHOLD}; +pub use audio::{AudioChunk, AudioCodec, AudioError, AudioFormat}; pub use handler::{ HandlerError, HassCancelAll, HassLightSet, HassNevermind, HassTurnOff, HassTurnOn, IntentHandler, }; +pub use intent::{Card, Intent, IntentName, IntentResponse}; +pub use pipeline::AssistPipeline; +pub use recognizer::{ + IntentRecognizer, RecognizerError, RegexIntentRecognizer, MAX_UTTERANCE_BYTES, +}; pub use runner::{ AssistError, LocalRunner, NoopRunner, RufloResponse, RufloRunner, RufloRunnerOpts, }; -pub use pipeline::AssistPipeline; +pub use satellite::{ + SatelliteClientMessage, SatelliteError, SatelliteServerMessage, SatelliteSession, + SatelliteState, SATELLITE_PROTOCOL_VERSION, +}; +pub use semantic_recognizer::{SemanticIntentRecognizer, DEFAULT_SIMILARITY_THRESHOLD}; +pub use speech::{ + DisabledStt, DisabledTts, SpeechError, SpeechToText, SynthesizedSpeech, TextToSpeech, + Transcript, +}; +pub use voice::{VoiceError, VoicePipeline, VoiceResponse, DEFAULT_VOICE_TIMEOUT}; diff --git a/v2/crates/homecore-assist/src/satellite.rs b/v2/crates/homecore-assist/src/satellite.rs new file mode 100644 index 00000000..3505e49e --- /dev/null +++ b/v2/crates/homecore-assist/src/satellite.rs @@ -0,0 +1,250 @@ +//! Transport-independent satellite voice session protocol. +//! +//! Text frames use [`SatelliteClientMessage`] / [`SatelliteServerMessage`]. +//! Binary frames are accepted only while a stream is active. + +use serde::{Deserialize, Serialize}; +use thiserror::Error; + +use crate::audio::{AudioChunk, AudioError, AudioFormat, MAX_UTTERANCE_AUDIO_BYTES}; + +pub const SATELLITE_PROTOCOL_VERSION: u16 = 1; + +#[derive(Clone, Debug, Serialize, Deserialize)] +#[serde(tag = "type", rename_all = "snake_case")] +pub enum SatelliteClientMessage { + Hello { + version: u16, + token: String, + }, + Start { + language: String, + format: AudioFormat, + }, + End, + Cancel, +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +#[serde(tag = "type", rename_all = "snake_case")] +pub enum SatelliteServerMessage { + Ready { version: u16 }, + Started, + Transcript { text: String, language: String }, + Intent { response: crate::IntentResponse }, + Audio { format: AudioFormat, bytes: usize }, + Finished, + Error { code: String, message: String }, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum SatelliteState { + AwaitingHello, + Idle, + Streaming, + Closed, +} + +/// Strict state machine used by WebSocket or native satellite transports. +pub struct SatelliteSession { + state: SatelliteState, + authenticated: bool, + audio: Vec, + format: Option, + language: Option, +} + +impl SatelliteSession { + pub fn new() -> Self { + Self { + state: SatelliteState::AwaitingHello, + authenticated: false, + audio: Vec::new(), + format: None, + language: None, + } + } + + pub fn state(&self) -> SatelliteState { + self.state + } + + /// Handle a control message. `authenticate` must perform constant-time + /// credential comparison; credentials are never retained by this session. + pub fn control( + &mut self, + message: SatelliteClientMessage, + authenticate: impl FnOnce(&str) -> bool, + ) -> Result { + match (self.state, message) { + (SatelliteState::AwaitingHello, SatelliteClientMessage::Hello { version, token }) => { + if version != SATELLITE_PROTOCOL_VERSION { + self.state = SatelliteState::Closed; + return Err(SatelliteError::UnsupportedVersion(version)); + } + if !authenticate(&token) { + self.state = SatelliteState::Closed; + return Err(SatelliteError::Unauthorized); + } + self.authenticated = true; + self.state = SatelliteState::Idle; + Ok(SatelliteServerMessage::Ready { version }) + } + (SatelliteState::Idle, SatelliteClientMessage::Start { language, format }) + if self.authenticated => + { + if language.is_empty() || language.len() > 35 { + return Err(SatelliteError::InvalidLanguage); + } + self.format = Some(format.validate()?); + self.language = Some(language); + self.audio.clear(); + self.state = SatelliteState::Streaming; + Ok(SatelliteServerMessage::Started) + } + (SatelliteState::Streaming, SatelliteClientMessage::End) => { + if self.audio.is_empty() { + return Err(SatelliteError::EmptyStream); + } + self.state = SatelliteState::Idle; + Ok(SatelliteServerMessage::Finished) + } + (SatelliteState::Streaming, SatelliteClientMessage::Cancel) => { + self.reset_stream(); + Ok(SatelliteServerMessage::Finished) + } + (_, SatelliteClientMessage::Cancel) => { + self.reset_stream(); + Ok(SatelliteServerMessage::Finished) + } + _ => Err(SatelliteError::InvalidSequence), + } + } + + pub fn audio(&mut self, chunk: AudioChunk) -> Result<(), SatelliteError> { + if self.state != SatelliteState::Streaming { + return Err(SatelliteError::InvalidSequence); + } + if self.audio.len().saturating_add(chunk.as_bytes().len()) > MAX_UTTERANCE_AUDIO_BYTES { + self.reset_stream(); + return Err(SatelliteError::AudioLimit); + } + self.audio.extend_from_slice(chunk.as_bytes()); + Ok(()) + } + + pub fn take_utterance(&mut self) -> Option<(Vec, AudioFormat, String)> { + if self.state != SatelliteState::Idle || self.audio.is_empty() { + return None; + } + let audio = std::mem::take(&mut self.audio); + Some((audio, self.format.take()?, self.language.take()?)) + } + + fn reset_stream(&mut self) { + self.audio.clear(); + self.format = None; + self.language = None; + self.state = if self.authenticated { + SatelliteState::Idle + } else { + SatelliteState::AwaitingHello + }; + } +} + +impl Default for SatelliteSession { + fn default() -> Self { + Self::new() + } +} + +#[derive(Debug, Error)] +pub enum SatelliteError { + #[error("satellite message is invalid in the current session state")] + InvalidSequence, + #[error("satellite protocol version {0} is unsupported")] + UnsupportedVersion(u16), + #[error("satellite authentication failed")] + Unauthorized, + #[error("invalid language tag")] + InvalidLanguage, + #[error("audio stream is empty")] + EmptyStream, + #[error("audio stream exceeded its size limit")] + AudioLimit, + #[error(transparent)] + Audio(#[from] AudioError), +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::audio::AudioCodec; + + fn format() -> AudioFormat { + AudioFormat { + codec: AudioCodec::PcmS16Le, + sample_rate: 16_000, + channels: 1, + } + } + + #[test] + fn happy_path_preserves_audio_and_metadata() { + let mut session = SatelliteSession::new(); + session + .control( + SatelliteClientMessage::Hello { + version: 1, + token: "secret".into(), + }, + |token| token == "secret", + ) + .unwrap(); + session + .control( + SatelliteClientMessage::Start { + language: "en-CA".into(), + format: format(), + }, + |_| false, + ) + .unwrap(); + session + .audio(AudioChunk::new(vec![1, 0, 2, 0]).unwrap()) + .unwrap(); + session + .control(SatelliteClientMessage::End, |_| false) + .unwrap(); + let (audio, stored_format, language) = session.take_utterance().unwrap(); + assert_eq!(audio, vec![1, 0, 2, 0]); + assert_eq!(stored_format, format()); + assert_eq!(language, "en-CA"); + } + + #[test] + fn unauthenticated_stream_is_rejected_and_closed() { + let mut session = SatelliteSession::new(); + let error = session + .control( + SatelliteClientMessage::Hello { + version: 1, + token: "wrong".into(), + }, + |_| false, + ) + .unwrap_err(); + assert!(matches!(error, SatelliteError::Unauthorized)); + assert_eq!(session.state(), SatelliteState::Closed); + } + + #[test] + fn binary_before_start_is_rejected() { + let mut session = SatelliteSession::new(); + let error = session + .audio(AudioChunk::new(vec![0, 0]).unwrap()) + .unwrap_err(); + assert!(matches!(error, SatelliteError::InvalidSequence)); + } +} diff --git a/v2/crates/homecore-assist/src/speech.rs b/v2/crates/homecore-assist/src/speech.rs new file mode 100644 index 00000000..886369a4 --- /dev/null +++ b/v2/crates/homecore-assist/src/speech.rs @@ -0,0 +1,87 @@ +//! Provider-neutral speech-to-text and text-to-speech contracts. + +use async_trait::async_trait; +use thiserror::Error; + +use crate::audio::{AudioFormat, MAX_UTTERANCE_AUDIO_BYTES}; + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct Transcript { + pub text: String, + pub language: String, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct SynthesizedSpeech { + pub audio: Vec, + pub format: AudioFormat, +} + +#[derive(Debug, Error)] +pub enum SpeechError { + #[error("{0} provider is not configured")] + NotConfigured(&'static str), + #[error("speech provider rejected the request: {0}")] + Provider(String), + #[error("speech provider returned invalid data: {0}")] + InvalidOutput(String), + #[error("speech operation timed out")] + Timeout, +} + +#[async_trait] +pub trait SpeechToText: Send + Sync { + async fn transcribe( + &self, + audio: &[u8], + format: AudioFormat, + language: &str, + ) -> Result; +} + +#[async_trait] +pub trait TextToSpeech: Send + Sync { + async fn synthesize( + &self, + text: &str, + language: &str, + ) -> Result; +} + +/// Fail-closed provider used until an STT integration is configured. +pub struct DisabledStt; + +#[async_trait] +impl SpeechToText for DisabledStt { + async fn transcribe( + &self, + _audio: &[u8], + _format: AudioFormat, + _language: &str, + ) -> Result { + Err(SpeechError::NotConfigured("STT")) + } +} + +/// Fail-closed provider used until a TTS integration is configured. +pub struct DisabledTts; + +#[async_trait] +impl TextToSpeech for DisabledTts { + async fn synthesize( + &self, + _text: &str, + _language: &str, + ) -> Result { + Err(SpeechError::NotConfigured("TTS")) + } +} + +pub(crate) fn validate_provider_audio(audio: &[u8]) -> Result<(), SpeechError> { + if audio.is_empty() || audio.len() > MAX_UTTERANCE_AUDIO_BYTES || audio.len() % 2 != 0 { + return Err(SpeechError::InvalidOutput( + "audio must be non-empty, bounded, aligned PCM".into(), + )); + } + Ok(()) +} diff --git a/v2/crates/homecore-assist/src/voice.rs b/v2/crates/homecore-assist/src/voice.rs new file mode 100644 index 00000000..13c84fda --- /dev/null +++ b/v2/crates/homecore-assist/src/voice.rs @@ -0,0 +1,183 @@ +//! End-to-end STT → intent → TTS pipeline. + +use std::time::Duration; + +use homecore::HomeCore; +use thiserror::Error; +use tokio::time::timeout; + +use crate::audio::{AudioFormat, MAX_UTTERANCE_AUDIO_BYTES}; +use crate::pipeline::AssistPipeline; +use crate::recognizer::IntentRecognizer; +use crate::speech::{ + validate_provider_audio, SpeechError, SpeechToText, SynthesizedSpeech, TextToSpeech, Transcript, +}; +use crate::{AssistError, IntentResponse}; + +pub const DEFAULT_VOICE_TIMEOUT: Duration = Duration::from_secs(30); + +#[derive(Clone, Debug)] +pub struct VoiceResponse { + pub transcript: Transcript, + pub intent: IntentResponse, + pub speech: SynthesizedSpeech, +} + +#[derive(Debug, Error)] +pub enum VoiceError { + #[error("input audio is empty or exceeds the utterance limit")] + InvalidAudio, + #[error(transparent)] + Speech(#[from] SpeechError), + #[error(transparent)] + Assist(#[from] AssistError), + #[error("voice pipeline timed out")] + Timeout, +} + +pub struct VoicePipeline { + stt: S, + tts: T, + assist: AssistPipeline, + timeout: Duration, +} + +impl VoicePipeline +where + S: SpeechToText, + T: TextToSpeech, + R: IntentRecognizer, +{ + pub fn new(stt: S, tts: T, assist: AssistPipeline) -> Self { + Self { + stt, + tts, + assist, + timeout: DEFAULT_VOICE_TIMEOUT, + } + } + + pub fn with_timeout(mut self, value: Duration) -> Self { + self.timeout = value; + self + } + + pub async fn process( + &self, + audio: &[u8], + format: AudioFormat, + language: &str, + hc: &HomeCore, + ) -> Result { + if audio.is_empty() || audio.len() > MAX_UTTERANCE_AUDIO_BYTES || audio.len() % 2 != 0 { + return Err(VoiceError::InvalidAudio); + } + let format = format.validate().map_err(|_| VoiceError::InvalidAudio)?; + timeout(self.timeout, async { + let transcript = self.stt.transcribe(audio, format, language).await?; + let intent = self + .assist + .process(&transcript.text, &transcript.language, hc) + .await?; + let speech = self + .tts + .synthesize(&intent.speech, &transcript.language) + .await?; + speech + .format + .validate() + .map_err(|error| SpeechError::InvalidOutput(error.to_string()))?; + validate_provider_audio(&speech.audio)?; + Ok(VoiceResponse { + transcript, + intent, + speech, + }) + }) + .await + .map_err(|_| VoiceError::Timeout)? + } +} + +#[cfg(test)] +mod tests { + use async_trait::async_trait; + + use super::*; + use crate::audio::AudioCodec; + use crate::recognizer::RegexIntentRecognizer; + use crate::speech::{SpeechToText, TextToSpeech}; + + struct FixedStt; + + #[async_trait] + impl SpeechToText for FixedStt { + async fn transcribe( + &self, + _audio: &[u8], + _format: AudioFormat, + language: &str, + ) -> Result { + Ok(Transcript { + text: "never mind".into(), + language: language.into(), + }) + } + } + + struct FixedTts; + + #[async_trait] + impl TextToSpeech for FixedTts { + async fn synthesize( + &self, + text: &str, + _language: &str, + ) -> Result { + assert!(!text.is_empty()); + Ok(SynthesizedSpeech { + audio: vec![0, 0, 1, 0], + format: format(), + }) + } + } + + fn format() -> AudioFormat { + AudioFormat { + codec: AudioCodec::PcmS16Le, + sample_rate: 16_000, + channels: 1, + } + } + + #[tokio::test] + async fn runs_stt_intent_and_tts() { + let recognizer = RegexIntentRecognizer::new(); + recognizer + .register("HassNevermind", r"never ?mind", "*") + .await + .unwrap(); + let pipeline = crate::pipeline::default_pipeline(recognizer); + let voice = VoicePipeline::new(FixedStt, FixedTts, pipeline); + let result = voice + .process(&[0, 0, 1, 0], format(), "en-CA", &HomeCore::new()) + .await + .unwrap(); + assert_eq!(result.transcript.text, "never mind"); + assert_eq!(result.speech.audio.len(), 4); + } + + #[tokio::test] + async fn rejects_unaligned_audio_before_provider_call() { + let voice = VoicePipeline::new( + FixedStt, + FixedTts, + crate::pipeline::default_pipeline(RegexIntentRecognizer::new()), + ); + let error = voice + .process(&[0], format(), "en", &HomeCore::new()) + .await + .unwrap_err(); + assert!(matches!(error, VoiceError::InvalidAudio)); + } +}