use super::GatewayState; use crate::config::Config; use crate::memory::MemoryCategory; use axum::Json; use axum::body::Body; use axum::extract::{Multipart, Path, Query, State}; use axum::http::{StatusCode, header}; use axum::response::{IntoResponse, Response}; use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; use std::collections::VecDeque; use std::path::{Path as FsPath, PathBuf}; use std::sync::Arc; use tokio::io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt}; use tokio_util::io::ReaderStream; const REDACTED: &str = "********"; const MAX_CONFIG_BYTES: usize = 1024 * 1024; const MAX_PROFILE_BYTES: usize = 256 * 1024; struct TemporaryUpload { path: PathBuf, committed: bool, } impl Drop for TemporaryUpload { fn drop(&mut self) { if !self.committed { let _ = std::fs::remove_file(&self.path); } } } #[derive(Serialize)] pub struct HealthResponse { status: String, version: String, } pub async fn health() -> Json { Json(HealthResponse { status: "ok".to_string(), version: env!("CARGO_PKG_VERSION").to_string(), }) } fn static_response(content_type: &'static str, content: &'static str) -> Response { Response::builder() .header(header::CONTENT_TYPE, content_type) .header(header::CACHE_CONTROL, "no-cache") .header("X-Content-Type-Options", "nosniff") .header( "Content-Security-Policy", "default-src 'self'; connect-src 'self' ws: wss:; img-src 'self' data:; style-src 'self'; script-src 'self'; base-uri 'none'; frame-ancestors 'none'", ) .body(Body::from(content)) .expect("valid static response") } pub async fn webui_index() -> Response { static_response( "text/html; charset=utf-8", include_str!(concat!(env!("OUT_DIR"), "/webui/index.html")), ) } pub async fn webui_script() -> Response { static_response( "text/javascript; charset=utf-8", include_str!(concat!(env!("OUT_DIR"), "/webui/app.js")), ) } pub async fn webui_styles() -> Response { static_response( "text/css; charset=utf-8", include_str!(concat!(env!("OUT_DIR"), "/webui/styles.css")), ) } pub async fn webui_theme_init() -> Response { static_response( "text/javascript; charset=utf-8", include_str!(concat!(env!("OUT_DIR"), "/webui/theme-init.js")), ) } const EMBEDDED_FONTS: &[(&str, &[u8])] = &[ ( "space-grotesk-500.woff2", include_bytes!(concat!(env!("OUT_DIR"), "/webui/fonts/space-grotesk-500.woff2")), ), ( "space-grotesk-700.woff2", include_bytes!(concat!(env!("OUT_DIR"), "/webui/fonts/space-grotesk-700.woff2")), ), ( "jetbrains-mono-400.woff2", include_bytes!(concat!(env!("OUT_DIR"), "/webui/fonts/jetbrains-mono-400.woff2")), ), ( "jetbrains-mono-700.woff2", include_bytes!(concat!(env!("OUT_DIR"), "/webui/fonts/jetbrains-mono-700.woff2")), ), ]; pub async fn webui_font(Path(name): Path) -> Response { let bytes = EMBEDDED_FONTS .iter() .find(|(font_name, _)| *font_name == name) .map(|(_, bytes)| *bytes); let Some(bytes) = bytes else { return StatusCode::NOT_FOUND.into_response(); }; Response::builder() .header(header::CONTENT_TYPE, "font/woff2") .header(header::CACHE_CONTROL, "public, max-age=31536000, immutable") .header("X-Content-Type-Options", "nosniff") .body(Body::from(bytes)) .expect("valid font response") } #[derive(Debug)] pub struct ApiError { status: StatusCode, message: String, } impl ApiError { fn bad_request(message: impl Into) -> Self { Self { status: StatusCode::BAD_REQUEST, message: message.into(), } } fn not_found(message: impl Into) -> Self { Self { status: StatusCode::NOT_FOUND, message: message.into(), } } fn payload_too_large(message: impl Into) -> Self { Self { status: StatusCode::PAYLOAD_TOO_LARGE, message: message.into(), } } fn conflict(message: impl Into) -> Self { Self { status: StatusCode::CONFLICT, message: message.into(), } } fn service_unavailable(message: impl Into) -> Self { Self { status: StatusCode::SERVICE_UNAVAILABLE, message: message.into(), } } fn internal(error: impl std::fmt::Display) -> Self { tracing::error!(error = %error, "WebUI API request failed"); Self { status: StatusCode::INTERNAL_SERVER_ERROR, message: error.to_string(), } } fn internal_with_message(error: impl std::fmt::Display, message: impl Into) -> Self { tracing::error!(error = %error, "WebUI API request failed"); Self { status: StatusCode::INTERNAL_SERVER_ERROR, message: message.into(), } } } pub async fn upload_file( State(state): State>, Path(client_id): Path, mut multipart: Multipart, ) -> Result<(StatusCode, Json), ApiError> { if !valid_client_id(&client_id) { return Err(ApiError::bad_request("invalid client id")); } if !state.uploads.enabled() { return Err(ApiError::bad_request("file transfer is disabled")); } let mut field = multipart .next_field() .await .map_err(|error| ApiError::bad_request(format!("invalid multipart body: {error}")))? .ok_or_else(|| ApiError::bad_request("file field is required"))?; if field.name() != Some("file") { return Err(ApiError::bad_request("expected multipart field named file")); } let original_name = field.file_name().unwrap_or("attachment").to_string(); let safe_name = crate::gateway::uploads::UploadRegistry::safe_file_name(&original_name); let (temporary_path, final_path, upload_id) = state .uploads .allocate_path(&client_id, &safe_name) .await .map_err(ApiError::internal)?; let mut temporary = TemporaryUpload { path: temporary_path.clone(), committed: false, }; let mut output = tokio::fs::File::create(&temporary_path) .await .map_err(ApiError::internal)?; let mut size = 0_u64; while let Some(chunk) = field .chunk() .await .map_err(|error| ApiError::bad_request(format!("failed to read upload: {error}")))? { size = size.saturating_add(chunk.len() as u64); if size > state.uploads.max_file_bytes() { drop(output); let _ = tokio::fs::remove_file(&temporary_path).await; return Err(ApiError::payload_too_large(format!( "file exceeds {} bytes", state.uploads.max_file_bytes() ))); } if let Err(error) = output.write_all(&chunk).await { drop(output); let _ = tokio::fs::remove_file(&temporary_path).await; return Err(ApiError::internal(error)); } } if size == 0 { drop(output); let _ = tokio::fs::remove_file(&temporary_path).await; return Err(ApiError::bad_request("empty files are not supported")); } output.sync_all().await.map_err(ApiError::internal)?; drop(output); tokio::fs::rename(&temporary_path, &final_path) .await .map_err(ApiError::internal)?; temporary.committed = true; let mime_type = mime_guess::from_path(&safe_name) .first_or_octet_stream() .to_string(); let media_type = media_type_for_mime(&mime_type).to_string(); let descriptor = match state .uploads .register( upload_id, client_id, final_path.clone(), crate::protocol::UploadDescriptor { upload_id: String::new(), name: safe_name, media_type, mime_type, size, expires_at: 0, }, ) .await { Ok(descriptor) => descriptor, Err(error) => { let _ = tokio::fs::remove_file(final_path).await; return Err(ApiError::bad_request(error.to_string())); } }; Ok((StatusCode::CREATED, Json(descriptor))) } #[derive(Debug, Default, Deserialize)] pub struct AttachmentQuery { disposition: Option, } pub async fn download_attachment( State(state): State>, Path((client_id, session_id, message_id, index)): Path<(String, String, String, usize)>, Query(query): Query, ) -> Result { if !valid_client_id(&client_id) { return Err(ApiError::not_found("attachment not found")); } let unified = crate::session::UnifiedSessionId::parse(&session_id) .filter(|id| id.channel == "cli_chat" && id.chat_id == client_id) .ok_or_else(|| ApiError::not_found("attachment not found"))?; let message = state .storage .get_message(&unified.to_string(), &message_id) .await .map_err(ApiError::internal)? .ok_or_else(|| ApiError::not_found("attachment not found"))?; let media_refs = message .media_refs .as_deref() .and_then(|value| serde_json::from_str::>(value).ok()) .unwrap_or_default(); let media_ref = media_refs .get(index) .ok_or_else(|| ApiError::not_found("attachment not found"))?; let path = FsPath::new(&media_ref.path); let metadata = tokio::fs::metadata(path) .await .map_err(|_| ApiError::not_found("the original file was moved or deleted"))?; if !metadata.is_file() { return Err(ApiError::not_found( "the original file was moved or deleted", )); } let file = tokio::fs::File::open(path) .await .map_err(|_| ApiError::not_found("the original file was moved or deleted"))?; let name = path .file_name() .and_then(|value| value.to_str()) .map(crate::gateway::uploads::UploadRegistry::safe_file_name) .unwrap_or_else(|| "attachment".to_string()); let mime_type = mime_guess::from_path(&name) .first_or_octet_stream() .to_string(); let inline = query.disposition.as_deref() == Some("inline") && inline_mime_allowed(&mime_type); let disposition = if inline { "inline" } else { "attachment" }; let encoded_name = encode_header_filename(&name); Response::builder() .status(StatusCode::OK) .header(header::CONTENT_TYPE, mime_type) .header(header::CONTENT_LENGTH, metadata.len()) .header( header::CONTENT_DISPOSITION, format!("{disposition}; filename*=UTF-8''{encoded_name}"), ) .header("X-Content-Type-Options", "nosniff") .header(header::CACHE_CONTROL, "private, no-store") .body(Body::from_stream(ReaderStream::new(file))) .map_err(ApiError::internal) } fn valid_client_id(value: &str) -> bool { !value.is_empty() && value.len() <= 64 && value .bytes() .all(|byte| byte.is_ascii_alphanumeric() || byte == b'-' || byte == b'_') } fn media_type_for_mime(mime: &str) -> &'static str { if mime.starts_with("image/") { "image" } else if mime.starts_with("audio/") { "audio" } else if mime.starts_with("video/") { "video" } else { "file" } } fn inline_mime_allowed(mime: &str) -> bool { matches!( mime, "image/png" | "image/jpeg" | "image/gif" | "image/webp" | "image/bmp" | "image/avif" | "image/x-icon" | "audio/mpeg" | "audio/ogg" | "audio/wav" | "video/mp4" | "video/webm" ) } fn encode_header_filename(value: &str) -> String { value .as_bytes() .iter() .map(|byte| match *byte { b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'.' | b'_' | b'-' => { (*byte as char).to_string() } other => format!("%{other:02X}"), }) .collect() } impl IntoResponse for ApiError { fn into_response(self) -> Response { (self.status, Json(json!({ "error": self.message }))).into_response() } } #[derive(Serialize)] pub struct ConfigResponse { config: Value, path: String, restart_required: bool, } #[derive(Serialize)] pub struct ReloadResponse { generation: u64, message: String, } pub async fn reload_config( State(state): State>, ) -> Result, ApiError> { let accepted = state.reload.request().await.map_err(|error| match error { super::reload::ReloadError::AlreadyPending => ApiError::conflict(error.to_string()), super::reload::ReloadError::ShuttingDown | super::reload::ReloadError::PreparationFailed(_) => { ApiError::service_unavailable(error.to_string()) } super::reload::ReloadError::InvalidConfig(_) | super::reload::ReloadError::ImmutableField(_) => ApiError::bad_request(error.to_string()), })?; Ok(Json(ReloadResponse { generation: accepted.generation, message: accepted.message, })) } pub async fn reload_status( State(state): State>, ) -> Json { Json(state.reload.status()) } pub async fn get_config( State(state): State>, ) -> Result, ApiError> { let raw = tokio::fs::read_to_string(&state.config_path) .await .map_err(ApiError::internal)?; let mut value: Value = serde_json::from_str(&raw).map_err(ApiError::internal)?; redact_secrets(&mut value); Ok(Json(ConfigResponse { config: value, path: state.config_path.display().to_string(), restart_required: false, })) } pub async fn put_config( State(state): State>, Json(mut incoming): Json, ) -> Result, ApiError> { if incoming.get("config").is_some() { incoming = incoming .get_mut("config") .map(Value::take) .ok_or_else(|| ApiError::bad_request("config is required"))?; } let encoded_size = serde_json::to_vec(&incoming) .map_err(|error| ApiError::bad_request(error.to_string()))? .len(); if encoded_size > MAX_CONFIG_BYTES { return Err(ApiError::bad_request("config exceeds 1 MiB")); } let current_raw = tokio::fs::read_to_string(&state.config_path) .await .map_err(ApiError::internal)?; let current: Value = serde_json::from_str(¤t_raw).map_err(ApiError::internal)?; restore_redacted_secrets(&mut incoming, ¤t); let parsed: Config = serde_json::from_value(incoming.clone()) .map_err(|error| ApiError::bad_request(format!("invalid config: {error}")))?; parsed .get_provider_config("default") .map_err(|error| ApiError::bad_request(format!("invalid default agent: {error}")))?; let pretty = serde_json::to_string_pretty(&incoming).map_err(ApiError::internal)? + "\n"; atomic_write(&state.config_path, pretty.as_bytes()).await?; tracing::info!(path = %state.config_path.display(), "Configuration updated from WebUI; reload or restart required"); let mut response = incoming; redact_secrets(&mut response); Ok(Json(ConfigResponse { config: response, path: state.config_path.display().to_string(), restart_required: true, })) } fn is_secret_key(key: &str) -> bool { let key = key.to_ascii_lowercase(); key.contains("api_key") || key.contains("secret") || key.contains("password") || key.ends_with("token") || key.ends_with("_token") || key == "authorization" } fn redact_secrets(value: &mut Value) { match value { Value::Object(map) => { for (key, value) in map { if is_secret_key(key) && value.is_string() { *value = Value::String(REDACTED.to_string()); } else { redact_secrets(value); } } } Value::Array(values) => values.iter_mut().for_each(redact_secrets), _ => {} } } fn restore_redacted_secrets(incoming: &mut Value, current: &Value) { match (incoming, current) { (Value::Object(incoming), Value::Object(current)) => { for (key, value) in incoming { if is_secret_key(key) && value.as_str() == Some(REDACTED) { if let Some(original) = current.get(key) { *value = original.clone(); } } else if let Some(original) = current.get(key) { restore_redacted_secrets(value, original); } } } (Value::Array(incoming), Value::Array(current)) => { for (value, original) in incoming.iter_mut().zip(current) { restore_redacted_secrets(value, original); } } _ => {} } } async fn atomic_write(path: &FsPath, content: &[u8]) -> Result<(), ApiError> { let parent = path.parent().unwrap_or_else(|| FsPath::new(".")); tokio::fs::create_dir_all(parent) .await .map_err(ApiError::internal)?; let temp = parent.join(format!(".picobot-webui-{}.tmp", crate::util::short_id())); tokio::fs::write(&temp, content) .await .map_err(ApiError::internal)?; if let Err(error) = tokio::fs::rename(&temp, path).await { let _ = tokio::fs::remove_file(&temp).await; return Err(ApiError::internal(error)); } Ok(()) } #[derive(Serialize)] pub struct ProfileResponse { name: String, content: String, path: String, } #[derive(Deserialize)] pub struct ProfileUpdate { content: String, } fn profile_path(name: &str) -> Result { let file = match name.to_ascii_lowercase().as_str() { "user" | "user.md" => "USER.md", "agents" | "agents.md" => "AGENTS.md", _ => return Err(ApiError::not_found("profile must be USER.md or AGENTS.md")), }; Ok(crate::config::get_user_config_dir().join(file)) } pub async fn get_profile(Path(name): Path) -> Result, ApiError> { let path = profile_path(&name)?; let content = tokio::fs::read_to_string(&path) .await .map_err(ApiError::internal)?; Ok(Json(ProfileResponse { name, content, path: path.display().to_string(), })) } pub async fn put_profile( Path(name): Path, Json(update): Json, ) -> Result, ApiError> { if update.content.len() > MAX_PROFILE_BYTES { return Err(ApiError::bad_request("profile exceeds 256 KiB")); } let path = profile_path(&name)?; atomic_write(&path, update.content.as_bytes()).await?; tracing::info!(profile = %name, path = %path.display(), "Assistant profile updated from WebUI"); Ok(Json(ProfileResponse { name, content: update.content, path: path.display().to_string(), })) } #[derive(Default, Deserialize)] pub struct LogsQuery { lines: Option, search: Option, } #[derive(Serialize)] pub struct LogsResponse { lines: Vec, files: Vec, } pub async fn get_logs(Query(query): Query) -> Result, ApiError> { let limit = query.lines.unwrap_or(500).clamp(1, 5000); let search = query.search.filter(|value| !value.is_empty()); let log_dir = crate::logging::get_default_log_dir(); let mut entries = tokio::fs::read_dir(&log_dir) .await .map_err(ApiError::internal)?; let mut paths = Vec::new(); while let Some(entry) = entries.next_entry().await.map_err(ApiError::internal)? { let path = entry.path(); if path.is_file() && path .file_name() .and_then(|value| value.to_str()) .is_some_and(|name| name.starts_with("picobot.log")) { paths.push(path); } } paths.sort(); paths = paths.into_iter().rev().take(7).collect(); paths.sort(); let files = paths .iter() .filter_map(|path| path.file_name()?.to_str().map(str::to_string)) .collect(); let mut output = VecDeque::with_capacity(limit); for path in paths { let content = read_file_tail(&path, 2 * 1024 * 1024).await?; for line in content.lines() { if search.as_ref().is_some_and(|needle| { !line .to_ascii_lowercase() .contains(&needle.to_ascii_lowercase()) }) { continue; } if output.len() == limit { output.pop_front(); } output.push_back(line.to_string()); } } Ok(Json(LogsResponse { lines: output.into(), files, })) } async fn read_file_tail(path: &FsPath, max_bytes: u64) -> Result { let mut file = tokio::fs::File::open(path) .await .map_err(ApiError::internal)?; let len = file.metadata().await.map_err(ApiError::internal)?.len(); if len > max_bytes { file.seek(std::io::SeekFrom::Start(len - max_bytes)) .await .map_err(ApiError::internal)?; } let mut bytes = Vec::with_capacity(len.min(max_bytes) as usize); file.read_to_end(&mut bytes) .await .map_err(ApiError::internal)?; let text = String::from_utf8_lossy(&bytes).into_owned(); Ok(if len > max_bytes { text.find('\n') .map_or(text.clone(), |newline| text[newline + 1..].to_string()) } else { text }) } #[derive(Default, Deserialize)] pub struct LimitQuery { limit: Option, } fn scheduler_snapshot(jobs: &[crate::storage::ScheduledJob]) -> Value { let enabled = jobs.iter().filter(|job| job.enabled).count(); let failed_jobs = jobs .iter() .filter(|job| { matches!( job.last_status.as_deref(), Some("error" | "timeout" | "delivery_error") ) }) .count(); let next_run_at = jobs .iter() .filter(|job| job.enabled) .map(|job| job.next_run_at) .min(); json!({ "jobs": jobs.len(), "enabled": enabled, "failed_jobs": failed_jobs, "next_run_at": next_run_at, }) } fn scheduler_status_error(error: impl std::fmt::Display) -> ApiError { ApiError::internal_with_message(error, "failed to load scheduler status") } fn channel_snapshot(mut channels: Vec<(String, bool)>) -> Vec { channels.sort_by(|left, right| left.0.cmp(&right.0)); channels .into_iter() .map(|(name, running)| { json!({ "name": name, "status": if running { "connected" } else { "stopped" }, }) }) .collect() } fn sorted_providers( mut providers: Vec, ) -> Vec { providers.sort_by(|left, right| left.name.cmp(&right.name)); providers } pub async fn get_status(State(state): State>) -> Result, ApiError> { let reload = state.reload.status(); let metrics = crate::observability::metrics::global_metrics().snapshot(); let depths = state.bus().queue_depths(); let active_lanes = state .outbound_lanes .load(std::sync::atomic::Ordering::Relaxed); let ws_connections = state .ws_connections .load(std::sync::atomic::Ordering::Relaxed); let background_tasks = state.task_supervisor.running_count(); let sessions_total = state.session_manager.session_count().await; let active_turns = state.session_manager.active_turn_count().await; let mut channel_states = Vec::new(); for name in state.channel_manager.list_channel_names().await { let running = state .channel_manager .get_channel(&name) .await .is_some_and(|channel| channel.is_running()); channel_states.push((name, running)); } let channels = channel_snapshot(channel_states); let jobs = state .storage .list_scheduled_jobs() .await .map_err(scheduler_status_error)?; let scheduler = scheduler_snapshot(&jobs); let mcp = crate::mcp::get_mcp_status() .into_iter() .map(|status| { json!({ "name": status.name, "connected": status.connected, "tools": status.tools.len(), }) }) .collect::>(); Ok(Json(json!({ "generation": reload.generation, "version": env!("CARGO_PKG_VERSION"), "uptime_secs": crate::gateway::process_uptime_secs(), "phase": reload.phase, "ws_connections": ws_connections, "background_tasks": background_tasks, "sessions": { "total": sessions_total, "active_turns": active_turns, }, "metrics": { "tokens_in": metrics.tokens_in, "tokens_out": metrics.tokens_out, "cost": metrics.cost, "tool_calls": metrics.tool_calls, "turns": metrics.turns, "turn_latency_p95_ms": metrics.turn_latency_p95_ms, }, "bus": { "inbound": { "depth": depths.inbound_depth, "cap": depths.inbound_cap }, "outbound": { "depth": depths.outbound_depth, "cap": depths.outbound_cap }, "control": { "depth": depths.control_depth, "cap": depths.control_cap }, "active_lanes": active_lanes, }, "providers": sorted_providers(metrics.providers), "channels": channels, "scheduler": scheduler, "mcp": mcp, }))) } pub async fn get_tasks( State(state): State>, Query(query): Query, ) -> Result, ApiError> { let limit = query.limit.unwrap_or(100).clamp(1, 500); let tasks = state .storage .list_recent_background_tasks(limit) .await .map_err(ApiError::internal)?; Ok(Json(json!({ "tasks": tasks }))) } pub async fn get_jobs(State(state): State>) -> Result, ApiError> { let jobs = state .storage .list_scheduled_jobs() .await .map_err(ApiError::internal)?; Ok(Json(json!({ "jobs": jobs }))) } pub async fn get_job_runs( State(state): State>, Path(id): Path, Query(query): Query, ) -> Result, ApiError> { state .storage .get_scheduled_job(&id) .await .map_err(|error| ApiError::not_found(error.to_string()))?; let limit = query.limit.unwrap_or(50).clamp(1, 500); let runs = state .storage .list_scheduled_job_runs(&id, limit) .await .map_err(ApiError::internal)?; Ok(Json(json!({ "runs": runs }))) } #[derive(Default, Deserialize)] pub struct MemoriesQuery { query: Option, category: Option, session_id: Option, limit: Option, } pub async fn get_memories( State(state): State>, Query(query): Query, ) -> Result, ApiError> { let category = query .category .as_deref() .filter(|value| !value.is_empty()) .map(|value| { MemoryCategory::parse(value) .ok_or_else(|| ApiError::bad_request("category must be knowledge or timeline")) }) .transpose()?; let limit = query.limit.unwrap_or(100).clamp(1, 500); let session_id = query .session_id .as_deref() .filter(|value| !value.is_empty()); let memories = if let Some(search) = query .query .as_deref() .filter(|value| !value.trim().is_empty()) { state .storage .search_memories(search, category.as_ref(), session_id, limit) .await } else { state .storage .list_memories(category.as_ref(), session_id, limit) .await } .map_err(ApiError::internal)?; Ok(Json(json!({ "memories": memories }))) } #[cfg(test)] mod tests { use super::*; fn scheduled_job( id: &str, enabled: bool, next_run_at: i64, last_status: Option<&str>, ) -> crate::storage::ScheduledJob { crate::storage::ScheduledJob { id: id.to_string(), name: id.to_string(), schedule: crate::scheduler::Schedule::Every { every_ms: 60_000 }, prompt: String::new(), channel: "cli_chat".to_string(), chat_id: "test".to_string(), model: None, job_kind: crate::storage::JobKind::Task, delivery_policy: crate::storage::DeliveryPolicy::Never, enabled, delete_after_run: false, next_run_at, last_run_at: None, last_status: last_status.map(str::to_string), last_error: None, created_at: 0, updated_at: 0, } } #[test] fn scheduler_snapshot_classifies_failures_and_next_enabled_run() { let jobs = vec![ scheduled_job("healthy", true, 300, Some("ok")), scheduled_job("error", true, 200, Some("error")), scheduled_job("timeout", false, 100, Some("timeout")), scheduled_job("delivery", true, 400, Some("delivery_error")), scheduled_job("other", false, 50, Some("cancelled")), ]; assert_eq!( scheduler_snapshot(&jobs), json!({ "jobs": 5, "enabled": 3, "failed_jobs": 3, "next_run_at": 200, }) ); } #[test] fn scheduler_snapshot_has_null_next_run_without_enabled_jobs() { let jobs = vec![scheduled_job("disabled", false, 100, None)]; assert_eq!( scheduler_snapshot(&jobs), json!({ "jobs": 1, "enabled": 0, "failed_jobs": 0, "next_run_at": null, }) ); } #[test] fn channel_snapshot_is_sorted_and_maps_running_state() { let channels = channel_snapshot(vec![ ("zeta".to_string(), false), ("alpha".to_string(), true), ]); assert_eq!( channels, vec![ json!({ "name": "alpha", "status": "connected" }), json!({ "name": "zeta", "status": "stopped" }), ] ); } #[test] fn provider_snapshot_is_sorted_by_name() { let provider = |name: &str| crate::observability::metrics::ProviderSnapshot { name: name.to_string(), model: String::new(), status: "ok".to_string(), latency_ms: 0, latencies: Vec::new(), tokens_in: 0, tokens_out: 0, cost: 0.0, }; let providers = sorted_providers(vec![provider("zeta"), provider("alpha")]); assert_eq!( providers .iter() .map(|provider| provider.name.as_str()) .collect::>(), vec!["alpha", "zeta"] ); } #[test] fn scheduler_status_error_hides_internal_details() { let error = scheduler_status_error("database contained private payload"); assert_eq!(error.status, StatusCode::INTERNAL_SERVER_ERROR); assert_eq!(error.message, "failed to load scheduler status"); assert!(!error.message.contains("private payload")); } #[test] fn secrets_are_redacted_and_restored() { let current = json!({"api_key":"real", "nested":{"access_token":"token"}, "safe":"yes"}); let mut shown = current.clone(); redact_secrets(&mut shown); assert_eq!(shown["api_key"], REDACTED); assert_eq!(shown["nested"]["access_token"], REDACTED); restore_redacted_secrets(&mut shown, ¤t); assert_eq!(shown, current); } #[test] fn profile_names_are_allowlisted() { assert!(profile_path("USER.md").unwrap().ends_with("USER.md")); assert!(profile_path("../config.json").is_err()); } #[tokio::test] async fn embedded_webui_has_security_headers() { let response = webui_index().await; assert_eq!( response.headers().get("X-Content-Type-Options").unwrap(), "nosniff" ); assert!(response.headers().contains_key("Content-Security-Policy")); } #[test] fn attachment_headers_are_safely_encoded_and_inline_is_allowlisted() { assert_eq!( encode_header_filename("报告 1.pdf"), "%E6%8A%A5%E5%91%8A%201.pdf" ); assert!(inline_mime_allowed("image/png")); assert!(inline_mime_allowed("image/bmp")); assert!(!inline_mime_allowed("image/svg+xml")); assert!(!inline_mime_allowed("text/html")); } }