P0: Rust engine, wire daemon and audio-reactive descent surface

2026-11-01 00:36:04 +03:00
parent cd967cced4
commit db4368efae
68 changed files with 19507 additions and 3 deletions
@@ -0,0 +1,22 @@
+[package]
+name = "zhila-daemon"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+
+[[bin]]
+name = "zhila-daemon"
+path = "src/main.rs"
+
+[dependencies]
+zhila-core = { path = "../zhila-core" }
+zhila-engine = { path = "../zhila-engine" }
+zhila-protocol = { path = "../zhila-protocol" }
+tokio.workspace = true
+tokio-tungstenite.workspace = true
+futures-util.workspace = true
+serde_json.workspace = true
+anyhow.workspace = true
+clap.workspace = true
+tracing.workspace = true
+tracing-subscriber.workspace = true
@@ -0,0 +1,437 @@
+//! ZHILA local daemon: one WebSocket endpoint, one session, one engine.
+//!
+//! Text frames are JSON control messages; meter frames are binary. The daemon
+//! binds to loopback only and requires a token in the hello.
+
+use std::sync::Arc;
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
+
+use anyhow::Context;
+use clap::Parser;
+use futures_util::stream::SplitSink;
+use futures_util::{SinkExt, StreamExt};
+use tokio::net::{TcpListener, TcpStream};
+use tokio::sync::{Mutex, broadcast, mpsc};
+use tokio::time::{interval, timeout};
+use tokio_tungstenite::tungstenite::Message;
+use tokio_tungstenite::tungstenite::handshake::server::{ErrorResponse, Request, Response};
+use tokio_tungstenite::{WebSocketStream, accept_hdr_async};
+use tracing::{info, warn};
+use zhila_core::{CoreError, Engine, Session};
+use zhila_engine::{AudioEngine, NullEngine};
+use zhila_protocol::{
+    Actor, ClientMessage, CommandResult, ErrorCode, PROTOCOL_VERSION, ServerMessage,
+};
+
+const HELLO_TIMEOUT: Duration = Duration::from_secs(5);
+const METER_INTERVAL: Duration = Duration::from_millis(33);
+
+type WsSink = SplitSink<WebSocketStream<TcpStream>, Message>;
+
+#[derive(Debug, Parser)]
+#[command(
+    name = "zhila-daemon",
+    version,
+    about = "ZHILA local engine daemon (P0)"
+)]
+struct Args {
+    /// WebSocket port on loopback.
+    #[arg(long, default_value_t = 4317)]
+    port: u16,
+    /// Shared token for the hello. Dev default; change it for anything real.
+    #[arg(long, default_value = "dev")]
+    token: String,
+    /// Run without an output device; useful for UI tests.
+    #[arg(long)]
+    no_audio: bool,
+}
+
+#[derive(Clone)]
+enum Outbound {
+    Text(String),
+    Binary(Vec<u8>),
+}
+
+#[derive(Clone)]
+struct Shared {
+    session: Arc<Mutex<Session>>,
+    out: broadcast::Sender<Outbound>,
+    token: Arc<String>,
+    session_id: Arc<String>,
+}
+
+#[tokio::main(flavor = "current_thread")]
+async fn main() -> anyhow::Result<()> {
+    let filter = tracing_subscriber::EnvFilter::try_from_default_env()
+        .unwrap_or_else(|_| "zhila_daemon=info,zhila_engine=info".into());
+    tracing_subscriber::fmt()
+        .with_env_filter(filter)
+        .with_target(false)
+        .init();
+
+    let args = Args::parse();
+    let token = std::env::var("ZHILA_TOKEN").unwrap_or(args.token);
+
+    let pattern = zhila_core::default_pattern();
+    let audio_holder = if args.no_audio {
+        None
+    } else {
+        match AudioEngine::start(&pattern) {
+            Ok(engine) => Some(engine),
+            Err(error) => {
+                warn!("audio offline: {error}");
+                None
+            }
+        }
+    };
+    let control: Arc<dyn Engine> = match &audio_holder {
+        Some(engine) => engine.control(),
+        None => Arc::new(NullEngine::new()),
+    };
+    let shutdown = control.clone();
+    let info = control.info();
+    let session = Arc::new(Mutex::new(Session::with_pattern(control.clone(), pattern)));
+    let (out, _) = broadcast::channel::<Outbound>(128);
+    let session_id = format!("zhila-{:x}", now_ms());
+    let shared = Shared {
+        session,
+        out: out.clone(),
+        token: Arc::new(token.clone()),
+        session_id: Arc::new(session_id.clone()),
+    };
+
+    let meter_engine = control.clone();
+    let meter_out = out.clone();
+    tokio::spawn(async move {
+        let mut ticker = interval(METER_INTERVAL);
+        loop {
+            ticker.tick().await;
+            let frame = meter_engine.meter();
+            let _ = meter_out.send(Outbound::Binary(frame.encode()));
+        }
+    });
+
+    let listener = TcpListener::bind(("127.0.0.1", args.port))
+        .await
+        .with_context(|| format!("could not bind 127.0.0.1:{}", args.port))?;
+
+    print_banner(args.port, &token, &info, &session_id);
+    if token == "dev" {
+        warn!("using the development token; pass --token or ZHILA_TOKEN to change it");
+    }
+
+    loop {
+        tokio::select! {
+            accepted = listener.accept() => {
+                let (stream, peer) = accepted.context("accept failed")?;
+                let shared = shared.clone();
+                tokio::spawn(async move {
+                    if let Err(error) = handle(stream, shared).await {
+                        warn!("connection {peer}: {error:#}");
+                    }
+                });
+            }
+            _ = tokio::signal::ctrl_c() => {
+                info!("shutting down");
+                break;
+            }
+        }
+    }
+
+    shutdown.stop();
+    Ok(())
+}
+
+fn print_banner(port: u16, token: &str, info: &zhila_protocol::AudioInfo, session_id: &str) {
+    println!();
+    println!("  ZHILA · P0 engine");
+    println!("  ws://127.0.0.1:{port}  ·  token {token}");
+    if info.running {
+        let buffer = if info.buffer_frames == 0 {
+            "default".to_string()
+        } else {
+            info.buffer_frames.to_string()
+        };
+        println!(
+            "  audio {} · {} Hz · {} ch · buffer {}",
+            info.device, info.sample_rate, info.channels, buffer
+        );
+    } else {
+        println!("  audio offline · the surface will connect, but nothing will sound");
+    }
+    println!("  session {session_id}");
+    println!();
+}
+
+async fn handle(stream: TcpStream, shared: Shared) -> anyhow::Result<()> {
+    let peer = stream.peer_addr().ok();
+    let ws = accept_hdr_async(stream, handshake)
+        .await
+        .context("websocket handshake failed")?;
+
+    let (mut sink, mut source) = ws.split();
+
+    let first = match timeout(HELLO_TIMEOUT, source.next()).await {
+        Ok(Some(Ok(message))) => message,
+        Ok(Some(Err(error))) => return Err(error.into()),
+        Ok(None) => return Ok(()),
+        Err(_) => {
+            let _ = send_out(
+                &mut sink,
+                error_message(
+                    None,
+                    ErrorCode::Protocol,
+                    "expected a hello within 5 seconds",
+                ),
+            )
+            .await;
+            return Ok(());
+        }
+    };
+
+    let Message::Text(text) = first else {
+        let _ = send_out(
+            &mut sink,
+            error_message(None, ErrorCode::Protocol, "the first frame must be a hello"),
+        )
+        .await;
+        return Ok(());
+    };
+
+    let hello: ClientMessage = match serde_json::from_str(text.as_str()) {
+        Ok(message) => message,
+        Err(_) => {
+            let _ = send_out(
+                &mut sink,
+                error_message(None, ErrorCode::BadRequest, "malformed hello"),
+            )
+            .await;
+            return Ok(());
+        }
+    };
+
+    let (client, protocol, token) = match hello {
+        ClientMessage::Hello {
+            client,
+            protocol,
+            token,
+        } => (client, protocol, token),
+        ClientMessage::Command { .. } => {
+            let _ = send_out(
+                &mut sink,
+                error_message(None, ErrorCode::Protocol, "expected hello before commands"),
+            )
+            .await;
+            return Ok(());
+        }
+    };
+
+    if protocol != PROTOCOL_VERSION {
+        let _ = send_out(
+            &mut sink,
+            error_message(
+                None,
+                ErrorCode::Protocol,
+                format!(
+                    "protocol mismatch: this daemon speaks {PROTOCOL_VERSION}, the client sent {protocol}"
+                ),
+            ),
+        )
+        .await;
+        return Ok(());
+    }
+    if token.as_str() != shared.token.as_str() {
+        let _ = send_out(
+            &mut sink,
+            error_message(None, ErrorCode::Unauthorized, "the token was not accepted"),
+        )
+        .await;
+        return Ok(());
+    }
+
+    let state = shared.session.lock().await.snapshot();
+    info!(?peer, client = %client, "wire opened");
+    send_out(
+        &mut sink,
+        Outbound::Text(
+            serde_json::to_string(&ServerMessage::Hello {
+                protocol: PROTOCOL_VERSION,
+                session: shared.session_id.as_ref().clone(),
+                engine: state.audio.clone(),
+            })
+            .expect("hello serializes"),
+        ),
+    )
+    .await?;
+    let revision = state.revision;
+    send_out(
+        &mut sink,
+        Outbound::Text(
+            serde_json::to_string(&ServerMessage::Snapshot { revision, state })
+                .expect("snapshot serializes"),
+        ),
+    )
+    .await?;
+
+    let (direct_tx, mut direct_rx) = mpsc::unbounded_channel::<Outbound>();
+    let mut events = shared.out.subscribe();
+    let writer = tokio::spawn(async move {
+        loop {
+            tokio::select! {
+                direct = direct_rx.recv() => {
+                    match direct {
+                        Some(outbound) => {
+                            if send_out(&mut sink, outbound).await.is_err() {
+                                break;
+                            }
+                        }
+                        None => break,
+                    }
+                }
+                broadcast = events.recv() => {
+                    match broadcast {
+                        Ok(outbound) => {
+                            if send_out(&mut sink, outbound).await.is_err() {
+                                break;
+                            }
+                        }
+                        Err(broadcast::error::RecvError::Lagged(_)) => continue,
+                        Err(broadcast::error::RecvError::Closed) => break,
+                    }
+                }
+            }
+        }
+    });
+
+    while let Some(frame) = source.next().await {
+        let message = match frame {
+            Ok(Message::Text(text)) => text,
+            Ok(Message::Close(_)) => break,
+            Ok(_) => continue,
+            Err(error) => {
+                warn!("wire read error: {error}");
+                break;
+            }
+        };
+
+        let message: ClientMessage = match serde_json::from_str(message.as_str()) {
+            Ok(message) => message,
+            Err(_) => {
+                let _ = direct_tx.send(error_message(
+                    None,
+                    ErrorCode::BadRequest,
+                    "malformed message",
+                ));
+                continue;
+            }
+        };
+
+        match message {
+            ClientMessage::Hello { .. } => {
+                let _ = direct_tx.send(error_message(None, ErrorCode::Protocol, "already greeted"));
+            }
+            ClientMessage::Command { id, command } => {
+                let outcome = {
+                    let mut session = shared.session.lock().await;
+                    session.apply(command, Actor::Hand)
+                };
+                match outcome {
+                    Ok((result, emitted)) => {
+                        let revision = match &result {
+                            CommandResult::Snapshot { state } | CommandResult::State { state } => {
+                                state.revision
+                            }
+                        };
+                        let _ = direct_tx.send(Outbound::Text(
+                            serde_json::to_string(&ServerMessage::Result { id, result })
+                                .expect("result serializes"),
+                        ));
+                        for event in emitted {
+                            let message = ServerMessage::Event { revision, event };
+                            let _ = shared
+                                .out
+                                .send(Outbound::Text(serde_json::to_string(&message).unwrap()));
+                        }
+                    }
+                    Err(error) => {
+                        let _ = direct_tx.send(error_message(
+                            Some(id),
+                            error_code(&error),
+                            error.to_string(),
+                        ));
+                    }
+                }
+            }
+        }
+    }
+
+    drop(direct_tx);
+    let _ = writer.await;
+    info!(?peer, "wire closed");
+    Ok(())
+}
+
+#[allow(clippy::result_large_err)]
+fn handshake(request: &Request, response: Response) -> Result<Response, ErrorResponse> {
+    if let Some(origin) = request
+        .headers()
+        .get("origin")
+        .and_then(|value| value.to_str().ok())
+        && !origin_allowed(origin)
+    {
+        return Err(ErrorResponse::new(Some(
+            "origin is not allowed to talk to ZHILA".into(),
+        )));
+    }
+    Ok(response)
+}
+
+fn origin_allowed(origin: &str) -> bool {
+    if origin == "null" {
+        return true;
+    }
+    let Some((scheme, rest)) = origin.split_once("://") else {
+        return false;
+    };
+    if !matches!(scheme, "http" | "https" | "tauri") {
+        return false;
+    }
+    let authority = rest.split('/').next().unwrap_or_default();
+    let host = authority.split(':').next().unwrap_or_default();
+    matches!(host, "localhost" | "127.0.0.1" | "tauri.localhost")
+}
+
+fn error_code(error: &CoreError) -> ErrorCode {
+    match error {
+        CoreError::InvalidTempo | CoreError::InvalidVelocity | CoreError::InvalidGain => {
+            ErrorCode::BadRequest
+        }
+    }
+}
+
+fn error_message(id: Option<u64>, code: ErrorCode, message: impl Into<String>) -> Outbound {
+    Outbound::Text(
+        serde_json::to_string(&ServerMessage::Error {
+            id,
+            code,
+            message: message.into(),
+        })
+        .expect("error serializes"),
+    )
+}
+
+async fn send_out(sink: &mut WsSink, outbound: Outbound) -> anyhow::Result<()> {
+    let message = match outbound {
+        Outbound::Text(text) => Message::Text(text.into()),
+        Outbound::Binary(bytes) => Message::Binary(bytes.into()),
+    };
+    sink.send(message).await.context("websocket send failed")?;
+    Ok(())
+}
+
+fn now_ms() -> u64 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map(|duration| duration.as_millis() as u64)
+        .unwrap_or_default()
+}