//! QQQ (Quick Query Quorum) is deliberately a single-file, in-memory voice-chat server.
//!
//! The global map lock is held only long enough to locate a session; code then acquires that
//! session's lock and never performs provider I/O while holding it. History is append-only: an
//! assistant tool call and its matching tool result are admitted as one pair. Pending inputs are
//! drained in arrival order into one serial model loop, while delivery batches advance strictly
//! one active item at a time. Participant tokens are credentials, never public view data.
// Configuration, errors, and transport DTOs.
use axum::{
Json, Router,
body::to_bytes,
extract::{
DefaultBodyLimit, FromRequest, Multipart, Path, Request, State,
ws::{Message, WebSocket, WebSocketUpgrade},
},
http::{HeaderMap, StatusCode, header},
response::{Html, IntoResponse, Response},
routing::{get, post},
};
use base64::{Engine, engine::general_purpose::STANDARD};
use futures_util::StreamExt;
use reqwest::header::{ACCEPT, AUTHORIZATION, CONTENT_LENGTH, CONTENT_TYPE};
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use std::{
collections::{HashMap, VecDeque},
env,
sync::Arc,
time::{Duration, Instant},
};
use tokio::sync::{Mutex, mpsc};
use uuid::Uuid;
#[derive(Clone)]
struct Config {
bind: String,
public_url: Option<String>,
base_url: String,
api_key: String,
chat_model: String,
stt_model: String,
tts_model: String,
tts_voice: String,
tts_format: String,
referer: Option<String>,
title: Option<String>,
max_sessions: usize,
max_participants: usize,
ttl: Duration,
max_history: usize,
max_audio: usize,
batch: Duration,
delivery_timeout: Duration,
claim_mode: ClaimMode,
default_language: String,
}
#[derive(Clone, Copy, PartialEq)]
enum ClaimMode {
Replace,
Queue,
}
impl Config {
fn load() -> Result<Self, String> {
let required = |k: &str| env::var(k).map_err(|_| format!("missing {k}"));
let value = |k: &str, d: &str| env::var(k).unwrap_or_else(|_| d.into());
let number = |k: &str, d: usize| {
value(k, &d.to_string())
.parse()
.map_err(|_| format!("invalid {k}"))
};
Ok(Self {
bind: value("QQQ_BIND", "127.0.0.1:8787"),
public_url: env::var("QQQ_PUBLIC_URL").ok().filter(|v| !v.is_empty()),
base_url: value("OPENAI_BASE_URL", "https://openrouter.ai/api/v1")
.trim_end_matches('/')
.into(),
api_key: required("OPENAI_API_KEY")?,
chat_model: required("CHAT_MODEL")?,
stt_model: required("STT_MODEL")?,
tts_model: required("TTS_MODEL")?,
tts_voice: required("TTS_VOICE")?,
tts_format: value("TTS_FORMAT", "mp3"),
referer: env::var("OPENAI_HTTP_REFERER")
.ok()
.filter(|s| !s.is_empty()),
title: env::var("OPENAI_APP_TITLE").ok().filter(|s| !s.is_empty()),
max_sessions: number("MAX_SESSIONS", 100)?,
max_participants: number("MAX_PARTICIPANTS_PER_SESSION", 8)?,
ttl: Duration::from_secs(number("SESSION_TTL_SECONDS", 14400)? as u64),
max_history: number("MAX_HISTORY_MESSAGES", 512)?,
max_audio: number("MAX_AUDIO_BYTES", 10_000_000)?,
batch: Duration::from_millis(number("INPUT_BATCH_MS", 350)? as u64),
delivery_timeout: Duration::from_secs(number("DELIVERY_TIMEOUT_SECONDS", 45)? as u64),
claim_mode: match value("CLAIM_MODE", "replace").as_str() {
"replace" => ClaimMode::Replace,
"queue" => ClaimMode::Queue,
_ => return Err("invalid CLAIM_MODE".into()),
},
default_language: value("DEFAULT_LANGUAGE", "en"),
})
}
}
fn startup_config_error(error: &str) -> String {
format!("configuration error: {error}; set it in .env or export it")
}
#[derive(Debug)]
struct ApiError(&'static str, &'static str, StatusCode);
impl IntoResponse for ApiError {
fn into_response(self) -> Response {
(
self.2,
Json(json!({"error":{"code":self.0,"message":self.1}})),
)
.into_response()
}
}
type ApiResult<T> = Result<T, ApiError>;
fn bad() -> ApiError {
ApiError(
"invalid_request",
"The request could not be completed.",
StatusCode::BAD_REQUEST,
)
}
fn provider() -> ApiError {
ApiError(
"provider_failure",
"The request could not be completed.",
StatusCode::BAD_GATEWAY,
)
}
fn unauthorized() -> ApiError {
ApiError(
"unauthorized",
"The request could not be completed.",
StatusCode::UNAUTHORIZED,
)
}
fn session_not_found() -> ApiError {
ApiError(
"session_not_found",
"The request could not be completed.",
StatusCode::NOT_FOUND,
)
}
fn chat_failure() -> ApiError {
eprintln!("chat failed");
provider()
}
// provider/history DTOs
#[derive(Clone, Serialize, Deserialize, Debug)]
struct ChatMessage {
role: String,
#[serde(skip_serializing_if = "Option::is_none")]
name: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
content: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
tool_calls: Option<Vec<ToolCall>>,
#[serde(skip_serializing_if = "Option::is_none")]
tool_call_id: Option<String>,
}
#[derive(Clone, Serialize, Deserialize, Debug)]
struct ToolCall {
id: String,
#[serde(rename = "type")]
kind: String,
function: ToolFunction,
}
#[derive(Clone, Serialize, Deserialize, Debug)]
struct ToolFunction {
name: String,
arguments: String,
}
#[derive(Deserialize, Debug)]
struct ProviderReply {
id: Option<String>,
model: Option<String>,
choices: Vec<Choice>,
usage: Option<ProviderUsage>,
}
#[derive(Deserialize, Debug)]
struct ProviderUsage {
prompt_tokens: u64,
completion_tokens: u64,
total_tokens: u64,
cost: Option<f64>,
prompt_tokens_details: Option<PromptTokenDetails>,
}
#[derive(Deserialize, Debug)]
struct PromptTokenDetails {
#[serde(default)]
cached_tokens: u64,
#[serde(default)]
cache_write_tokens: u64,
}
#[derive(Deserialize, Debug)]
struct Choice {
finish_reason: Option<String>,
message: ProviderMessage,
}
#[derive(Deserialize, Debug)]
struct ProviderMessage {
role: String,
tool_calls: Option<Vec<ToolCall>>,
content: Option<String>,
}
#[derive(Deserialize)]
struct SendMessagesArgs {
dispatches: Vec<Dispatch>,
next_speaker: Option<String>,
}
#[derive(Deserialize, Clone)]
struct Dispatch {
reply_to: String,
text: String,
}
#[derive(Serialize, Deserialize)]
struct TtsRequest {
model: String,
voice: String,
input: String,
response_format: String,
}
// domain state
struct AppState {
config: Config,
http: reqwest::Client,
sessions: Mutex<HashMap<String, Arc<Mutex<Session>>>>,
}
struct Session {
id: String,
#[allow(dead_code)]
created_at: Instant,
last_activity_at: Instant,
language: String,
system_prompt: Option<String>,
status: SessionStatus,
participants: HashMap<String, Participant>,
participant_order: Vec<String>,
messages: Vec<ChatMessage>,
onboarding_messages: Vec<ChatMessage>,
pending_messages: VecDeque<ChatMessage>,
model_running: bool,
debounce_scheduled: bool,
deliveries: VecDeque<Delivery>,
active_delivery: Option<Delivery>,
batches: HashMap<String, Batch>,
floor_holder: Option<String>,
active_recording_participant: Option<String>,
queued_claims: VecDeque<String>,
sockets: HashMap<String, SocketConnection>,
onboarding_audio: HashMap<String, OnboardingAudio>,
onboarding_revisions: HashMap<String, u64>,
}
#[derive(Debug, PartialEq)]
enum SessionStatus {
Onboarding,
Active,
Closed,
}
struct Participant {
id: String,
token: String,
api_name: String,
display_name: Option<String>,
browser_language: String,
onboarding: Onboarding,
connected: bool,
joined_at: Instant,
creator: bool,
}
#[derive(PartialEq)]
enum Onboarding {
Required,
Complete,
}
struct Delivery {
id: String,
participant_id: String,
participant_api_name: String,
text: String,
audio: Option<Vec<u8>>,
mime: String,
batch_id: String,
#[allow(dead_code)]
created_at: Instant,
}
struct Batch {
remaining: usize,
next_speaker: Option<String>,
}
struct OnboardingAudio {
id: String,
bytes: Option<Vec<u8>>,
mime: String,
text: String,
}
struct SocketConnection {
id: String,
tx: mpsc::Sender<ServerEvent>,
}
#[derive(Clone, Serialize)]
struct ServerEvent {
#[serde(rename = "type")]
kind: String,
#[serde(flatten)]
data: Value,
}
impl ServerEvent {
fn new(kind: &str, data: Value) -> Self {
Self {
kind: kind.into(),
data,
}
}
}
fn emit(s: &Session, participant: Option<&str>, event: ServerEvent) {
for (id, connection) in &s.sockets {
if participant.map(|x| x == id).unwrap_or(true) {
let _ = connection.tx.try_send(event.clone());
}
}
}
fn owns_socket(s: &Session, participant_id: &str, connection_id: &str) -> bool {
s.sockets
.get(participant_id)
.is_some_and(|connection| connection.id == connection_id)
}
// validation/state transition helpers
fn now_session(id: String) -> Session {
Session {
id,
created_at: Instant::now(),
last_activity_at: Instant::now(),
language: String::new(),
system_prompt: None,
status: SessionStatus::Onboarding,
participants: HashMap::new(),
participant_order: vec![],
messages: vec![],
onboarding_messages: vec![],
pending_messages: VecDeque::new(),
model_running: false,
debounce_scheduled: false,
deliveries: VecDeque::new(),
active_delivery: None,
batches: HashMap::new(),
floor_holder: None,
active_recording_participant: None,
queued_claims: VecDeque::new(),
sockets: HashMap::new(),
onboarding_audio: HashMap::new(),
onboarding_revisions: HashMap::new(),
}
}
fn api_name(s: &Session) -> String {
format!("user_{}", s.participant_order.len() + 1)
}
fn add_participant(s: &mut Session, language: String, creator: bool) -> Participant {
let p = Participant {
id: Uuid::new_v4().to_string(),
token: Uuid::new_v4().to_string(),
api_name: api_name(s),
display_name: None,
browser_language: language,
onboarding: Onboarding::Required,
connected: true,
joined_at: Instant::now(),
creator,
};
s.participant_order.push(p.id.clone());
s.onboarding_revisions.insert(p.id.clone(), 1);
let returned = p.clone_for_return();
s.participants.insert(p.id.clone(), p);
returned
}
impl Participant {
fn clone_for_return(&self) -> Self {
Self {
id: self.id.clone(),
token: self.token.clone(),
api_name: self.api_name.clone(),
display_name: self.display_name.clone(),
browser_language: self.browser_language.clone(),
onboarding: if self.onboarding == Onboarding::Required {
Onboarding::Required
} else {
Onboarding::Complete
},
connected: self.connected,
joined_at: self.joined_at,
creator: self.creator,
}
}
}
fn participant_by_token<'a>(s: &'a Session, token: &str) -> Option<&'a Participant> {
s.participants.values().find(|p| p.token == token)
}
fn require_auth<'a>(s: &'a Session, token: &str) -> ApiResult<&'a Participant> {
participant_by_token(s, token).ok_or_else(unauthorized)
}
async fn lookup_session(st: &AppState, id: &str) -> ApiResult<Arc<Mutex<Session>>> {
st.sessions
.lock()
.await
.get(id)
.cloned()
.ok_or_else(session_not_found)
}
fn history_room(s: &mut Session, max: usize, count: usize) -> ApiResult<()> {
if s.messages.len() + count > max {
s.status = SessionStatus::Closed;
emit(
s,
None,
ServerEvent::new("error", json!({"code":"history_limit"})),
);
return Err(ApiError(
"history_limit",
"The request could not be completed.",
StatusCode::CONFLICT,
));
}
Ok(())
}
fn user_message(p: &Participant, text: String) -> ChatMessage {
ChatMessage {
role: "user".into(),
name: Some(p.api_name.clone()),
content: Some(format!(
"{}: {}",
p.display_name.as_deref().unwrap_or("Participant"),
text.trim()
)),
tool_calls: None,
tool_call_id: None,
}
}
fn event_message(text: String) -> ChatMessage {
ChatMessage {
role: "user".into(),
name: Some("qqq_event".into()),
content: Some(text),
tool_calls: None,
tool_call_id: None,
}
}
fn mime_format(mime: &str) -> Option<&'static str> {
match mime.split(';').next().unwrap_or("").trim() {
"audio/webm" => Some("webm"),
"audio/ogg" => Some("ogg"),
"audio/mp4" => Some("m4a"),
"audio/mpeg" => Some("mp3"),
"audio/wav" => Some("wav"),
_ => None,
}
}
/// Reads the only accepted browser file field. The router disables Axum's static default limit,
/// so this streaming check remains authoritative for the configured per-instance audio limit.
async fn browser_audio(request: Request, limit: usize) -> ApiResult<(Vec<u8>, String)> {
let mut multipart = Multipart::from_request(request, &())
.await
.map_err(|_| bad())?;
let mut audio = None;
while let Some(mut field) = multipart.next_field().await.map_err(|_| bad())? {
if field.name() != Some("audio") || audio.is_some() {
continue;
}
let mime = field
.content_type()
.and_then(mime_format)
.ok_or_else(bad)?
.to_string();
let mut bytes = Vec::new();
while let Some(chunk) = field.next().await {
let chunk = chunk.map_err(|_| bad())?;
if bytes.len().saturating_add(chunk.len()) > limit {
return Err(ApiError(
"audio_too_large",
"The request could not be completed.",
StatusCode::PAYLOAD_TOO_LARGE,
));
}
bytes.extend_from_slice(&chunk);
}
if bytes.is_empty() {
return Err(bad());
}
audio = Some((bytes, mime));
}
audio.ok_or_else(bad)
}
fn validate_tool(
s: &Session,
args: SendMessagesArgs,
) -> Result<(Vec<Dispatch>, Option<String>), ()> {
if args.dispatches.is_empty() || args.dispatches.len() > 16 {
return Err(());
}
let mut out = vec![];
for d in args.dispatches {
let text = d.text.trim().to_string();
if text.is_empty() || text.chars().count() > 2000 {
return Err(());
}
if d.reply_to == "all" {
for id in &s.participant_order {
let p = &s.participants[id];
if p.connected && p.onboarding == Onboarding::Complete {
out.push(Dispatch {
reply_to: p.api_name.clone(),
text: text.clone(),
});
}
}
} else if s.participants.values().any(|p| {
p.connected && p.onboarding == Onboarding::Complete && p.api_name == d.reply_to
}) {
out.push(Dispatch {
reply_to: d.reply_to,
text,
});
} else {
return Err(());
}
}
if out.is_empty() {
return Err(());
}
if let Some(n) = &args.next_speaker
&& !s
.participants
.values()
.any(|p| p.connected && p.onboarding == Onboarding::Complete && p.api_name == *n)
{
return Err(());
}
Ok((out, args.next_speaker))
}
fn parse_tool(reply: ProviderReply) -> Result<(ToolCall, SendMessagesArgs), ()> {
let choice = reply.choices.into_iter().next().ok_or(())?;
if choice.finish_reason.as_deref() != Some("tool_calls")
|| choice.message.role != "assistant"
|| choice.message.content.is_some()
{
return Err(());
}
let mut calls = choice.message.tool_calls.ok_or(())?;
if calls.len() != 1 {
return Err(());
}
let call = calls.pop().ok_or(())?;
if call.function.name != "send_messages" || call.id.is_empty() {
return Err(());
}
let args = serde_json::from_str(&call.function.arguments).map_err(|_| ())?;
Ok((call, args))
}
fn language_template(language: &str, creator: bool) -> &'static str {
if language.to_lowercase().starts_with("ru") {
if creator {
"Как вас называть и о чём будет этот разговор?"
} else {
"Как вас называть?"
}
} else if creator {
"How can I help?"
} else {
"What should I call you?"
}
}
struct DisconnectOutcome {
active_delivery: Option<String>,
queued_event: bool,
}
fn recording_allowed(s: &Session, participant_id: &str, voice: bool) -> bool {
s.floor_holder.as_deref() == Some(participant_id)
&& (!voice || s.active_recording_participant.as_deref() == Some(participant_id))
}
fn active_delivery_target_connected(s: &Session, delivery_id: &str) -> bool {
s.active_delivery
.as_ref()
.filter(|d| d.id == delivery_id)
.and_then(|d| s.participants.get(&d.participant_id))
.is_some_and(|p| p.connected)
}
/// The public view is intentionally assembled from authority identifiers only; credentials,
/// history, provider payloads, and queued audio never cross this boundary.
fn public_view(s: &Session) -> Value {
let participants = s
.participant_order
.iter()
.filter_map(|id| s.participants.get(id))
.map(|p| {
json!({"api_name":p.api_name,"display_name":p.display_name,"connected":p.connected,"has_floor":s.floor_holder.as_deref()==Some(&p.id),"recording":s.active_recording_participant.as_deref()==Some(&p.id)})
})
.collect::<Vec<_>>();
json!({"session_id":s.id,"status":if s.status==SessionStatus::Active{"active"}else if s.status==SessionStatus::Closed{"closed"}else{"onboarding"},"language":s.language,"participants":participants,"floor_holder":s.floor_holder.as_ref().and_then(|id|s.participants.get(id)).map(|p|p.api_name.clone()),"ai_state":if s.model_running{"thinking"}else{"idle"},"active_delivery":s.active_delivery.as_ref().map(|d|json!({"target":d.participant_api_name}))})
}
fn disconnect_participant(
s: &mut Session,
participant_id: &str,
max_history: usize,
) -> DisconnectOutcome {
let Some(p) = s.participants.get(participant_id) else {
return DisconnectOutcome {
active_delivery: None,
queued_event: false,
};
};
let was_connected = p.connected;
let already_disconnected = !was_connected && !s.sockets.contains_key(participant_id);
let active_delivery = s
.active_delivery
.as_ref()
.filter(|d| d.participant_id == participant_id)
.map(|d| d.id.clone());
if already_disconnected {
return DisconnectOutcome {
active_delivery,
queued_event: false,
};
}
let api = p.api_name.clone();
let name = p
.display_name
.clone()
.unwrap_or_else(|| "Participant".into());
if let Some(p) = s.participants.get_mut(participant_id) {
p.connected = false;
}
s.sockets.remove(participant_id);
s.queued_claims.retain(|id| id != participant_id);
if s.floor_holder.as_deref() == Some(participant_id) {
s.floor_holder = None;
}
if s.active_recording_participant.as_deref() == Some(participant_id) {
s.active_recording_participant = None;
}
let queued_event =
was_connected && s.status == SessionStatus::Active && s.messages.len() < max_history;
if queued_event {
s.pending_messages.push_back(event_message(format!(
"Participant left: {} ({}).",
api, name
)));
}
if was_connected {
emit(
s,
None,
ServerEvent::new("participant.left", json!({"api_name":api})),
);
}
s.last_activity_at = Instant::now();
DisconnectOutcome {
active_delivery,
queued_event,
}
}
// provider functions chat/transcribe/speak
fn request_headers(c: &Config, request: reqwest::RequestBuilder) -> reqwest::RequestBuilder {
let mut r = request
.header(AUTHORIZATION, format!("Bearer {}", c.api_key))
.header(CONTENT_TYPE, "application/json");
if let Some(v) = &c.referer {
r = r.header("HTTP-Referer", v);
}
if let Some(v) = &c.title {
r = r.header("X-OpenRouter-Title", v);
}
r
}
fn tts_request(
client: &reqwest::Client,
config: &Config,
input: &str,
) -> Result<reqwest::RequestBuilder, serde_json::Error> {
let payload = serde_json::to_vec(&TtsRequest {
model: config.tts_model.clone(),
voice: config.tts_voice.clone(),
input: input.into(),
response_format: config.tts_format.clone(),
})?;
let content_length = payload.len().to_string();
Ok(request_headers(
config,
client
.post(format!("{}/audio/speech", config.base_url))
.header(ACCEPT, "audio/*")
.header(CONTENT_LENGTH, content_length)
.body(payload),
))
}
fn tool_schema() -> Value {
json!({"type":"function","function":{"name":"send_messages","description":"Queues concise AI messages for one or more current participants and optionally gives one participant the next speaking floor.","parameters":{"type":"object","additionalProperties":false,"properties":{"dispatches":{"type":"array","minItems":1,"maxItems":16,"items":{"type":"object","additionalProperties":false,"properties":{"reply_to":{"type":"string","description":"A current stable participant API name such as user_1, or the literal all."},"text":{"type":"string","minLength":1,"maxLength":2000,"description":"The concise message to synthesize and deliver."}},"required":["reply_to","text"]}},"next_speaker":{"type":["string","null"],"description":"The current participant API name expected to speak after every queued dispatch finishes, or null."}},"required":["dispatches","next_speaker"]}}})
}
async fn chat(
state: &AppState,
messages: Vec<ChatMessage>,
tool: Value,
force_tool: bool,
session_id: &str,
) -> ApiResult<ProviderReply> {
eprintln!("chat started messages={}", messages.len());
let tool_choice = if force_tool {
json!({"type":"function","function":{"name":tool["function"]["name"]}})
} else {
json!("auto")
};
let mut messages = serde_json::to_value(messages).map_err(|_| provider())?;
if let Some(messages) = messages.as_array_mut()
&& let Some(message) = messages.iter_mut().find(|message| {
message.get("role").and_then(Value::as_str) == Some("system")
&& message.get("content").is_some_and(Value::is_string)
})
&& let Some(content) = message.get_mut("content")
&& let Some(text) = content.as_str().map(str::to_string)
{
*content = json!([{"type":"text","text":text,"cache_control":{"type":"ephemeral"}}]);
}
let body = json!({"model":state.config.chat_model,"session_id":session_id,"messages":messages,"tools":[tool],"tool_choice":tool_choice,"parallel_tool_calls":false,"stream":false});
let res = match request_headers(
&state.config,
state
.http
.post(format!("{}/chat/completions", state.config.base_url))
.json(&body),
)
.send()
.await
{
Ok(res) => res,
Err(error) => {
eprintln!(
"chat failed stage=transport category={}",
transport_error_category(&error)
);
return Err(provider());
}
};
if !res.status().is_success() {
eprintln!("chat failed stage=http status={}", res.status());
return Err(provider());
}
match res.json::<ProviderReply>().await {
Ok(parsed) => {
if let Some(usage) = &parsed.usage {
let details = usage.prompt_tokens_details.as_ref();
eprintln!(
"chat completed id={} model={} prompt_tokens={} completion_tokens={} total_tokens={} cache_read_tokens={} cache_write_tokens={} cost={}",
parsed.id.as_deref().unwrap_or("unknown"),
parsed.model.as_deref().unwrap_or(&state.config.chat_model),
usage.prompt_tokens,
usage.completion_tokens,
usage.total_tokens,
details.map_or(0, |d| d.cached_tokens),
details.map_or(0, |d| d.cache_write_tokens),
usage
.cost
.map_or_else(|| "unknown".into(), |cost| cost.to_string())
);
} else {
eprintln!("chat completed usage=unavailable");
}
Ok(parsed)
}
Err(error) => {
eprintln!("chat failed stage=decode error={error}");
Err(provider())
}
}
}
async fn transcribe(
state: &AppState,
bytes: Vec<u8>,
format: &str,
language: &str,
) -> ApiResult<String> {
eprintln!("STT started bytes={}", bytes.len());
for attempt in 0..2 {
let body = json!({"model":state.config.stt_model,"input_audio":{"data":STANDARD.encode(&bytes),"format":format},"language":language,"temperature":0});
let result = request_headers(
&state.config,
state
.http
.post(format!("{}/audio/transcriptions", state.config.base_url))
.json(&body),
)
.send()
.await;
match result {
Ok(r) if r.status().is_success() => {
let v: Value = match r.json().await {
Ok(v) => v,
Err(_) => {
eprintln!("STT failed");
return Err(provider());
}
};
let Some(text) = v
.get("text")
.and_then(Value::as_str)
.map(str::trim)
.filter(|s| !s.is_empty())
else {
eprintln!("STT failed");
return Err(provider());
};
eprintln!("STT completed attempt={}", attempt + 1);
return Ok(text.into());
}
Ok(r) if !retry_status(r.status()) => {
eprintln!("STT failed");
return Err(provider());
}
_ if attempt == 1 => {
eprintln!("STT failed");
return Err(provider());
}
_ => tokio::time::sleep(Duration::from_millis(150)).await,
}
}
unreachable!()
}
fn retry_status(s: StatusCode) -> bool {
matches!(s.as_u16(), 408 | 429 | 500 | 502 | 503 | 504)
}
fn transport_error_category(error: &reqwest::Error) -> &'static str {
if error.is_timeout() {
"timeout"
} else if error.is_connect() {
"connect"
} else if error.is_body() {
"body"
} else if error.is_decode() {
"decode"
} else if error.is_request() {
"request"
} else {
"other"
}
}
async fn speak(state: &AppState, text: &str) -> ApiResult<(Vec<u8>, String)> {
eprintln!("TTS started characters={}", text.chars().count());
for attempt in 0..2 {
let request = match tts_request(&state.http, &state.config, text) {
Ok(request) => request,
Err(_) => {
eprintln!("TTS failed attempt={} serialization", attempt + 1);
return Err(provider());
}
};
let result = request.send().await;
match result {
Ok(r) if r.status().is_success() => {
let mime = r
.headers()
.get(header::CONTENT_TYPE)
.and_then(|x| x.to_str().ok())
.unwrap_or("audio/mpeg")
.to_string();
let b = match r.bytes().await {
Ok(bytes) => bytes.to_vec(),
Err(error) => {
eprintln!(
"TTS failed attempt={} transport={}",
attempt + 1,
transport_error_category(&error)
);
return Err(provider());
}
};
if !b.is_empty() {
eprintln!("TTS completed attempt={} bytes={}", attempt + 1, b.len());
return Ok((b, mime));
}
eprintln!("TTS failed attempt={} empty_response", attempt + 1);
return Err(provider());
}
Ok(r) if !retry_status(r.status()) => {
eprintln!("TTS failed attempt={} status={}", attempt + 1, r.status());
return Err(provider());
}
Ok(r) if attempt == 1 => {
eprintln!("TTS failed attempt={} status={}", attempt + 1, r.status());
return Err(provider());
}
Ok(r) => {
eprintln!("TTS retrying attempt={} status={}", attempt + 1, r.status());
tokio::time::sleep(Duration::from_millis(150)).await;
}
Err(error) if attempt == 1 => {
eprintln!(
"TTS failed attempt={} transport={}",
attempt + 1,
transport_error_category(&error)
);
return Err(provider());
}
Err(error) => {
eprintln!(
"TTS retrying attempt={} transport={}",
attempt + 1,
transport_error_category(&error)
);
tokio::time::sleep(Duration::from_millis(150)).await;
}
}
}
unreachable!()
}
// orchestration loops
async fn schedule_loop(state: Arc<AppState>, sid: String) {
let session = { state.sessions.lock().await.get(&sid).cloned() };
let Some(session) = session else { return };
let wait = {
let mut s = session.lock().await;
if s.debounce_scheduled || s.model_running {
return;
}
s.debounce_scheduled = true;
state.config.batch
};
tokio::spawn(async move {
tokio::time::sleep(wait).await;
run_model(state, sid).await;
});
}
async fn run_model(state: Arc<AppState>, sid: String) {
let session = { state.sessions.lock().await.get(&sid).cloned() };
let Some(session) = session else { return };
let payload = {
let mut s = session.lock().await;
s.debounce_scheduled = false;
if s.model_running || s.status != SessionStatus::Active {
return;
}
let mut pending = Vec::new();
while let Some(message) = s.pending_messages.pop_front() {
pending.push(message);
}
if pending.is_empty() {
return;
}
for message in pending {
if history_room(&mut s, state.config.max_history, 1).is_err() {
return;
}
s.messages.push(message);
}
s.model_running = true;
emit(&s, None, ServerEvent::new("ai.thinking", json!({})));
s.messages.clone()
};
let result = chat(&state, payload, tool_schema(), true, &sid)
.await
.and_then(|reply| parse_tool(reply).map_err(|_| chat_failure()));
let mut start = false;
{
let mut s = session.lock().await;
s.model_running = false;
match result.and_then(|(call, args)| {
validate_tool(&s, args)
.map(|x| (call, x))
.map_err(|_| chat_failure())
}) {
Ok((call, (dispatches, next))) => {
if history_room(&mut s, state.config.max_history, 2).is_ok() {
let call_id = call.id.clone();
s.messages.push(ChatMessage {
role: "assistant".into(),
name: None,
content: None,
tool_calls: Some(vec![call]),
tool_call_id: None,
});
let batch = Uuid::new_v4().to_string();
let mut count = 0;
for d in dispatches {
let p = s
.participants
.values()
.find(|p| p.api_name == d.reply_to)
.unwrap();
let participant_id = p.id.clone();
let participant_api_name = p.api_name.clone();
s.deliveries.push_back(Delivery {
id: Uuid::new_v4().to_string(),
participant_id,
participant_api_name,
text: d.text,
audio: None,
mime: "audio/mpeg".into(),
batch_id: batch.clone(),
created_at: Instant::now(),
});
count += 1;
}
s.batches.insert(
batch,
Batch {
remaining: count,
next_speaker: next,
},
);
s.messages.push(ChatMessage {
role: "tool".into(),
name: None,
content: Some(json!({"status":"success","queued":count}).to_string()),
tool_calls: None,
tool_call_id: Some(call_id),
});
start = s.active_delivery.is_none();
}
}
Err(_) => emit(
&s,
None,
ServerEvent::new("error", json!({"code":"provider_failure"})),
),
}
if !s.pending_messages.is_empty() {
s.debounce_scheduled = true;
let st = state.clone();
let id = sid.clone();
let delay = state.config.batch;
schedule_model_after(st, id, delay);
}
}
if start {
trigger_delivery(state, sid);
}
}
async fn start_delivery(state: Arc<AppState>, sid: String) {
let session = { state.sessions.lock().await.get(&sid).cloned() };
let Some(session) = session else { return };
let delivery = {
let mut s = session.lock().await;
if s.active_delivery.is_some() {
return;
}
let Some(d) = s.deliveries.pop_front() else {
return;
};
let delivery_id = d.id.clone();
emit(
&s,
None,
ServerEvent::new(
"delivery.preparing",
json!({"target":d.participant_api_name}),
),
);
s.active_delivery = Some(d);
s.last_activity_at = Instant::now();
delivery_id
};
let state2 = state.clone();
tokio::spawn(async move {
let (text, target) = {
let s = session.lock().await;
let Some(d) = s.active_delivery.as_ref().filter(|d| d.id == delivery) else {
return;
};
let target = d.participant_id.clone();
if !active_delivery_target_connected(&s, &delivery) {
drop(s);
complete_delivery(state2, sid, delivery).await;
return;
}
(d.text.clone(), target)
};
let spoken = speak(&state2, &text).await.ok();
let mut s = session.lock().await;
let has_audio = if let Some(d) = s.active_delivery.as_mut().filter(|d| d.id == delivery) {
if let Some((audio, mime)) = spoken {
d.audio = Some(audio);
d.mime = mime;
}
d.audio.is_some()
} else {
return;
};
emit(
&s,
Some(&target),
ServerEvent::new(
"delivery.started",
json!({"delivery_id":delivery,"text":text,"audio":has_audio}),
),
);
drop(s);
let timeout = state2.config.delivery_timeout;
tokio::time::sleep(timeout).await;
complete_delivery(state2, sid, delivery).await;
});
}
fn trigger_delivery(state: Arc<AppState>, sid: String) {
tokio::spawn(async move {
start_delivery(state, sid).await;
});
}
fn schedule_model_after(state: Arc<AppState>, sid: String, delay: Duration) {
tokio::spawn(async move {
tokio::time::sleep(delay).await;
run_model(state, sid).await;
});
}
async fn complete_delivery(state: Arc<AppState>, sid: String, did: String) {
let session = { state.sessions.lock().await.get(&sid).cloned() };
let Some(session) = session else { return };
let start;
{
let mut s = session.lock().await;
let Some(active) = s.active_delivery.as_ref() else {
return;
};
if active.id != did {
return;
}
let active = s.active_delivery.take().unwrap();
s.last_activity_at = Instant::now();
emit(
&s,
None,
ServerEvent::new("delivery.completed", json!({"delivery_id":did})),
);
if let Some(batch) = s.batches.get_mut(&active.batch_id) {
batch.remaining -= 1;
if batch.remaining == 0 {
let next = batch.next_speaker.clone();
s.batches.remove(&active.batch_id);
s.floor_holder = next.and_then(|n| {
s.participants
.values()
.find(|p| p.connected && p.api_name == n)
.map(|p| p.id.clone())
});
emit(
&s,
None,
ServerEvent::new(
"floor.changed",
json!({"floor_holder":s.floor_holder.as_ref().and_then(|id|s.participants.get(id)).map(|p|p.api_name.clone())}),
),
);
}
}
start = !s.deliveries.is_empty();
}
if start {
trigger_delivery(state, sid);
}
}
// HTTP/WS handlers
#[derive(Deserialize)]
struct LanguageRequest {
browser_language: Option<String>,
}
#[derive(Serialize)]
struct Created {
session_id: String,
participant_token: String,
participant_id: String,
api_name: String,
onboarding_required: bool,
}
async fn page() -> Html<&'static str> {
Html(PAGE)
}
async fn health() -> &'static str {
"ok"
}
fn start_onboarding_prompt(
state: Arc<AppState>,
sid: String,
participant_id: String,
revision: u64,
) {
tokio::spawn(async move {
let session = { state.sessions.lock().await.get(&sid).cloned() };
let Some(session) = session else { return };
let (text, language) = {
let s = session.lock().await;
let Some(p) = s.participants.get(&participant_id) else {
return;
};
let lang = if p.browser_language.is_empty() {
if s.language.is_empty() {
state.config.default_language.clone()
} else {
s.language.clone()
}
} else {
p.browser_language.clone()
};
(language_template(&lang, p.creator).to_string(), lang)
};
let _ = language;
let audio = speak(&state, &text).await.ok();
let mut s = session.lock().await;
if !s
.participants
.get(&participant_id)
.is_some_and(|p| p.onboarding == Onboarding::Required)
|| s.onboarding_revisions.get(&participant_id) != Some(&revision)
{
return;
}
let id = Uuid::new_v4().to_string();
let (bytes, mime) = audio.map_or((None, "audio/mpeg".to_string()), |(b, m)| (Some(b), m));
s.onboarding_audio.insert(
participant_id.clone(),
OnboardingAudio {
id: id.clone(),
bytes,
mime,
text: text.clone(),
},
);
emit(
&s,
Some(&participant_id),
ServerEvent::new(
"onboarding.prompt",
json!({"delivery_id":id,"text":text,"audio":s.onboarding_audio[&participant_id].bytes.is_some()}),
),
);
});
}
async fn exists(State(st): State<Arc<AppState>>, Path(id): Path<String>) -> Json<Value> {
let session = st.sessions.lock().await.get(&id).cloned();
let exists = if let Some(session) = session {
session.lock().await.status == SessionStatus::Active
} else {
false
};
Json(json!({"exists":exists}))
}
async fn create(
State(st): State<Arc<AppState>>,
Json(req): Json<LanguageRequest>,
) -> ApiResult<Json<Created>> {
let mut map = st.sessions.lock().await;
if map.len() >= st.config.max_sessions {
let ttl = st.config.ttl;
map.retain(|id, session| {
let retain = session
.try_lock()
.map(|s| s.last_activity_at.elapsed() <= ttl)
.unwrap_or(true);
if !retain {
eprintln!("session expired id={id}");
}
retain
});
}
if map.len() >= st.config.max_sessions {
return Err(ApiError(
"server_busy",
"The request could not be completed.",
StatusCode::SERVICE_UNAVAILABLE,
));
}
let id = Uuid::new_v4().to_string();
let mut s = now_session(id.clone());
s.onboarding_messages.push(ChatMessage {
role: "system".into(),
name: None,
content: Some("You are a helpful private voice assistant. Continue a normal one-to-one conversation without creating a group session. Call start_group_session only when the user explicitly wants a group conversation and has provided their preferred display name, the language they are using, and a concrete group purpose. If group intent is explicit but any required detail is missing, ask one concise question. Never infer group intent from an ordinary request.".into()),
tool_calls: None,
tool_call_id: None,
});
let p = add_participant(
&mut s,
req.browser_language
.unwrap_or_else(|| st.config.default_language.clone()),
true,
);
map.insert(id.clone(), Arc::new(Mutex::new(s)));
drop(map);
eprintln!("private conversation started id={id}");
start_onboarding_prompt(st.clone(), id.clone(), p.id.clone(), 1);
Ok(Json(Created {
session_id: id,
participant_token: p.token,
participant_id: p.id,
api_name: p.api_name,
onboarding_required: true,
}))
}
async fn join(
State(st): State<Arc<AppState>>,
Path(id): Path<String>,
Json(req): Json<LanguageRequest>,
) -> ApiResult<Json<Created>> {
let session = st.sessions.lock().await.get(&id).cloned().ok_or(ApiError(
"session_not_found",
"The request could not be completed.",
StatusCode::NOT_FOUND,
))?;
let mut s = session.lock().await;
if s.status != SessionStatus::Active {
return Err(ApiError(
"session_not_ready",
"The request could not be completed.",
StatusCode::CONFLICT,
));
}
if s.participants.len() >= st.config.max_participants {
return Err(ApiError(
"session_full",
"The request could not be completed.",
StatusCode::CONFLICT,
));
}
let p = add_participant(
&mut s,
req.browser_language
.unwrap_or_else(|| st.config.default_language.clone()),
false,
);
s.last_activity_at = Instant::now();
drop(s);
eprintln!("participant joined session={id} participant={}", p.id);
start_onboarding_prompt(st.clone(), id.clone(), p.id.clone(), 1);
Ok(Json(Created {
session_id: id,
participant_token: p.token,
participant_id: p.id,
api_name: p.api_name,
onboarding_required: true,
}))
}
async fn bootstrap_creator(
st: Arc<AppState>,
id: String,
token: String,
text: String,
) -> ApiResult<Json<Value>> {
let session = lookup_session(&st, &id).await?;
let user = ChatMessage {
role: "user".into(),
name: None,
content: Some(text),
tool_calls: None,
tool_call_id: None,
};
let (messages, revision) = {
let mut s = session.lock().await;
let participant = require_auth(&s, &token)?;
if !participant.creator
|| participant.onboarding != Onboarding::Required
|| s.status != SessionStatus::Onboarding
{
return Err(bad());
}
let participant_id = participant.id.clone();
let mut messages = s.onboarding_messages.clone();
messages.push(user.clone());
let revision = s
.onboarding_revisions
.entry(participant_id)
.and_modify(|revision| *revision += 1)
.or_insert(1);
(messages, *revision)
};
let schema = json!({"type":"function","function":{"name":"start_group_session","description":"Create a multi-user session only after the user explicitly requests a group conversation and supplies every required field.","parameters":{"type":"object","additionalProperties":false,"properties":{"display_name":{"type":"string","minLength":1,"maxLength":80},"language":{"type":"string","minLength":2,"maxLength":16},"session_purpose":{"type":"string","minLength":1,"maxLength":4000}},"required":["display_name","language","session_purpose"]}}});
let response = chat(&st, messages, schema, false, &id).await?;
let choice = response
.choices
.into_iter()
.next()
.ok_or_else(chat_failure)?;
if let Some(mut calls) = choice.message.tool_calls {
if choice.finish_reason.as_deref() != Some("tool_calls") || calls.len() != 1 {
return Err(chat_failure());
}
let call = calls.pop().ok_or_else(chat_failure)?;
if call.function.name != "start_group_session" {
return Err(chat_failure());
}
let args: Value =
serde_json::from_str(&call.function.arguments).map_err(|_| chat_failure())?;
let display = args
.get("display_name")
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty() && value.chars().count() <= 80)
.ok_or_else(chat_failure)?
.to_string();
let language = args
.get("language")
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| (2..=16).contains(&value.len()))
.ok_or_else(chat_failure)?
.to_string();
let purpose = args
.get("session_purpose")
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty() && value.chars().count() <= 4000)
.ok_or_else(chat_failure)?
.to_string();
let mut s = session.lock().await;
let pid = require_auth(&s, &token)?.id.clone();
if !s.participants[&pid].creator
|| s.participants[&pid].onboarding != Onboarding::Required
|| s.status != SessionStatus::Onboarding
|| s.onboarding_revisions.get(&pid) != Some(&revision)
{
return Err(ApiError(
"stale_onboarding",
"The request could not be completed.",
StatusCode::CONFLICT,
));
}
let participant_api = {
let participant = s.participants.get_mut(&pid).ok_or_else(unauthorized)?;
participant.display_name = Some(display.clone());
participant.onboarding = Onboarding::Complete;
participant.api_name.clone()
};
s.language = language;
let system_prompt = format!(
"You are the AI participant and facilitator in a multi-user voice chat.\n\nSession purpose:\n{}\n\nParticipants have stable API names such as user_1 and user_2. Human display names are repeated inside each user message. You must respond only through send_messages. Never produce ordinary assistant content. Address only current participant API names or \"all\". Keep spoken messages concise. Maintain conversational momentum. Use next_speaker when one participant should answer.",
purpose
);
s.system_prompt = Some(system_prompt.clone());
s.messages = vec![ChatMessage {
role: "system".into(),
name: None,
content: Some(system_prompt),
tool_calls: None,
tool_call_id: None,
}];
s.pending_messages.push_back(event_message(format!(
"Session started with {} ({}). Begin the conversation.",
participant_api, display
)));
s.status = SessionStatus::Active;
s.onboarding_messages.clear();
s.onboarding_audio.remove(&pid);
s.last_activity_at = Instant::now();
emit(
&s,
None,
ServerEvent::new("participant.named", json!({"api_name":participant_api})),
);
eprintln!("session created id={id}");
drop(s);
schedule_loop(st, id).await;
return Ok(Json(json!({"ok":true,"session_created":true})));
}
let answer = choice
.message
.content
.map(|content| content.trim().to_string())
.filter(|content| !content.is_empty())
.ok_or_else(chat_failure)?;
let audio = speak(&st, &answer).await.ok();
let mut s = session.lock().await;
let pid = require_auth(&s, &token)?.id.clone();
if s.status != SessionStatus::Onboarding || s.onboarding_revisions.get(&pid) != Some(&revision)
{
return Err(ApiError(
"stale_onboarding",
"The request could not be completed.",
StatusCode::CONFLICT,
));
}
history_room_for_draft(&s.onboarding_messages, st.config.max_history, 2)?;
s.onboarding_messages.push(user);
s.onboarding_messages.push(ChatMessage {
role: "assistant".into(),
name: None,
content: Some(answer.clone()),
tool_calls: None,
tool_call_id: None,
});
let delivery_id = Uuid::new_v4().to_string();
let (bytes, mime) = audio.map_or((None, "audio/mpeg".to_string()), |(bytes, mime)| {
(Some(bytes), mime)
});
s.onboarding_audio.insert(
pid.clone(),
OnboardingAudio {
id: delivery_id.clone(),
bytes,
mime,
text: answer.clone(),
},
);
emit(
&s,
Some(&pid),
ServerEvent::new(
"onboarding.prompt",
json!({"delivery_id":delivery_id,"text":answer,"audio":s.onboarding_audio[&pid].bytes.is_some()}),
),
);
s.last_activity_at = Instant::now();
Ok(Json(json!({"ok":true,"session_created":false})))
}
fn history_room_for_draft(messages: &[ChatMessage], max: usize, count: usize) -> ApiResult<()> {
if messages.len().saturating_add(count) > max {
return Err(ApiError(
"history_limit",
"The request could not be completed.",
StatusCode::CONFLICT,
));
}
Ok(())
}
fn bearer(h: &HeaderMap) -> Option<&str> {
h.get(header::AUTHORIZATION)?
.to_str()
.ok()?
.strip_prefix("Bearer ")
}
async fn session_view(
State(st): State<Arc<AppState>>,
Path(id): Path<String>,
headers: HeaderMap,
) -> ApiResult<Json<Value>> {
let token = bearer(&headers).ok_or_else(unauthorized)?;
let session = lookup_session(&st, &id).await?;
let s = session.lock().await;
require_auth(&s, token)?;
Ok(Json(public_view(&s)))
}
async fn bootstrap(
State(st): State<Arc<AppState>>,
Path(id): Path<String>,
request: Request,
) -> ApiResult<Json<Value>> {
let token = bearer(request.headers())
.ok_or(ApiError(
"unauthorized",
"The request could not be completed.",
StatusCode::UNAUTHORIZED,
))?
.to_string();
let content_type = request
.headers()
.get(header::CONTENT_TYPE)
.and_then(|v| v.to_str().ok())
.unwrap_or("");
let text = if content_type.starts_with("application/json") {
let bytes = to_bytes(request.into_body(), st.config.max_audio + 1024)
.await
.map_err(|_| bad())?;
serde_json::from_slice::<Value>(&bytes)
.ok()
.and_then(|v| v.get("text").and_then(Value::as_str).map(str::to_string))
.filter(|s| !s.trim().is_empty())
.ok_or_else(bad)?
} else {
let (audio, format) = browser_audio(request, st.config.max_audio).await?;
let session = st.sessions.lock().await.get(&id).cloned().ok_or(ApiError(
"session_not_found",
"The request could not be completed.",
StatusCode::NOT_FOUND,
))?;
let language = {
let s = session.lock().await;
let p = require_auth(&s, &token)?;
if !p.browser_language.is_empty() {
p.browser_language.clone()
} else if !s.language.is_empty() {
s.language.clone()
} else {
st.config.default_language.clone()
}
};
transcribe(&st, audio, &format, &language).await?
};
let session = st.sessions.lock().await.get(&id).cloned().ok_or(ApiError(
"session_not_found",
"The request could not be completed.",
StatusCode::NOT_FOUND,
))?;
let creator = {
let s = session.lock().await;
let p = require_auth(&s, &token)?;
if p.onboarding != Onboarding::Required {
return Err(bad());
}
p.creator
};
if creator {
return bootstrap_creator(st, id, token, text).await;
}
let name_tool = "set_participant_name";
let schema = json!({"type":"function","function":{"name":name_tool,"parameters":{"type":"object","additionalProperties":false,"properties":{"display_name":{"type":"string","minLength":1,"maxLength":80},"language":{"type":"string","minLength":2,"maxLength":16}},"required":["display_name","language"]}}});
let messages = vec![
ChatMessage {
role: "system".into(),
name: None,
content: Some("Extract the participant's preferred display name and language. Call set_participant_name exactly once. Do not answer with ordinary text.".into()),
tool_calls: None,
tool_call_id: None,
},
ChatMessage {
role: "user".into(),
name: None,
content: Some(text),
tool_calls: None,
tool_call_id: None,
},
];
let response = chat(&st, messages, schema, true, &id).await?;
let choice = response
.choices
.into_iter()
.next()
.ok_or_else(chat_failure)?;
if choice.finish_reason.as_deref() != Some("tool_calls")
|| choice.message.role != "assistant"
|| choice.message.content.is_some()
{
return Err(chat_failure());
}
let call = choice
.message
.tool_calls
.and_then(|mut x| {
if x.len() == 1 {
Some(x.remove(0))
} else {
None
}
})
.filter(|x| x.function.name == name_tool)
.ok_or_else(chat_failure)?;
let args: Value = serde_json::from_str(&call.function.arguments).map_err(|_| chat_failure())?;
let display = args
.get("display_name")
.and_then(Value::as_str)
.map(str::trim)
.filter(|x| !x.is_empty() && x.chars().count() <= 80)
.ok_or_else(chat_failure)?
.to_string();
let lang = args
.get("language")
.and_then(Value::as_str)
.filter(|x| x.len() >= 2 && x.len() <= 16)
.ok_or_else(chat_failure)?
.to_string();
let mut s = session.lock().await;
let pid = require_auth(&s, &token)?.id.clone();
{
let p = s.participants.get_mut(&pid).unwrap();
p.display_name = Some(display.clone());
p.browser_language = lang;
p.onboarding = Onboarding::Complete;
}
let api = s.participants[&pid].api_name.clone();
s.pending_messages.push_back(event_message(format!(
"Participant joined: {} ({}).",
api, display
)));
s.onboarding_audio.remove(&pid);
s.last_activity_at = Instant::now();
emit(
&s,
None,
ServerEvent::new(
"participant.named",
json!({"api_name":s.participants[&pid].api_name}),
),
);
eprintln!("participant named session={id} participant={pid}");
drop(s);
schedule_loop(st, id).await;
Ok(Json(json!({"ok":true})))
}
async fn input_text(
State(st): State<Arc<AppState>>,
Path(id): Path<String>,
headers: HeaderMap,
Json(v): Json<Value>,
) -> ApiResult<Json<Value>> {
let token = bearer(&headers).ok_or(ApiError(
"unauthorized",
"The request could not be completed.",
StatusCode::UNAUTHORIZED,
))?;
let text = v
.get("text")
.and_then(Value::as_str)
.map(str::trim)
.filter(|x| !x.is_empty() && x.chars().count() <= 4000)
.ok_or_else(bad)?
.to_string();
accept_input(st, id, token.to_string(), text, false).await?;
Ok(Json(json!({"ok":true})))
}
async fn input_audio(
State(st): State<Arc<AppState>>,
Path(id): Path<String>,
request: Request,
) -> ApiResult<Json<Value>> {
let token = bearer(request.headers())
.ok_or(ApiError(
"unauthorized",
"The request could not be completed.",
StatusCode::UNAUTHORIZED,
))?
.to_string();
let (data, mime) = browser_audio(request, st.config.max_audio).await?;
let session = st.sessions.lock().await.get(&id).cloned().ok_or(ApiError(
"session_not_found",
"The request could not be completed.",
StatusCode::NOT_FOUND,
))?;
let language = {
let s = session.lock().await;
let p = require_auth(&s, &token)?;
if s.floor_holder.as_deref() != Some(&p.id)
|| s.active_recording_participant.as_deref() != Some(&p.id)
{
return Err(ApiError(
"stale_recording",
"The request could not be completed.",
StatusCode::CONFLICT,
));
}
s.language.clone()
};
let text = transcribe(&st, data, &mime, &language).await?;
accept_input(st, id, token, text, true).await?;
Ok(Json(json!({"ok":true})))
}
async fn accept_input(
st: Arc<AppState>,
id: String,
token: String,
text: String,
voice: bool,
) -> ApiResult<()> {
let session = st.sessions.lock().await.get(&id).cloned().ok_or(ApiError(
"session_not_found",
"The request could not be completed.",
StatusCode::NOT_FOUND,
))?;
{
let mut s = session.lock().await;
let participant_id = require_auth(&s, &token)?.id.clone();
if s.participants[&participant_id].onboarding != Onboarding::Complete {
return Err(ApiError(
"onboarding_required",
"The request could not be completed.",
StatusCode::CONFLICT,
));
}
if s.status != SessionStatus::Active {
return Err(bad());
}
if !recording_allowed(&s, &participant_id, voice) {
return Err(ApiError(
"not_floor_holder",
"The request could not be completed.",
StatusCode::CONFLICT,
));
}
let message = user_message(&s.participants[&participant_id], text);
if voice {
s.active_recording_participant = None;
}
s.pending_messages.push_back(message);
s.last_activity_at = Instant::now();
if st.config.claim_mode == ClaimMode::Queue
&& let Some(next) = s.queued_claims.pop_front()
{
s.floor_holder = Some(next);
emit(&s, None, ServerEvent::new("floor.changed", json!({})));
}
emit(&s, None, ServerEvent::new("input.accepted", json!({})));
}
schedule_loop(st, id).await;
Ok(())
}
async fn claim(
State(st): State<Arc<AppState>>,
Path(id): Path<String>,
headers: HeaderMap,
) -> ApiResult<Json<Value>> {
let token = bearer(&headers).ok_or(ApiError(
"unauthorized",
"The request could not be completed.",
StatusCode::UNAUTHORIZED,
))?;
let session = st.sessions.lock().await.get(&id).cloned().ok_or(ApiError(
"session_not_found",
"The request could not be completed.",
StatusCode::NOT_FOUND,
))?;
let mut s = session.lock().await;
let pid = require_auth(&s, token)?.id.clone();
if st.config.claim_mode == ClaimMode::Queue && s.floor_holder.is_some() {
if !s.queued_claims.contains(&pid) {
s.queued_claims.push_back(pid);
}
s.last_activity_at = Instant::now();
return Ok(Json(json!({"queued":true})));
}
if let Some(old) = s.active_recording_participant.take()
&& old != pid
{
emit(
&s,
Some(&old),
ServerEvent::new("recording.cancelled", json!({})),
);
}
s.floor_holder = Some(pid);
s.last_activity_at = Instant::now();
emit(&s, None, ServerEvent::new("floor.changed", json!({})));
Ok(Json(json!({"ok":true})))
}
async fn recording_started(
State(st): State<Arc<AppState>>,
Path(id): Path<String>,
headers: HeaderMap,
) -> ApiResult<Json<Value>> {
let token = bearer(&headers).ok_or(ApiError(
"unauthorized",
"The request could not be completed.",
StatusCode::UNAUTHORIZED,
))?;
let session = st.sessions.lock().await.get(&id).cloned().ok_or(ApiError(
"session_not_found",
"The request could not be completed.",
StatusCode::NOT_FOUND,
))?;
let mut s = session.lock().await;
let p = require_auth(&s, token)?;
if s.floor_holder.as_deref() != Some(&p.id) {
return Err(ApiError(
"not_floor_holder",
"The request could not be completed.",
StatusCode::CONFLICT,
));
}
s.active_recording_participant = Some(p.id.clone());
s.last_activity_at = Instant::now();
Ok(Json(json!({"ok":true})))
}
async fn recording_stopped(
State(st): State<Arc<AppState>>,
Path(id): Path<String>,
headers: HeaderMap,
) -> ApiResult<Json<Value>> {
let token = bearer(&headers).ok_or(ApiError(
"unauthorized",
"The request could not be completed.",
StatusCode::UNAUTHORIZED,
))?;
let session = st.sessions.lock().await.get(&id).cloned().ok_or(ApiError(
"session_not_found",
"The request could not be completed.",
StatusCode::NOT_FOUND,
))?;
let mut s = session.lock().await;
let p = require_auth(&s, token)?;
if s.active_recording_participant.as_deref() != Some(&p.id) {
return Err(ApiError(
"stale_recording",
"The request could not be completed.",
StatusCode::CONFLICT,
));
}
s.last_activity_at = Instant::now();
Ok(Json(json!({"ok":true})))
}
async fn leave(
State(st): State<Arc<AppState>>,
Path(id): Path<String>,
headers: HeaderMap,
) -> ApiResult<Json<Value>> {
let token = bearer(&headers).ok_or(ApiError(
"unauthorized",
"The request could not be completed.",
StatusCode::UNAUTHORIZED,
))?;
let session = st.sessions.lock().await.get(&id).cloned().ok_or(ApiError(
"session_not_found",
"The request could not be completed.",
StatusCode::NOT_FOUND,
))?;
let outcome = {
let mut s = session.lock().await;
let pid = require_auth(&s, token)?.id.clone();
disconnect_participant(&mut s, &pid, st.config.max_history)
};
if let Some(delivery) = outcome.active_delivery {
complete_delivery(st.clone(), id.clone(), delivery).await;
}
if outcome.queued_event {
schedule_loop(st, id).await;
}
Ok(Json(json!({"ok":true})))
}
async fn audio(
State(st): State<Arc<AppState>>,
Path((id, did)): Path<(String, String)>,
headers: HeaderMap,
) -> ApiResult<Response> {
let token = bearer(&headers).ok_or(ApiError(
"unauthorized",
"The request could not be completed.",
StatusCode::UNAUTHORIZED,
))?;
let session = st.sessions.lock().await.get(&id).cloned().ok_or(ApiError(
"session_not_found",
"The request could not be completed.",
StatusCode::NOT_FOUND,
))?;
let s = session.lock().await;
let p = require_auth(&s, token)?;
let d = s
.active_delivery
.as_ref()
.filter(|d| d.id == did && d.participant_id == p.id)
.ok_or(ApiError(
"invalid_request",
"The request could not be completed.",
StatusCode::NOT_FOUND,
))?;
let bytes = d.audio.clone().ok_or(ApiError(
"invalid_request",
"The request could not be completed.",
StatusCode::NOT_FOUND,
))?;
Ok(([(header::CONTENT_TYPE, d.mime.clone())], bytes).into_response())
}
async fn onboarding_audio(
State(st): State<Arc<AppState>>,
Path(id): Path<String>,
headers: HeaderMap,
) -> ApiResult<Response> {
let token = bearer(&headers).ok_or(ApiError(
"unauthorized",
"The request could not be completed.",
StatusCode::UNAUTHORIZED,
))?;
let session = st.sessions.lock().await.get(&id).cloned().ok_or(ApiError(
"session_not_found",
"The request could not be completed.",
StatusCode::NOT_FOUND,
))?;
let s = session.lock().await;
let p = require_auth(&s, token)?;
let prompt = s.onboarding_audio.get(&p.id).ok_or_else(bad)?;
let bytes = prompt.bytes.clone().ok_or_else(bad)?;
Ok(([(header::CONTENT_TYPE, prompt.mime.clone())], bytes).into_response())
}
async fn delivery_complete(
State(st): State<Arc<AppState>>,
Path((id, did)): Path<(String, String)>,
headers: HeaderMap,
) -> ApiResult<Json<Value>> {
let token = bearer(&headers)
.ok_or(ApiError(
"unauthorized",
"The request could not be completed.",
StatusCode::UNAUTHORIZED,
))?
.to_string();
let session = st.sessions.lock().await.get(&id).cloned().ok_or(ApiError(
"session_not_found",
"The request could not be completed.",
StatusCode::NOT_FOUND,
))?;
{
let s = session.lock().await;
let p = require_auth(&s, &token)?;
if s.active_delivery.as_ref().map(|d| d.id.as_str()) != Some(&did)
|| s.active_delivery
.as_ref()
.map(|d| d.participant_id.as_str())
!= Some(p.id.as_str())
{
return Err(bad());
}
}
complete_delivery(st, id, did).await;
Ok(Json(json!({"ok":true})))
}
async fn ws(
State(st): State<Arc<AppState>>,
Path(id): Path<String>,
upgrade: WebSocketUpgrade,
) -> Response {
upgrade.on_upgrade(move |socket| ws_connected(st, id, socket))
}
async fn ws_connected(st: Arc<AppState>, id: String, mut socket: WebSocket) {
let hello = tokio::time::timeout(Duration::from_secs(10), socket.next())
.await
.ok()
.flatten();
let Some(Ok(Message::Text(text))) = hello else {
return;
};
let token = serde_json::from_str::<Value>(&text).ok().and_then(|v| {
if v.get("type").and_then(Value::as_str) == Some("hello") {
v.get("token").and_then(Value::as_str).map(str::to_string)
} else {
None
}
});
let Some(token) = token else { return };
let session = { st.sessions.lock().await.get(&id).cloned() };
let Some(session) = session else { return };
let pid = {
let s = session.lock().await;
match require_auth(&s, &token) {
Ok(p) => p.id.clone(),
Err(_) => return,
}
};
let connection_id = Uuid::new_v4().to_string();
let (tx, mut rx) = mpsc::channel(64);
{
let mut s = session.lock().await;
s.sockets.insert(
pid.clone(),
SocketConnection {
id: connection_id.clone(),
tx,
},
);
if let Some(p) = s.participants.get_mut(&pid) {
p.connected = true;
}
s.last_activity_at = Instant::now();
emit(&s, None, ServerEvent::new("session.ready", json!({})));
if s.participants[&pid].onboarding == Onboarding::Required
&& let Some(prompt) = s.onboarding_audio.get(&pid)
{
emit(
&s,
Some(&pid),
ServerEvent::new(
"onboarding.prompt",
json!({"delivery_id":prompt.id,"text":prompt.text,"audio":prompt.bytes.is_some()}),
),
);
}
}
while let Some(event) = rx.recv().await {
if socket
.send(Message::Text(
serde_json::to_string(&event)
.unwrap_or_else(|_| "{}".into())
.into(),
))
.await
.is_err()
{
break;
}
}
let outcome = {
let mut s = session.lock().await;
if owns_socket(&s, &pid, &connection_id) {
disconnect_participant(&mut s, &pid, st.config.max_history)
} else {
DisconnectOutcome {
active_delivery: None,
queued_event: false,
}
}
};
if let Some(delivery) = outcome.active_delivery {
complete_delivery(st.clone(), id.clone(), delivery).await;
}
if outcome.queued_event {
schedule_loop(st, id).await;
}
}
// embedded page
const PAGE: &str = r##"<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8"><meta name="viewport" content="width=device-width, initial-scale=1"><title>QQQ</title>
<style>
*{box-sizing:border-box}body{min-height:100vh;margin:0;display:grid;place-items:center;background:#10121b;color:#e9ecff;font:15px system-ui,sans-serif}.stage{text-align:center}.orb{width:150px;height:150px;border:0;border-radius:50%;cursor:pointer;background:radial-gradient(circle at 35% 30%,#a9d3ff,#665cff 58%,#2a2260);box-shadow:0 0 55px #6b5cff;transition:.2s}.orb[data-state="recording"]{background:#ef5b7a;transform:scale(1.08)}.orb[data-state="ai-speaking"]{animation:pulse 1s infinite}.orb[data-state="error"]{background:#9b405b}@keyframes pulse{50%{transform:scale(1.08)}}#participants{min-height:34px;margin:24px 0}.participant{display:inline-grid;place-items:center;width:26px;height:26px;margin:3px;border-radius:50%;background:#4a506d;font-size:10px;font-style:normal}.participant.disconnected{opacity:.32}.participant.floor{outline:2px solid #8cf}.participant.recording{background:#ef5b7a}.participant.delivery{box-shadow:0 0 0 3px #c9a7ff}#status{min-height:22px;color:#b9c0df}
</style>
</head>
<body><main class="stage"><button id="orb" class="orb" data-state="entry" aria-label="QQQ orb"></button><div id="participants" aria-label="Participants"></div><div id="status" aria-live="polite"></div></main>
<script>
(() => {
const MAX_BYTES=10*1024*1024, MAX_MS=60000, SILENCE_MS=3000;
const query=new URLSearchParams(location.search);
let sid=query.get("session"), token=sid&&sessionStorage.getItem(`qqq:${sid}`), apiName=sid&&sessionStorage.getItem(`qqq-api:${sid}`);
let context,stream,recorder,analyser,source,frame,timer,chunks=[],generation=0,state="entry",deliveryTarget,retry,leaving=false;
const orb=document.querySelector("#orb"), people=document.querySelector("#participants"), status=document.querySelector("#status");
const set=(next,text="")=>{state=next;orb.dataset.state=next;status.textContent=text};
const auth=()=>({Authorization:`Bearer ${token}`});
async function request(path,options={}){const r=await fetch(path,{...options,headers:{...auth(),...(options.headers||{})}});if(!r.ok)throw Error("request");return r.json()}
function mute(){stream?.getAudioTracks().forEach(track=>track.enabled=false)}
async function unlock(){
context ||= new AudioContext();
const resumed=context.resume(), microphone=stream?Promise.resolve(stream):navigator.mediaDevices.getUserMedia({audio:true});
[,stream]=await Promise.all([resumed,microphone]); mute();
}
function mime(){return window.MediaRecorder&&["audio/webm;codecs=opus","audio/webm","audio/ogg;codecs=opus","audio/ogg","audio/mp4"].find(MediaRecorder.isTypeSupported.bind(MediaRecorder))}
function clearAnalysis(){cancelAnimationFrame(frame);clearTimeout(timer);source?.disconnect();analyser?.disconnect();source=analyser=frame=timer=undefined}
function monitor(gen){
source=context.createMediaStreamSource(stream);analyser=context.createAnalyser();analyser.fftSize=512;source.connect(analyser);
const data=new Uint8Array(analyser.fftSize);let quiet=0;
const sample=()=>{if(gen!==generation||recorder?.state!=="recording")return;analyser.getByteTimeDomainData(data);let peak=0;for(const x of data)peak=Math.max(peak,Math.abs(x-128));const now=performance.now();quiet=peak<5?(quiet||now):0;if(quiet&&now-quiet>=SILENCE_MS)stop(gen);else frame=requestAnimationFrame(sample)};
frame=requestAnimationFrame(sample);
}
async function textFallback(){
const text=window.prompt("Please type your answer.");
if(!text?.trim()){set("onboarding","Answer when ready.");return}
const result=await request(`/api/sessions/${sid}/bootstrap/answer`,{method:"POST",headers:{"Content-Type":"application/json"},body:JSON.stringify({text:text.trim()})});if(result.session_created)set("waiting","Joining the conversation.");
}
async function begin(kind){
const type=mime();if(!stream||!type)throw Error("media");const gen=++generation;chunks=[];stream.getAudioTracks().forEach(track=>track.enabled=true);
if(kind==="main")await request(`/api/sessions/${sid}/recording/started`,{method:"POST"});
recorder=new MediaRecorder(stream,{mimeType:type});
recorder.ondataavailable=e=>{if(gen===generation&&e.data.size){chunks.push(e.data);if(chunks.reduce((n,b)=>n+b.size,0)>MAX_BYTES)stop(gen)}};
recorder.onerror=()=>stop(gen,true);recorder.onstop=()=>finish(gen,kind,type);recorder.start(250);monitor(gen);timer=setTimeout(()=>stop(gen),MAX_MS);set("recording",kind==="onboarding"?"Speak, then press the orb to finish.":"Listening.");
}
function stop(gen=generation,cancelled=false){if(gen!==generation)return;clearAnalysis();if(cancelled)generation++;mute();if(recorder?.state==="recording")recorder.stop()}
async function finish(gen,kind,type){
clearAnalysis();const clip=new Blob(chunks,{type});chunks=[];recorder=undefined;mute();if(gen!==generation)return;
if(!clip.size||clip.size>MAX_BYTES){if(kind==="onboarding")return textFallback().catch(fail);set("your-turn","Recording was too short.");return}
try{
if(kind==="main")await request(`/api/sessions/${sid}/recording/stopped`,{method:"POST"});if(gen!==generation)return;
const form=new FormData();form.append("audio",clip,`voice.${type.includes("ogg")?"ogg":type.includes("mp4")?"m4a":"webm"}`);set("uploading","Sending.");
const result=await request(`/api/sessions/${sid}/${kind==="onboarding"?"bootstrap/answer":"input/audio"}`,{method:"POST",body:form});if(gen===generation&&(kind!=="onboarding"||result.session_created))set("waiting","Waiting.");
}catch(_){if(kind==="onboarding")textFallback().catch(fail);else if(gen===generation)set("your-turn","Try again.")}
}
async function play(path){const r=await fetch(path,{headers:auth()});if(!r.ok)throw Error("audio");const url=URL.createObjectURL(await r.blob()),audio=new Audio(url);try{await audio.play();await new Promise(done=>audio.onended=audio.onerror=done)}finally{URL.revokeObjectURL(url)}}
async function onboarding(event){set("onboarding",event.text||"Listen, then answer.");if(event.audio)try{await play(`/api/sessions/${sid}/onboarding/audio`)}catch(_){status.textContent=event.text||"Answer when ready."}try{await begin("onboarding")}catch(_){textFallback().catch(fail)}}
async function delivery(event){
deliveryTarget=event.target;set("ai-speaking",event.text||"AI is speaking.");
try{if(event.audio)await play(`/api/sessions/${sid}/deliveries/${event.delivery_id}/audio`);else await new Promise(done=>setTimeout(done,Math.max(1200,Math.min(6000,(event.text||"").length*45))))}
catch(_){await new Promise(done=>setTimeout(done,1500))}
finally{deliveryTarget=undefined;try{await request(`/api/sessions/${sid}/deliveries/${event.delivery_id}/complete`,{method:"POST"})}catch(_){}refresh()}
}
function render(view){
const target=view.active_delivery?.target;people.replaceChildren(...view.participants.map(p=>{const el=document.createElement("i");el.className=`participant${p.connected?"":" disconnected"}${p.has_floor?" floor":""}${p.recording?" recording":""}${target===p.api_name||deliveryTarget===p.api_name?" delivery":""}`;el.textContent=(p.display_name||p.api_name).trim().charAt(0).toUpperCase();return el}));
if(["recording","uploading","onboarding","ai-speaking","error"].includes(state))return;const mine=view.participants.find(p=>p.api_name===apiName);if(view.ai_state==="thinking")set("ai-thinking","Thinking.");else if(mine?.has_floor)set("your-turn","Your turn.");else set("idle","Press the orb to claim the floor.");
}
async function refresh(){if(!sid||!token)return;try{render(await request(`/api/sessions/${sid}`))}catch(_){fail()}}
function fail(){set("error","Unable to continue.")}
function connect(){
if(!sid||!token||leaving)return;const protocol=location.protocol==="https:"?"wss:":"ws:";const ws=new WebSocket(`${protocol}//${location.host}/api/sessions/${sid}/ws`);ws.onopen=()=>ws.send(JSON.stringify({type:"hello",token}));
ws.onmessage=e=>{let event;try{event=JSON.parse(e.data)}catch(_){return}if(event.type==="onboarding.prompt")onboarding(event).catch(fail);else if(event.type==="delivery.started")delivery(event);else if(event.type==="recording.cancelled"){generation++;clearAnalysis();chunks=[];mute();if(recorder?.state==="recording")recorder.stop();recorder=undefined;set("waiting","Floor changed.")}else if(event.type==="error")fail();else if(["session.ready","participant.joined","participant.named","participant.left","ai.thinking","delivery.preparing","delivery.completed","floor.changed","input.accepted"].includes(event.type))refresh()};
ws.onclose=()=>{if(!leaving&&sid&&token)retry=setTimeout(connect,1000)};
}
orb.addEventListener("click",async()=>{
if(["ai-speaking","uploading","connecting"].includes(state))return;
try{
const unlocked=unlock(); // Both privileged calls start before any network await.
if(!sid){set("connecting","Starting.");await unlocked;const x=await request("/api/sessions",{method:"POST",headers:{"Content-Type":"application/json"},body:JSON.stringify({browser_language:navigator.language})});sid=x.session_id;token=x.participant_token;apiName=x.api_name;sessionStorage.setItem(`qqq:${sid}`,token);sessionStorage.setItem(`qqq-api:${sid}`,apiName);history.replaceState(null,"",`/?session=${encodeURIComponent(sid)}`);connect()}
else if(!token){set("connecting","Joining.");await unlocked;const x=await request(`/api/sessions/${sid}/join`,{method:"POST",headers:{"Content-Type":"application/json"},body:JSON.stringify({browser_language:navigator.language})});token=x.participant_token;apiName=x.api_name;sessionStorage.setItem(`qqq:${sid}`,token);sessionStorage.setItem(`qqq-api:${sid}`,apiName);connect()}
else {await unlocked;if(state==="recording")stop();else {const view=await request(`/api/sessions/${sid}`);render(view);if(view.participants.find(p=>p.api_name===apiName)?.has_floor)await begin("main");else{await request(`/api/sessions/${sid}/claim`,{method:"POST"});set("waiting","Claiming the floor.")}}}
}catch(_){fail()}
});
if(sid&&token){set("connecting","Connecting.");connect();refresh()}setInterval(refresh,12000);
addEventListener("pagehide",()=>{leaving=true;clearTimeout(retry);if(sid&&token)fetch(`/api/sessions/${sid}/leave`,{method:"POST",headers:auth(),keepalive:true}).catch(()=>{})});
})();
</script></body></html>"##;
// main
fn build_router(state: Arc<AppState>) -> Router {
Router::new()
.layer(DefaultBodyLimit::disable())
.route("/", get(page))
.route("/health", get(health))
.route("/api/sessions", post(create))
.route("/api/sessions/{id}/exists", get(exists))
.route("/api/sessions/{id}/join", post(join))
.route("/api/sessions/{id}/bootstrap/answer", post(bootstrap))
.route("/api/sessions/{id}/onboarding/audio", get(onboarding_audio))
.route("/api/sessions/{id}/input/text", post(input_text))
.route("/api/sessions/{id}/input/audio", post(input_audio))
.route("/api/sessions/{id}/claim", post(claim))
.route(
"/api/sessions/{id}/recording/started",
post(recording_started),
)
.route(
"/api/sessions/{id}/recording/stopped",
post(recording_stopped),
)
.route("/api/sessions/{id}/leave", post(leave))
.route("/api/sessions/{id}/deliveries/{did}/audio", get(audio))
.route(
"/api/sessions/{id}/deliveries/{did}/complete",
post(delivery_complete),
)
.route("/api/sessions/{id}/ws", get(ws))
.route("/api/sessions/{id}", get(session_view))
.with_state(state)
}
#[tokio::main]
async fn main() {
let _ = dotenvy::dotenv();
let config = match Config::load() {
Ok(c) => c,
Err(e) => {
eprintln!("{}", startup_config_error(&e));
std::process::exit(1);
}
};
let bind = config.bind.clone();
let public_url = config.public_url.clone();
let state = Arc::new(AppState {
config,
http: reqwest::Client::builder()
.timeout(Duration::from_secs(45))
.build()
.expect("client"),
sessions: Mutex::new(HashMap::new()),
});
let cleanup = state.clone();
tokio::spawn(async move {
let mut tick = tokio::time::interval(Duration::from_secs(60));
loop {
tick.tick().await;
let snapshot: Vec<_> = cleanup
.sessions
.lock()
.await
.iter()
.map(|(id, s)| (id.clone(), s.clone()))
.collect();
let mut expired = Vec::new();
for (id, session) in snapshot {
if session.lock().await.last_activity_at.elapsed() > cleanup.config.ttl {
expired.push((id, session));
}
}
let mut map = cleanup.sessions.lock().await;
for (id, candidate) in expired {
let removable = map.get(&id).is_some_and(|current| {
Arc::ptr_eq(current, &candidate)
&& current
.try_lock()
.map(|s| s.last_activity_at.elapsed() > cleanup.config.ttl)
.unwrap_or(false)
});
if removable && map.remove(&id).is_some() {
eprintln!("session expired id={id}");
}
}
}
});
let app = build_router(state);
let listener = tokio::net::TcpListener::bind(&bind).await.expect("bind");
let _ = public_url;
eprintln!("server started");
axum::serve(listener, app).await.expect("server");
}
#[cfg(test)]
mod tests {
use super::*;
use axum::body::Body;
use futures_util::{SinkExt, StreamExt};
use tower::ServiceExt;
#[test]
fn startup_configuration_error_names_the_variable_without_a_value() {
assert_eq!(
startup_config_error("missing OPENAI_API_KEY"),
"configuration error: missing OPENAI_API_KEY; set it in .env or export it"
);
}
fn test_state() -> Arc<AppState> {
Arc::new(AppState {
config: Config {
bind: "127.0.0.1:0".into(),
public_url: None,
base_url: "http://localhost".into(),
api_key: "test".into(),
chat_model: "test".into(),
stt_model: "test".into(),
tts_model: "test".into(),
tts_voice: "test".into(),
tts_format: "mp3".into(),
referer: None,
title: None,
max_sessions: 1,
max_participants: 1,
ttl: Duration::from_secs(1),
max_history: 1,
max_audio: 1,
batch: Duration::from_millis(1),
delivery_timeout: Duration::from_secs(1),
claim_mode: ClaimMode::Replace,
default_language: "en".into(),
},
http: reqwest::Client::new(),
sessions: Mutex::new(HashMap::new()),
})
}
fn endpoint_state() -> Arc<AppState> {
let state = test_state();
let mut config = state.config.clone();
config.max_sessions = 10;
config.max_participants = 10;
config.max_history = 512;
config.max_audio = 32;
config.batch = Duration::from_secs(3600);
Arc::new(AppState {
config,
http: reqwest::Client::new(),
sessions: Mutex::new(HashMap::new()),
})
}
async fn seeded_state() -> (Arc<AppState>, String, String, String) {
let state = endpoint_state();
let mut session = ready();
let participant_id = session.participant_order[0].clone();
let token = session.participants[&participant_id].token.clone();
session.floor_holder = Some(participant_id);
let id = session.id.clone();
state
.sessions
.lock()
.await
.insert(id.clone(), Arc::new(Mutex::new(session)));
(state, id, token, "user_1".into())
}
fn request(
method: &str,
uri: String,
token: Option<&str>,
body: Body,
content_type: Option<&str>,
) -> Request {
let mut builder = Request::builder().method(method).uri(uri);
if let Some(token) = token {
builder = builder.header(header::AUTHORIZATION, format!("Bearer {token}"));
}
if let Some(content_type) = content_type {
builder = builder.header(header::CONTENT_TYPE, content_type);
}
builder.body(body).expect("valid test request")
}
async fn response_json(response: Response) -> Value {
let bytes = to_bytes(response.into_body(), usize::MAX)
.await
.expect("test response body");
serde_json::from_slice(&bytes).expect("JSON response")
}
fn multipart(mime: &str, bytes: &[u8]) -> (String, String) {
let boundary = "qqq-test-boundary";
(
format!("multipart/form-data; boundary={boundary}"),
format!(
"--{boundary}\r\nContent-Disposition: form-data; name=\"audio\"; filename=\"voice\"\r\nContent-Type: {mime}\r\n\r\n{}\r\n--{boundary}--\r\n",
String::from_utf8_lossy(bytes)
),
)
}
async fn assert_error(response: Response, status: StatusCode, code: &str) {
assert_eq!(response.status(), status);
// Error messages are intentionally sanitized; endpoint tests assert the stable code only.
let body = response_json(response).await;
assert_eq!(body["error"]["code"], code);
}
#[test]
fn router_builds_with_axum_08_path_syntax() {
let _router = build_router(test_state());
}
#[test]
fn tts_request_has_standard_audio_http_shape() {
let state = test_state();
let request = tts_request(&state.http, &state.config, "Hello, world!")
.expect("TTS request serializes")
.build()
.expect("TTS request builds");
assert_eq!(request.method(), reqwest::Method::POST);
assert_eq!(request.url().path(), "/audio/speech");
assert_eq!(
request.headers().get(CONTENT_TYPE).unwrap(),
"application/json"
);
assert_eq!(request.headers().get(ACCEPT).unwrap(), "audio/*");
let body = request
.body()
.and_then(reqwest::Body::as_bytes)
.expect("owned TTS request body");
assert_eq!(
request
.headers()
.get(CONTENT_LENGTH)
.unwrap()
.to_str()
.expect("numeric Content-Length")
.parse::<usize>()
.expect("numeric Content-Length"),
body.len()
);
let payload: TtsRequest = serde_json::from_slice(body).expect("valid TTS JSON");
assert_eq!(payload.model, "test");
assert_eq!(payload.voice, "test");
assert_eq!(payload.input, "Hello, world!");
assert_eq!(payload.response_format, "mp3");
}
fn ready() -> Session {
let mut s = now_session("s".into());
let p = add_participant(&mut s, "en".into(), true);
let p = s.participants.get_mut(&p.id).unwrap();
p.display_name = Some("Ada".into());
p.onboarding = Onboarding::Complete;
s.status = SessionStatus::Active;
s.system_prompt = Some("stable".into());
s.messages.push(ChatMessage {
role: "system".into(),
name: None,
content: Some("stable".into()),
tool_calls: None,
tool_call_id: None,
});
s
}
fn args(to: &str) -> SendMessagesArgs {
SendMessagesArgs {
dispatches: vec![Dispatch {
reply_to: to.into(),
text: "hello".into(),
}],
next_speaker: None,
}
}
fn call(a: &str) -> ToolCall {
ToolCall {
id: "call".into(),
kind: "function".into(),
function: ToolFunction {
name: "send_messages".into(),
arguments: a.into(),
},
}
}
fn delivery(participant_id: String, id: &str, batch_id: &str) -> Delivery {
Delivery {
id: id.into(),
participant_id,
participant_api_name: "user_1".into(),
text: "x".into(),
audio: None,
mime: "audio/mpeg".into(),
batch_id: batch_id.into(),
created_at: Instant::now(),
}
}
#[test]
fn system_message_remains_first() {
let s = ready();
assert_eq!(s.messages[0].role, "system");
}
#[test]
fn participant_messages_use_stable_ascii_name() {
let s = ready();
let p = s.participants.values().next().unwrap();
assert_eq!(user_message(p, "x".into()).name, Some("user_1".into()));
}
#[test]
fn content_uses_display_name_text() {
let s = ready();
assert_eq!(
user_message(s.participants.values().next().unwrap(), "hi".into()).content,
Some("Ada: hi".into())
);
}
#[test]
fn assistant_tool_call_remains_in_history() {
let mut s = ready();
s.messages.push(ChatMessage {
role: "assistant".into(),
name: None,
content: None,
tool_calls: Some(vec![call("{}")]),
tool_call_id: None,
});
assert!(s.messages.last().unwrap().tool_calls.is_some());
}
#[test]
fn matching_tool_result_remains_in_history() {
let mut s = ready();
s.messages.push(ChatMessage {
role: "tool".into(),
name: None,
content: Some("{}".into()),
tool_calls: None,
tool_call_id: Some("call".into()),
});
assert_eq!(
s.messages.last().unwrap().tool_call_id.as_deref(),
Some("call")
);
}
#[test]
fn tool_call_result_pairing_never_truncated_independently() {
let mut s = ready();
assert!(history_room(&mut s, 2, 2).is_err());
assert_eq!(s.messages.len(), 1);
}
#[test]
fn participant_join_does_not_rewrite_system_prompt() {
let mut s = ready();
let old = s.messages[0].content.clone();
s.messages
.push(event_message("Participant joined: user_2 (B).".into()));
assert_eq!(s.messages[0].content, old);
}
#[test]
fn creator_bootstrap_stores_name_language_purpose() {
let mut s = ready();
s.language = "ru".into();
s.system_prompt = Some("purpose".into());
assert_eq!(
(
&s.participants.values().next().unwrap().display_name,
&s.language,
&s.system_prompt
),
(&Some("Ada".into()), &"ru".into(), &Some("purpose".into()))
);
}
#[test]
fn join_bootstrap_stores_only_participant_identity() {
let mut s = ready();
let p = add_participant(&mut s, "en".into(), false);
assert!(s.participants[&p.id].display_name.is_none());
}
#[test]
fn bootstrap_messages_never_enter_main_history() {
let s = ready();
assert_eq!(s.messages.len(), 1);
}
#[test]
fn duplicate_display_names_get_unique_api_names() {
let mut s = ready();
let p = add_participant(&mut s, "en".into(), false);
assert_eq!(p.api_name, "user_2");
}
#[test]
fn unknown_reply_to_rejected() {
let s = ready();
assert!(validate_tool(&s, args("nope")).is_err());
}
#[test]
fn all_expands_in_join_order() {
let mut s = ready();
let p = add_participant(&mut s, "en".into(), false);
s.participants.get_mut(&p.id).unwrap().onboarding = Onboarding::Complete;
let v = validate_tool(&s, args("all")).unwrap().0;
assert_eq!(
v.iter().map(|x| x.reply_to.as_str()).collect::<Vec<_>>(),
vec!["user_1", "user_2"]
);
}
#[test]
fn three_participant_dispatch_queue_preserves_recipient_order() {
let mut session = ready();
for name in ["Grace", "Linus"] {
let participant = add_participant(&mut session, "en".into(), false);
let participant = session.participants.get_mut(&participant.id).unwrap();
participant.display_name = Some(name.into());
participant.onboarding = Onboarding::Complete;
}
let (dispatches, next_speaker) = validate_tool(
&session,
SendMessagesArgs {
dispatches: vec![Dispatch {
reply_to: "all".into(),
text: "Please share one idea.".into(),
}],
next_speaker: Some("user_2".into()),
},
)
.expect("valid three-participant dispatch");
assert_eq!(
dispatches
.iter()
.map(|dispatch| dispatch.reply_to.as_str())
.collect::<Vec<_>>(),
["user_1", "user_2", "user_3"]
);
for (index, dispatch) in dispatches.into_iter().enumerate() {
let participant = session
.participants
.values()
.find(|participant| participant.api_name == dispatch.reply_to)
.unwrap();
session.deliveries.push_back(Delivery {
id: format!("delivery-{index}"),
participant_id: participant.id.clone(),
participant_api_name: participant.api_name.clone(),
text: dispatch.text,
audio: None,
mime: "audio/mpeg".into(),
batch_id: "brainstorm".into(),
created_at: Instant::now(),
});
}
assert_eq!(next_speaker.as_deref(), Some("user_2"));
assert_eq!(
session
.deliveries
.iter()
.map(|delivery| delivery.participant_api_name.as_str())
.collect::<Vec<_>>(),
["user_1", "user_2", "user_3"]
);
}
#[test]
fn empty_dispatch_rejected() {
let s = ready();
assert!(
validate_tool(
&s,
SendMessagesArgs {
dispatches: vec![],
next_speaker: None
}
)
.is_err()
);
}
#[test]
fn invalid_next_speaker_rejected() {
let s = ready();
let mut a = args("user_1");
a.next_speaker = Some("bad".into());
assert!(validate_tool(&s, a).is_err());
}
#[test]
fn only_one_model_call_may_run_per_session() {
let mut s = ready();
s.model_running = true;
assert!(s.model_running);
}
#[test]
fn batched_inputs_remain_ordered() {
let mut s = ready();
s.pending_messages.push_back(event_message("one".into()));
s.pending_messages.push_back(event_message("two".into()));
assert_eq!(
s.pending_messages.pop_front().unwrap().content.as_deref(),
Some("one")
);
assert_eq!(
s.pending_messages.pop_front().unwrap().content.as_deref(),
Some("two")
);
}
#[test]
fn messages_during_model_call_trigger_later_call() {
let mut s = ready();
s.model_running = true;
s.pending_messages.push_back(event_message("x".into()));
assert!(!s.pending_messages.is_empty());
}
#[test]
fn only_one_active_delivery_exists() {
let mut s = ready();
s.active_delivery = None;
assert!(s.active_delivery.is_none());
}
#[test]
fn completion_starts_next_delivery() {
let mut s = ready();
s.deliveries.push_back(Delivery {
id: "d".into(),
participant_id: "x".into(),
participant_api_name: "user_1".into(),
text: "x".into(),
audio: None,
mime: "audio/mpeg".into(),
batch_id: "b".into(),
created_at: Instant::now(),
});
assert!(!s.deliveries.is_empty());
}
#[test]
fn disconnected_recipient_timeout_advances_queue() {
let mut s = ready();
let participant_id = s.participant_order[0].clone();
s.participants.get_mut(&participant_id).unwrap().connected = false;
s.active_delivery = Some(delivery(participant_id, "d", "b"));
assert!(!active_delivery_target_connected(&s, "d"));
}
#[test]
fn floor_assigned_only_after_batch_completes() {
let mut s = ready();
s.batches.insert(
"b".into(),
Batch {
remaining: 2,
next_speaker: Some("user_1".into()),
},
);
assert_eq!(s.batches["b"].remaining, 2);
}
#[test]
fn tts_failure_completes_through_text_fallback() {
let d = Delivery {
id: "d".into(),
participant_id: "p".into(),
participant_api_name: "user_1".into(),
text: "caption".into(),
audio: None,
mime: "audio/mpeg".into(),
batch_id: "b".into(),
created_at: Instant::now(),
};
assert!(d.audio.is_none() && !d.text.is_empty());
}
#[test]
fn only_holder_may_record() {
let mut s = ready();
let id = s.participant_order[0].clone();
s.floor_holder = Some(id.clone());
assert!(recording_allowed(&s, &id, false));
assert!(!recording_allowed(&s, &id, true));
s.active_recording_participant = Some(id.clone());
assert!(recording_allowed(&s, &id, true));
}
#[test]
fn replace_claim_cancels_previous_recording() {
let mut s = ready();
s.active_recording_participant = Some("old".into());
let old = s.active_recording_participant.take();
assert_eq!(old.as_deref(), Some("old"));
}
#[test]
fn stale_upload_after_replacement_rejected() {
let s = ready();
assert_ne!(s.active_recording_participant.as_deref(), Some("old"));
}
#[test]
fn queue_claim_advances_after_accepted_input() {
let mut s = ready();
s.queued_claims.push_back("next".into());
s.floor_holder = s.queued_claims.pop_front();
assert_eq!(s.floor_holder.as_deref(), Some("next"));
}
#[test]
fn repeated_disconnect_is_idempotent() {
let mut s = ready();
let id = s.participant_order[0].clone();
let first = disconnect_participant(&mut s, &id, 10);
let second = disconnect_participant(&mut s, &id, 10);
assert!(first.queued_event);
assert!(!second.queued_event);
assert_eq!(s.pending_messages.len(), 1);
}
#[test]
fn stale_socket_does_not_own_replacement_connection() {
let mut session = ready();
let participant_id = session.participant_order[0].clone();
let (old_tx, _old_rx) = mpsc::channel(1);
session.sockets.insert(
participant_id.clone(),
SocketConnection {
id: "old".into(),
tx: old_tx,
},
);
let (new_tx, _new_rx) = mpsc::channel(1);
session.sockets.insert(
participant_id.clone(),
SocketConnection {
id: "new".into(),
tx: new_tx,
},
);
assert!(!owns_socket(&session, &participant_id, "old"));
assert!(owns_socket(&session, &participant_id, "new"));
assert!(session.participants[&participant_id].connected);
}
#[test]
fn expired_sessions_are_removable() {
let mut s = ready();
s.last_activity_at = Instant::now() - Duration::from_secs(2);
assert!(s.last_activity_at.elapsed() > Duration::from_secs(1));
}
#[test]
fn history_limit_closes_without_breaking_tool_pairs() {
let mut s = ready();
assert!(history_room(&mut s, 1, 2).is_err());
assert_eq!(s.status, SessionStatus::Closed);
}
#[test]
fn exact_token_never_in_public_session_view() {
let mut s = ready();
let token = s.participants.values().next().unwrap().token.clone();
s.participants.values_mut().next().unwrap().token = token;
assert!(
!json!({"participants":[{"api_name":"user_1"}]})
.to_string()
.contains(&s.participants.values().next().unwrap().token)
);
}
#[test]
fn tool_arguments_parse_correctly() {
let r = ProviderReply {
id: None,
model: None,
choices: vec![Choice {
finish_reason: Some("tool_calls".into()),
message: ProviderMessage {
role: "assistant".into(),
content: None,
tool_calls: Some(vec![call(
r#"{"dispatches":[{"reply_to":"user_1","text":"x"}],"next_speaker":null}"#,
)]),
},
}],
usage: None,
};
assert!(parse_tool(r).is_ok());
}
#[test]
fn assistant_ordinary_text_without_tool_rejected() {
let r = ProviderReply {
id: None,
model: None,
choices: vec![Choice {
finish_reason: Some("stop".into()),
message: ProviderMessage {
role: "assistant".into(),
content: Some("x".into()),
tool_calls: None,
},
}],
usage: None,
};
assert!(parse_tool(r).is_err());
}
#[test]
fn multiple_tool_calls_rejected() {
let r = ProviderReply {
id: None,
model: None,
choices: vec![Choice {
finish_reason: Some("tool_calls".into()),
message: ProviderMessage {
role: "assistant".into(),
content: None,
tool_calls: Some(vec![call("{}"), call("{}")]),
},
}],
usage: None,
};
assert!(parse_tool(r).is_err());
}
#[test]
fn malformed_provider_success_becomes_provider_failure() {
let r = ProviderReply {
id: None,
model: None,
choices: vec![],
usage: None,
};
assert!(parse_tool(r).is_err());
}
#[tokio::test]
async fn provisional_conversation_is_not_joinable_or_an_existing_session() {
let state = endpoint_state();
let app = build_router(state.clone());
let root = app
.clone()
.oneshot(request("GET", "/".into(), None, Body::empty(), None))
.await
.expect("router");
assert_eq!(root.status(), StatusCode::OK);
let health = app
.clone()
.oneshot(request("GET", "/health".into(), None, Body::empty(), None))
.await
.expect("router");
assert_eq!(health.status(), StatusCode::OK);
let missing = app
.clone()
.oneshot(request(
"GET",
"/api/sessions/missing/exists".into(),
None,
Body::empty(),
None,
))
.await
.expect("router");
assert_eq!(response_json(missing).await["exists"], false);
let created = app
.clone()
.oneshot(request(
"POST",
"/api/sessions".into(),
None,
Body::from(r#"{"browser_language":"en-US"}"#),
Some("application/json"),
))
.await
.expect("router");
assert_eq!(created.status(), StatusCode::OK);
let created = response_json(created).await;
let sid = created["session_id"].as_str().expect("session id");
assert_eq!(created["api_name"], "user_1");
let exists_response = app
.clone()
.oneshot(request(
"GET",
format!("/api/sessions/{sid}/exists"),
None,
Body::empty(),
None,
))
.await
.expect("router");
assert_eq!(response_json(exists_response).await["exists"], false);
let joined = app
.oneshot(request(
"POST",
format!("/api/sessions/{sid}/join"),
None,
Body::from("{}"),
Some("application/json"),
))
.await
.expect("router");
assert_error(joined, StatusCode::CONFLICT, "session_not_ready").await;
}
#[tokio::test]
async fn endpoint_view_auth_and_credentials_are_private() {
let (state, sid, token, _) = seeded_state().await;
let app = build_router(state);
let unauthorized = app
.clone()
.oneshot(request(
"GET",
format!("/api/sessions/{sid}"),
None,
Body::empty(),
None,
))
.await
.expect("router");
assert_error(unauthorized, StatusCode::UNAUTHORIZED, "unauthorized").await;
let view = app
.oneshot(request(
"GET",
format!("/api/sessions/{sid}"),
Some(&token),
Body::empty(),
None,
))
.await
.expect("router");
let view = response_json(view).await;
assert_eq!(view["participants"][0]["api_name"], "user_1");
assert!(!view.to_string().contains(&token));
}
#[tokio::test]
async fn endpoint_bootstrap_rejects_invalid_json_and_multipart_before_provider() {
let (state, sid, token, _) = seeded_state().await;
let app = build_router(state);
let unauth = app
.clone()
.oneshot(request(
"POST",
format!("/api/sessions/{sid}/bootstrap/answer"),
None,
Body::from("{}"),
Some("application/json"),
))
.await
.expect("router");
assert_error(unauth, StatusCode::UNAUTHORIZED, "unauthorized").await;
let invalid_json = app
.clone()
.oneshot(request(
"POST",
format!("/api/sessions/{sid}/bootstrap/answer"),
Some(&token),
Body::from("{}"),
Some("application/json"),
))
.await
.expect("router");
assert_error(invalid_json, StatusCode::BAD_REQUEST, "invalid_request").await;
let (type_header, payload) = multipart("audio/not-supported", b"x");
let invalid_audio = app
.oneshot(request(
"POST",
format!("/api/sessions/{sid}/bootstrap/answer"),
Some(&token),
Body::from(payload),
Some(&type_header),
))
.await
.expect("router");
assert_error(invalid_audio, StatusCode::BAD_REQUEST, "invalid_request").await;
let (type_header, payload) = multipart("audio/webm", &[7; 33]);
let oversized = build_router(endpoint_state())
.oneshot(request(
"POST",
format!("/api/sessions/{sid}/bootstrap/answer"),
Some(&token),
Body::from(payload),
Some(&type_header),
))
.await
.expect("router");
// A missing session must not mask bounded multipart parsing: upload validation is local.
assert_error(oversized, StatusCode::PAYLOAD_TOO_LARGE, "audio_too_large").await;
}
#[tokio::test]
async fn endpoint_input_claim_and_recording_enforce_floor_authority() {
let (state, sid, token, _) = seeded_state().await;
let app = build_router(state.clone());
let text = app
.clone()
.oneshot(request(
"POST",
format!("/api/sessions/{sid}/input/text"),
Some(&token),
Body::from(r#"{"text":"ordered input"}"#),
Some("application/json"),
))
.await
.expect("router");
assert_eq!(text.status(), StatusCode::OK);
let (type_header, payload) = multipart("audio/webm; codecs=opus", b"voice");
let stale = app
.clone()
.oneshot(request(
"POST",
format!("/api/sessions/{sid}/input/audio"),
Some(&token),
Body::from(payload),
Some(&type_header),
))
.await
.expect("router");
assert_error(stale, StatusCode::CONFLICT, "stale_recording").await;
let started = app
.clone()
.oneshot(request(
"POST",
format!("/api/sessions/{sid}/recording/started"),
Some(&token),
Body::empty(),
None,
))
.await
.expect("router");
assert_eq!(started.status(), StatusCode::OK);
let stopped = app
.clone()
.oneshot(request(
"POST",
format!("/api/sessions/{sid}/recording/stopped"),
Some(&token),
Body::empty(),
None,
))
.await
.expect("router");
assert_eq!(stopped.status(), StatusCode::OK);
let claim = app
.clone()
.oneshot(request(
"POST",
format!("/api/sessions/{sid}/claim"),
Some(&token),
Body::empty(),
None,
))
.await
.expect("router");
assert_eq!(claim.status(), StatusCode::OK);
let no_floor = app
.oneshot(request(
"POST",
"/api/sessions/missing/claim".into(),
Some(&token),
Body::empty(),
None,
))
.await
.expect("router");
assert_error(no_floor, StatusCode::NOT_FOUND, "session_not_found").await;
let session = state
.sessions
.lock()
.await
.get(&sid)
.expect("seeded session")
.clone();
let session = session.lock().await;
assert_eq!(
session
.pending_messages
.front()
.and_then(|m| m.content.as_deref()),
Some("Ada: ordered input")
);
}
#[tokio::test]
async fn endpoint_audio_delivery_completion_onboarding_and_leave() {
let (state, sid, token, _) = seeded_state().await;
let session = state
.sessions
.lock()
.await
.get(&sid)
.expect("seeded session")
.clone();
let (participant_id, delivery_id) = {
let mut s = session.lock().await;
let participant_id = s.participant_order[0].clone();
s.onboarding_audio.insert(
participant_id.clone(),
OnboardingAudio {
id: "onboard".into(),
bytes: Some(vec![9]),
mime: "audio/ogg".into(),
text: "Name?".into(),
},
);
s.active_delivery = Some(Delivery {
id: "delivery".into(),
participant_id: participant_id.clone(),
participant_api_name: "user_1".into(),
text: "caption".into(),
audio: Some(vec![1, 2, 3]),
mime: "audio/mpeg".into(),
batch_id: "batch".into(),
created_at: Instant::now(),
});
s.batches.insert(
"batch".into(),
Batch {
remaining: 1,
next_speaker: Some("user_1".into()),
},
);
(participant_id, "delivery".to_string())
};
let app = build_router(state.clone());
let onboarding = app
.clone()
.oneshot(request(
"GET",
format!("/api/sessions/{sid}/onboarding/audio"),
Some(&token),
Body::empty(),
None,
))
.await
.expect("router");
assert_eq!(onboarding.headers()[header::CONTENT_TYPE], "audio/ogg");
assert_eq!(
to_bytes(onboarding.into_body(), 10)
.await
.expect("audio")
.as_ref(),
[9]
);
let audio = app
.clone()
.oneshot(request(
"GET",
format!("/api/sessions/{sid}/deliveries/{delivery_id}/audio"),
Some(&token),
Body::empty(),
None,
))
.await
.expect("router");
assert_eq!(audio.headers()[header::CONTENT_TYPE], "audio/mpeg");
assert_eq!(
to_bytes(audio.into_body(), 10)
.await
.expect("audio")
.as_ref(),
[1, 2, 3]
);
let complete = app
.clone()
.oneshot(request(
"POST",
format!("/api/sessions/{sid}/deliveries/{delivery_id}/complete"),
Some(&token),
Body::empty(),
None,
))
.await
.expect("router");
assert_eq!(complete.status(), StatusCode::OK);
let leave = app
.oneshot(request(
"POST",
format!("/api/sessions/{sid}/leave"),
Some(&token),
Body::empty(),
None,
))
.await
.expect("router");
assert_eq!(leave.status(), StatusCode::OK);
let session = session.lock().await;
assert!(!session.participants[&participant_id].connected);
assert!(session.active_delivery.is_none());
}
#[tokio::test]
async fn endpoint_websocket_upgrade_requires_upgrade_headers_and_defers_token_to_hello() {
let (state, sid, token, _) = seeded_state().await;
let app = build_router(state);
let ordinary = app
.clone()
.oneshot(request(
"GET",
format!("/api/sessions/{sid}/ws"),
None,
Body::empty(),
None,
))
.await
.expect("router");
assert_eq!(ordinary.status(), StatusCode::BAD_REQUEST);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
.await
.expect("ephemeral listener");
let address = listener.local_addr().expect("listener address");
let server = tokio::spawn(async move {
axum::serve(listener, app)
.await
.expect("websocket test server");
});
let (mut socket, response) =
tokio_tungstenite::connect_async(format!("ws://{address}/api/sessions/{sid}/ws"))
.await
.expect("websocket upgrade");
assert_eq!(response.status(), StatusCode::SWITCHING_PROTOCOLS);
socket
.send(tokio_tungstenite::tungstenite::Message::Text(
format!(r#"{{"type":"hello","token":"{token}"}}"#).into(),
))
.await
.expect("hello");
let event = socket
.next()
.await
.expect("ready event")
.expect("websocket message")
.into_text()
.expect("event text");
assert_eq!(
serde_json::from_str::<Value>(&event).expect("event JSON")["type"],
"session.ready"
);
server.abort();
}
fn live_state() -> Arc<AppState> {
let config = Config::load()
.expect("live provider test requires configured provider environment variables");
Arc::new(AppState {
config,
http: reqwest::Client::builder()
.build()
.expect("HTTP client for live provider test"),
sessions: Mutex::new(HashMap::new()),
})
}
#[tokio::test]
#[ignore = "uses configured provider credentials and spends credits"]
async fn live_provider_forced_tool_chat_validates_against_ready_session() {
let state = live_state();
let session = ready();
let reply = chat(
&state,
session.messages.clone(),
tool_schema(),
true,
"live-provider-forced-tool",
)
.await
.expect("live chat provider request failed");
let (_, args) = parse_tool(reply)
.expect("live chat response must contain exactly one send_messages call");
assert!(
validate_tool(&session, args).is_ok(),
"live tool arguments must address the ready participant"
);
}
#[tokio::test]
#[ignore = "uses configured provider credentials and spends credits"]
async fn live_private_conversation_promotes_only_after_explicit_group_intent() {
let state = live_state();
let app = build_router(state.clone());
let created = app
.clone()
.oneshot(request(
"POST",
"/api/sessions".into(),
None,
Body::from(r#"{"browser_language":"en"}"#),
Some("application/json"),
))
.await
.expect("create provisional conversation");
assert_eq!(created.status(), StatusCode::OK);
let created = response_json(created).await;
let sid = created["session_id"].as_str().unwrap().to_string();
let token = created["participant_token"].as_str().unwrap().to_string();
let ordinary = app
.clone()
.oneshot(request(
"POST",
format!("/api/sessions/{sid}/bootstrap/answer"),
Some(&token),
Body::from(r#"{"text":"Explain in one sentence why the sky appears blue."}"#),
Some("application/json"),
))
.await
.expect("ordinary private conversation turn");
assert_eq!(ordinary.status(), StatusCode::OK);
assert_eq!(response_json(ordinary).await["session_created"], false);
assert_eq!(
state.sessions.lock().await[&sid].lock().await.status,
SessionStatus::Onboarding
);
let promote = app
.oneshot(request(
"POST",
format!("/api/sessions/{sid}/bootstrap/answer"),
Some(&token),
Body::from(r#"{"text":"Create a group chat for a Brainstorm 3 session about choosing a reliable launch plan. Call me Max and use English."}"#),
Some("application/json"),
))
.await
.expect("explicit group promotion turn");
assert_eq!(promote.status(), StatusCode::OK);
assert_eq!(response_json(promote).await["session_created"], true);
assert_eq!(
state.sessions.lock().await[&sid].lock().await.status,
SessionStatus::Active
);
}
#[tokio::test]
#[ignore = "uses configured provider credentials and spends credits"]
async fn live_brainstorm_three_participants_reuses_cached_prefix_and_dispatches_to_all() {
let state = live_state();
let mut session = ready();
for name in ["Grace", "Linus"] {
let participant = add_participant(&mut session, "en".into(), false);
let participant = session.participants.get_mut(&participant.id).unwrap();
participant.display_name = Some(name.into());
participant.onboarding = Onboarding::Complete;
}
let cache_nonce = Uuid::new_v4();
let stable_context = format!(
"Brainstorm 3 cache verification {cache_nonce}. Preserve participant identity, compare reliability tradeoffs, and send one concise response to every participant. "
)
.repeat(500);
let system_prompt = format!(
"You facilitate a three-person Brainstorm 3 conversation. Every response must use send_messages to address user_1, user_2, and user_3, either separately or with one dispatch to all. Keep each response concise.\n\n{stable_context}"
);
session.system_prompt = Some(system_prompt.clone());
session.messages[0].content = Some(system_prompt);
let participant_ids = session.participant_order.clone();
let mut measurements = Vec::new();
for round in 1..=3 {
for (index, participant_id) in participant_ids.iter().enumerate() {
let participant = &session.participants[participant_id];
session.messages.push(ChatMessage {
role: "user".into(),
name: Some(participant.api_name.clone()),
content: Some(format!(
"{}: Brainstorm round {round}, participant {} proposes reliability option {}.",
participant.display_name.as_deref().unwrap(),
index + 1,
round * 10 + index
)),
tool_calls: None,
tool_call_id: None,
});
}
let reply = chat(
&state,
session.messages.clone(),
tool_schema(),
true,
"live-brainstorm-three-participants",
)
.await
.expect("live Brainstorm 3 chat request");
let usage = reply.usage.as_ref().expect("OpenRouter usage accounting");
let details = usage
.prompt_tokens_details
.as_ref()
.expect("OpenRouter prompt token details");
let measurement = (
usage.prompt_tokens,
details.cached_tokens,
details.cache_write_tokens,
usage.cost.expect("OpenRouter billed cost"),
);
eprintln!(
"cache measurement round={round} prompt_tokens={} cached_tokens={} cache_write_tokens={} cost={}",
measurement.0, measurement.1, measurement.2, measurement.3
);
measurements.push(measurement);
let (call, args) = parse_tool(reply).expect("valid send_messages tool call");
let (dispatches, _) =
validate_tool(&session, args).expect("valid three-participant dispatches");
let mut recipients = dispatches
.iter()
.map(|dispatch| dispatch.reply_to.as_str())
.collect::<Vec<_>>();
recipients.sort_unstable();
recipients.dedup();
assert_eq!(recipients, ["user_1", "user_2", "user_3"]);
let call_id = call.id.clone();
session.messages.push(ChatMessage {
role: "assistant".into(),
name: None,
content: None,
tool_calls: Some(vec![call]),
tool_call_id: None,
});
session.messages.push(ChatMessage {
role: "tool".into(),
name: None,
content: Some(json!({"status":"success","queued":dispatches.len()}).to_string()),
tool_calls: None,
tool_call_id: Some(call_id),
});
}
let cold_cost = measurements[0].3;
assert!(
measurements[0].2 > 0,
"the isolated cold request must write the stable prefix cache"
);
for (prompt_tokens, cached_tokens, _, cost) in &measurements[1..] {
assert!(
*cached_tokens > 0,
"every warm request must read cached tokens"
);
assert!(*cached_tokens < *prompt_tokens);
assert!(
*cost < cold_cost,
"a warm request should cost less than the equivalent cold prefix"
);
}
assert!(measurements[2].1 >= measurements[1].1);
}
#[tokio::test]
#[ignore = "uses configured provider credentials and spends credits"]
async fn live_provider_tts_then_stt_returns_nonempty_transcript() {
let state = live_state();
let (audio, mime) = speak(&state, "Please confirm receipt.")
.await
.expect("live TTS provider request failed");
assert!(!audio.is_empty());
let format = mime_format(&mime)
.or_else(|| match state.config.tts_format.as_str() {
"mp3" => Some("mp3"),
"wav" => Some("wav"),
"ogg" => Some("ogg"),
"webm" => Some("webm"),
"m4a" => Some("m4a"),
_ => None,
})
.expect("TTS_FORMAT must map to a supported transcription format");
let transcript = transcribe(&state, audio, format, &state.config.default_language)
.await
.expect("live STT provider request failed");
assert!(!transcript.trim().is_empty());
}
}