From d81912da5641a40b77e90900a182dc335a1b94bf Mon Sep 17 00:00:00 2001 From: Alex Oliveira <4482374+aroff@users.noreply.github.com> Date: Wed, 13 May 2026 18:07:03 +0000 Subject: [PATCH] chore(develop): 55-feat-pending-inspect-api-and-cli-human-queue-ailoop-queue --- ailoop-cli/src/cli/mod.rs | 2 + ailoop-cli/src/cli/queue.rs | 18 ++ ailoop-cli/src/cli/queue_handlers.rs | 69 +++++ ailoop-cli/src/cli/task_handlers.rs | 7 +- ailoop-cli/src/main.rs | 6 + ailoop-cli/tests/queue_cli_test.rs | 173 ++++++++++++ ailoop-core/src/client/mod.rs | 1 + ailoop-core/src/client/pending_client.rs | 47 ++++ ailoop-core/src/lib.rs | 2 + ailoop-py/uv.lock | 2 +- ailoop-server/src/server/api.rs | 68 ++++- ailoop-server/src/server/core.rs | 46 +++- ailoop-server/src/server/providers/mod.rs | 4 +- .../src/server/providers/pending_prompt.rs | 43 ++- ailoop-server/tests/pending_api_test.rs | 255 ++++++++++++++++++ .../tests/provider_pending_prompt_test.rs | 26 +- 16 files changed, 756 insertions(+), 13 deletions(-) create mode 100644 ailoop-cli/src/cli/queue.rs create mode 100644 ailoop-cli/src/cli/queue_handlers.rs create mode 100644 ailoop-cli/tests/queue_cli_test.rs create mode 100644 ailoop-core/src/client/pending_client.rs create mode 100644 ailoop-server/tests/pending_api_test.rs diff --git a/ailoop-cli/src/cli/mod.rs b/ailoop-cli/src/cli/mod.rs index 480b662..2c4f249 100644 --- a/ailoop-cli/src/cli/mod.rs +++ b/ailoop-cli/src/cli/mod.rs @@ -6,6 +6,8 @@ pub mod handlers; pub mod message_converter; pub mod provider; pub mod provider_handlers; +pub mod queue; +pub mod queue_handlers; pub mod task; pub mod task_handlers; pub mod terminal_input; diff --git a/ailoop-cli/src/cli/queue.rs b/ailoop-cli/src/cli/queue.rs new file mode 100644 index 0000000..6996353 --- /dev/null +++ b/ailoop-cli/src/cli/queue.rs @@ -0,0 +1,18 @@ +//! CLI arguments for the `ailoop queue` subcommand. + +use clap::Args; + +#[derive(Args)] +pub struct QueueArgs { + /// Server base URL (default: AILOOP_SERVER env or http://127.0.0.1:8080) + #[arg(long, default_value = "")] + pub server: String, + + /// Filter by channel name + #[arg(long)] + pub channel: Option, + + /// Output in JSON format + #[arg(long)] + pub json: bool, +} diff --git a/ailoop-cli/src/cli/queue_handlers.rs b/ailoop-cli/src/cli/queue_handlers.rs new file mode 100644 index 0000000..ce892f6 --- /dev/null +++ b/ailoop-cli/src/cli/queue_handlers.rs @@ -0,0 +1,69 @@ +//! Handler for the `ailoop queue` subcommand. + +use super::queue::QueueArgs; +use super::task_handlers::resolve_server_url; +use ailoop_core::PendingClient; +use anyhow::Result; + +pub async fn handle_queue_commands(args: QueueArgs) -> Result<()> { + let server_url = resolve_server_url(args.server)?; + let client = PendingClient::new(&server_url); + let response = client.list_pending(args.channel.as_deref()).await?; + + if args.json { + println!("{}", serde_json::to_string_pretty(&response)?); + return Ok(()); + } + + let filter_label = match &args.channel { + Some(ch) => format!("channel={}", ch), + None => "all channels".to_string(), + }; + + println!( + "Human queue: {} pending ({})", + response.total_count, filter_label + ); + println!(); + + if response.items.is_empty() { + println!("(no pending items)"); + return Ok(()); + } + + let terminal_width: usize = std::env::var("COLUMNS") + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(80); + + let fixed_cols_width = 2 + 2 + 4 + 2 + 10 + 2 + 10 + 2; + let title_width = terminal_width.saturating_sub(fixed_cols_width).max(10); + + println!( + "{:<2} {:<2} {:<10} {:<10} Title", + "#", "Ty", "Channel", "Msg" + ); + + for item in &response.items { + let ty_char = match item.kind.as_str() { + "decision" => "D", + "authorize" => "A", + "navigate" => "N", + _ => "?", + }; + + let channel_display: String = item.channel.chars().take(10).collect(); + + let msg_id_str = item.message_id.to_string().replace('-', ""); + let msg_display = format!("{}...", &msg_id_str[..8]); + + let title_display: String = item.label.chars().take(title_width).collect(); + + println!( + "{:<2} {:<2} {:<10} {:<10} {}", + item.position, ty_char, channel_display, msg_display, title_display + ); + } + + Ok(()) +} diff --git a/ailoop-cli/src/cli/task_handlers.rs b/ailoop-cli/src/cli/task_handlers.rs index b8eddfd..03fc440 100644 --- a/ailoop-cli/src/cli/task_handlers.rs +++ b/ailoop-cli/src/cli/task_handlers.rs @@ -381,7 +381,7 @@ async fn handle_task_blocked(channel: String, server: String, json: bool) -> Res Ok(()) } -fn resolve_server_url(server: String) -> Result { +pub fn resolve_server_url(server: String) -> Result { if server.is_empty() { get_server_url() } else { @@ -402,5 +402,10 @@ fn parse_task_state(value: &str) -> Result { } fn get_server_url() -> Result { + if let Ok(url) = std::env::var("AILOOP_SERVER") { + if !url.is_empty() { + return Ok(url); + } + } Ok("http://127.0.0.1:8080".to_string()) } diff --git a/ailoop-cli/src/main.rs b/ailoop-cli/src/main.rs index 410508c..85cfd69 100644 --- a/ailoop-cli/src/main.rs +++ b/ailoop-cli/src/main.rs @@ -230,6 +230,9 @@ enum Commands { #[command(subcommand)] command: cli::provider::ProviderCommands, }, + + /// Inspect the human prompt queue + Queue(cli::queue::QueueArgs), } #[tokio::main] @@ -316,6 +319,9 @@ async fn main() -> Result<()> { Commands::Provider { command } => { crate::cli::provider_handlers::handle_provider_commands(command).await?; } + Commands::Queue(args) => { + crate::cli::queue_handlers::handle_queue_commands(args).await?; + } } Ok(()) diff --git a/ailoop-cli/tests/queue_cli_test.rs b/ailoop-cli/tests/queue_cli_test.rs new file mode 100644 index 0000000..90fe8a4 --- /dev/null +++ b/ailoop-cli/tests/queue_cli_test.rs @@ -0,0 +1,173 @@ +//! Integration tests for `ailoop queue` CLI subcommand. + +mod common; + +use ailoop_server::AiloopServer; +use anyhow::{Context, Result}; +use std::process::Command; +use std::time::Duration; +use tokio::sync::oneshot; +use tokio::task::JoinHandle; +use tokio::time::sleep; + +const TEST_HOST: &str = "127.0.0.1"; + +async fn spawn_test_server( + host: &str, +) -> Result<(u16, oneshot::Sender<()>, JoinHandle>)> { + let port = common::find_free_port(host).context("Failed to find free port for test server")?; + let server = AiloopServer::new(host.to_string(), port, "queue-test".to_string()); + let (shutdown_tx, shutdown_rx) = oneshot::channel(); + let server_handle = tokio::spawn(async move { + server + .start_with_shutdown(async move { + let _ = shutdown_rx.await; + }) + .await + }); + wait_for_server_ready(host, port, Duration::from_secs(10)).await?; + Ok((port, shutdown_tx, server_handle)) +} + +async fn wait_for_server_ready(host: &str, port: u16, timeout: Duration) -> Result<()> { + let start = std::time::Instant::now(); + loop { + if start.elapsed() > timeout { + return Err(anyhow::anyhow!( + "Server readiness check timed out after {:?}", + timeout + )); + } + if tokio::net::TcpStream::connect(format!("{}:{}", host, port)) + .await + .is_ok() + { + break; + } + sleep(Duration::from_millis(50)).await; + } + Ok(()) +} + +async fn run_cmd(args: &[&str]) -> (bool, String, String) { + let args: Vec = args.iter().map(|s| s.to_string()).collect(); + tokio::task::spawn_blocking(move || { + let output = Command::new("cargo") + .args(["run", "--bin", "ailoop", "--"]) + .args(&args) + .env_remove("AILOOP_SERVER") + .env_remove("AILOOP_MODE") + .output() + .expect("Failed to run ailoop"); + ( + output.status.success(), + String::from_utf8_lossy(&output.stdout).trim().to_string(), + String::from_utf8_lossy(&output.stderr).trim().to_string(), + ) + }) + .await + .expect("spawn_blocking panicked") +} + +#[tokio::test] +async fn test_queue_json_output_shape() -> Result<()> { + let _port_lock = common::port_allocation_lock().context("port allocation lock")?; + let (port, shutdown_tx, server_handle) = spawn_test_server(TEST_HOST).await?; + let server_url = format!("http://{}:{}", TEST_HOST, port); + + let (ok, stdout, stderr) = run_cmd(&["queue", "--server", &server_url, "--json"]).await; + + assert!(ok, "ailoop queue --json should succeed; stderr: {}", stderr); + + let json: serde_json::Value = serde_json::from_str(&stdout) + .unwrap_or_else(|e| panic!("output should be valid JSON: {}; stdout: {}", e, stdout)); + + assert!( + json.get("items").is_some(), + "JSON must have 'items' key; got: {}", + json + ); + assert!( + json.get("total_count").is_some(), + "JSON must have 'total_count' key; got: {}", + json + ); + assert_eq!(json["total_count"], 0); + assert!(json["items"].as_array().unwrap().is_empty()); + + let _ = shutdown_tx.send(()); + let _ = server_handle.await; + Ok(()) +} + +#[tokio::test] +async fn test_queue_human_readable_header() -> Result<()> { + let _port_lock = common::port_allocation_lock().context("port allocation lock")?; + let (port, shutdown_tx, server_handle) = spawn_test_server(TEST_HOST).await?; + let server_url = format!("http://{}:{}", TEST_HOST, port); + + let (ok, stdout, stderr) = run_cmd(&["queue", "--server", &server_url]).await; + + assert!(ok, "ailoop queue should succeed; stderr: {}", stderr); + + assert!( + stdout.contains("Human queue:"), + "output must contain header 'Human queue:'; got: {}", + stdout + ); + assert!( + stdout.contains("pending"), + "output must contain 'pending'; got: {}", + stdout + ); + + let _ = shutdown_tx.send(()); + let _ = server_handle.await; + Ok(()) +} + +#[tokio::test] +async fn test_queue_ailoop_server_env_var() -> Result<()> { + let _port_lock = common::port_allocation_lock().context("port allocation lock")?; + let (port, shutdown_tx, server_handle) = spawn_test_server(TEST_HOST).await?; + let server_url = format!("http://{}:{}", TEST_HOST, port); + + // Use AILOOP_SERVER env var instead of --server flag + let args: Vec = vec![ + "run".to_string(), + "--bin".to_string(), + "ailoop".to_string(), + "--".to_string(), + "queue".to_string(), + "--json".to_string(), + ]; + let (ok, stdout, stderr) = tokio::task::spawn_blocking(move || { + let output = Command::new("cargo") + .args(&args) + .env("AILOOP_SERVER", &server_url) + .output() + .expect("Failed to run ailoop"); + ( + output.status.success(), + String::from_utf8_lossy(&output.stdout).trim().to_string(), + String::from_utf8_lossy(&output.stderr).trim().to_string(), + ) + }) + .await + .expect("spawn_blocking panicked"); + + assert!( + ok, + "ailoop queue with AILOOP_SERVER should succeed; stderr: {}", + stderr + ); + + let json: serde_json::Value = serde_json::from_str(&stdout) + .unwrap_or_else(|e| panic!("output should be valid JSON: {}; stdout: {}", e, stdout)); + assert!(json.get("items").is_some()); + assert!(json.get("total_count").is_some()); + + let _ = shutdown_tx.send(()); + let _ = server_handle.await; + Ok(()) +} diff --git a/ailoop-core/src/client/mod.rs b/ailoop-core/src/client/mod.rs index 6cd6975..f496c9d 100644 --- a/ailoop-core/src/client/mod.rs +++ b/ailoop-core/src/client/mod.rs @@ -6,6 +6,7 @@ use crate::models::{ }; use anyhow::Result; +pub mod pending_client; pub mod task_client; /// Send a structured decision and wait for the human's selection (returns the Response message). diff --git a/ailoop-core/src/client/pending_client.rs b/ailoop-core/src/client/pending_client.rs new file mode 100644 index 0000000..8907ef1 --- /dev/null +++ b/ailoop-core/src/client/pending_client.rs @@ -0,0 +1,47 @@ +//! HTTP client for the pending prompt inspection API. + +use serde::{Deserialize, Serialize}; +use uuid::Uuid; + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct PendingItemResponse { + pub message_id: Uuid, + pub kind: String, + pub channel: String, + pub position: usize, + pub label: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct PendingListResponse { + pub items: Vec, + pub total_count: usize, +} + +pub struct PendingClient { + base_url: String, + client: reqwest::Client, +} + +impl PendingClient { + pub fn new(base_url: impl Into) -> Self { + let base = base_url.into(); + let base = base.trim_end_matches('/').to_string(); + Self { + base_url: base, + client: reqwest::Client::new(), + } + } + + pub async fn list_pending(&self, channel: Option<&str>) -> anyhow::Result { + let mut url = format!("{}/api/v1/pending", self.base_url); + if let Some(ch) = channel { + url.push_str(&format!("?channel={}", ch)); + } + let resp = self.client.get(&url).send().await?; + if !resp.status().is_success() { + anyhow::bail!("Server returned {}", resp.status()); + } + Ok(resp.json::().await?) + } +} diff --git a/ailoop-core/src/lib.rs b/ailoop-core/src/lib.rs index adfd6da..d496c9d 100644 --- a/ailoop-core/src/lib.rs +++ b/ailoop-core/src/lib.rs @@ -26,3 +26,5 @@ pub mod server; pub mod services; pub mod terminal; pub mod transport; + +pub use client::pending_client::{PendingClient, PendingItemResponse, PendingListResponse}; diff --git a/ailoop-py/uv.lock b/ailoop-py/uv.lock index 8c1dacc..e62e311 100644 --- a/ailoop-py/uv.lock +++ b/ailoop-py/uv.lock @@ -4,7 +4,7 @@ requires-python = ">=3.11" [[package]] name = "ailoop-py" -version = "1.0.6" +version = "1.0.7" source = { editable = "." } dependencies = [ { name = "httpx" }, diff --git a/ailoop-server/src/server/api.rs b/ailoop-server/src/server/api.rs index 57fcfdd..6d20373 100644 --- a/ailoop-server/src/server/api.rs +++ b/ailoop-server/src/server/api.rs @@ -115,6 +115,29 @@ pub struct TasksResponse { pub total_count: usize, } +/// Query parameters for GET /api/v1/pending +#[derive(Debug, Deserialize)] +struct PendingQuery { + channel: Option, +} + +/// Per-item shape in the pending list response +#[derive(Debug, Clone, Serialize)] +pub struct PendingItemResponse { + pub message_id: Uuid, + pub kind: String, + pub channel: String, + pub position: usize, + pub label: String, +} + +/// Top-level response for GET /api/v1/pending +#[derive(Debug, Clone, Serialize)] +pub struct PendingListResponse { + pub items: Vec, + pub total_count: usize, +} + /// Query parameters for message history #[derive(Debug, Deserialize)] struct MessagesQuery { @@ -155,6 +178,7 @@ pub(crate) fn create_api_router() -> axum::Router { ) .route("/api/stats", axum::routing::get(handle_get_stats)) .route("/api/v1/health", axum::routing::get(handle_get_health)) + .route("/api/v1/pending", axum::routing::get(handle_get_pending)) .route( "/api/v1/messages", axum::routing::post(handle_post_messages), @@ -278,16 +302,58 @@ async fn handle_get_health( State(state): State, ) -> Result, ApiError> { let broadcast_stats = state.broadcast_manager.get_stats().await; + let queue_size = state + .pending_prompt_registry + .snapshot_pending(None) + .await + .len(); Ok(Json(HealthResponse { status: "healthy".to_string(), version: env!("CARGO_PKG_VERSION").to_string(), active_connections: broadcast_stats.total_viewers, - queue_size: 0, + queue_size, active_channels: broadcast_stats.active_channels, })) } +/// Handle GET /api/v1/pending +async fn handle_get_pending( + State(state): State, + Query(query): Query, +) -> Result, ApiError> { + if let Some(ref ch) = query.channel { + ailoop_core::channel::validation::validate_channel_name(ch) + .map_err(|e| ApiError::ValidationError(e.to_string()))?; + } + + let snapshots = state + .pending_prompt_registry + .snapshot_pending(query.channel.as_deref()) + .await; + + let items: Vec = snapshots + .into_iter() + .map(|s| { + let kind = match s.prompt_type { + crate::server::providers::PromptType::Decision => "decision", + crate::server::providers::PromptType::Authorization => "authorize", + crate::server::providers::PromptType::Navigation => "navigate", + }; + PendingItemResponse { + message_id: s.message_id, + kind: kind.to_string(), + channel: s.channel, + position: s.position + 1, + label: s.label, + } + }) + .collect(); + + let total_count = items.len(); + Ok(Json(PendingListResponse { items, total_count })) +} + /// Handle POST /api/v1/messages async fn handle_post_messages( State(state): State, diff --git a/ailoop-server/src/server/core.rs b/ailoop-server/src/server/core.rs index 62bfa29..4ec0f41 100644 --- a/ailoop-server/src/server/core.rs +++ b/ailoop-server/src/server/core.rs @@ -360,7 +360,13 @@ impl AiloopServer { let (resolved_id, resolved_label, resolved_index, response_type) = loop { let (rx, completer) = pending_registry - .register(message.id, reply_to_id.clone(), PromptType::Decision) + .register( + message.id, + reply_to_id.clone(), + PromptType::Decision, + message.channel.clone(), + strip_markdown(&summary), + ) .await; enum Outcome { @@ -550,7 +556,13 @@ impl AiloopServer { .await; let (rx, completer) = pending_registry - .register(message.id, reply_to_id, PromptType::Authorization) + .register( + message.id, + reply_to_id, + PromptType::Authorization, + message.channel.clone(), + action.clone(), + ) .await; let timeout_duration = resolve_effective_timeout(timeout_secs, config); @@ -714,7 +726,13 @@ impl AiloopServer { .await; let (rx, completer) = pending_registry - .register(message.id, reply_to_id, PromptType::Navigation) + .register( + message.id, + reply_to_id, + PromptType::Navigation, + message.channel.clone(), + url.clone(), + ) .await; let timeout_duration = resolve_effective_timeout(0, config); @@ -1041,6 +1059,28 @@ impl Drop for RawModeGuard { } } +/// Strip common Markdown syntax to produce a plain-text label for display. +fn strip_markdown(input: &str) -> String { + let mut result = input.to_string(); + // Remove bold/italic markers + result = result.replace("**", " "); + result = result.replace('*', " "); + result = result.replace('_', " "); + result = result.replace('`', " "); + // Remove heading markers at start of words + let mut stripped = String::new(); + for word in result.split_whitespace() { + let trimmed = word.trim_start_matches('#'); + if !trimmed.is_empty() { + if !stripped.is_empty() { + stripped.push(' '); + } + stripped.push_str(trimmed); + } + } + stripped +} + /// Axum handler: root GET — WebSocket upgrade or web UI fallback. async fn root_handler( State(state): State, diff --git a/ailoop-server/src/server/providers/mod.rs b/ailoop-server/src/server/providers/mod.rs index 87d2c60..4782fd7 100644 --- a/ailoop-server/src/server/providers/mod.rs +++ b/ailoop-server/src/server/providers/mod.rs @@ -16,8 +16,8 @@ mod sink; mod telegram; pub use pending_prompt::{ - resolve_effective_timeout, PendingPromptCompleter, PendingPromptRegistry, PromptType, - RecvTimeoutError, DEFAULT_PROMPT_TIMEOUT_SECS, + resolve_effective_timeout, PendingPromptCompleter, PendingPromptRegistry, PendingSnapshot, + PromptType, RecvTimeoutError, DEFAULT_PROMPT_TIMEOUT_SECS, }; pub use reply_source::{ProviderReply, ReplySource}; pub use sink::NotificationSink; diff --git a/ailoop-server/src/server/providers/pending_prompt.rs b/ailoop-server/src/server/providers/pending_prompt.rs index 4e25d63..e736000 100644 --- a/ailoop-server/src/server/providers/pending_prompt.rs +++ b/ailoop-server/src/server/providers/pending_prompt.rs @@ -1,6 +1,7 @@ //! Pending prompt registry: match provider replies to waiting prompts use ailoop_core::models::{Configuration, MessageContent, ResponseType}; +use serde::Serialize; use std::collections::VecDeque; use std::sync::Arc; use tokio::sync::{oneshot, RwLock}; @@ -11,7 +12,7 @@ use uuid::Uuid; pub const DEFAULT_PROMPT_TIMEOUT_SECS: u64 = 300; /// Type of interactive prompt -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] pub enum PromptType { Authorization, Navigation, @@ -24,10 +25,24 @@ struct PendingEntry { entry_id: Uuid, message_id: Uuid, reply_to_message_id: Option, + prompt_type: PromptType, + channel: String, + label: String, _created_at: std::time::Instant, tx: oneshot::Sender, } +/// Read-only clone of a pending entry for introspection. +#[derive(Debug, Clone, Serialize)] +pub struct PendingSnapshot { + pub entry_id: Uuid, + pub message_id: Uuid, + pub prompt_type: PromptType, + pub channel: String, + pub label: String, + pub position: usize, +} + /// Completer for a single registered prompt (e.g. terminal response wins). #[derive(Clone)] pub struct PendingPromptCompleter { @@ -66,7 +81,9 @@ impl PendingPromptRegistry { &self, message_id: Uuid, reply_to_message_id: Option, - _prompt_type: PromptType, + prompt_type: PromptType, + channel: String, + label: String, ) -> (oneshot::Receiver, PendingPromptCompleter) { let entry_id = Uuid::new_v4(); let (tx, rx) = oneshot::channel(); @@ -74,6 +91,9 @@ impl PendingPromptRegistry { entry_id, message_id, reply_to_message_id, + prompt_type, + channel, + label, _created_at: std::time::Instant::now(), tx, }; @@ -84,6 +104,25 @@ impl PendingPromptRegistry { }; (rx, completer) } + + /// Return a read-only snapshot of all pending entries, optionally filtered by channel. + /// Acquires a read lock; does not modify the deque or any oneshot channel. + pub async fn snapshot_pending(&self, channel_filter: Option<&str>) -> Vec { + let guard = self.inner.read().await; + guard + .iter() + .enumerate() + .filter(|(_, e)| channel_filter.is_none_or(|ch| e.channel == ch)) + .map(|(idx, e)| PendingSnapshot { + entry_id: e.entry_id, + message_id: e.message_id, + prompt_type: e.prompt_type, + channel: e.channel.clone(), + label: e.label.clone(), + position: idx, + }) + .collect() + } } impl PendingPromptRegistry { diff --git a/ailoop-server/tests/pending_api_test.rs b/ailoop-server/tests/pending_api_test.rs new file mode 100644 index 0000000..7bda93f --- /dev/null +++ b/ailoop-server/tests/pending_api_test.rs @@ -0,0 +1,255 @@ +//! Integration tests for GET /api/v1/pending + +use ailoop_server::server::providers::PromptType; +use ailoop_server::{router, AiloopAppState, ServeConfig}; +use axum::{ + body::Body, + http::{Request, StatusCode}, +}; +use std::sync::Arc; +use tower::ServiceExt; +use uuid::Uuid; + +fn default_config() -> ServeConfig { + ServeConfig { + host: "127.0.0.1".to_string(), + port: 3000, + default_channel: "default".to_string(), + base_path: None, + web: false, + auth: None, + cors: None, + } +} + +#[tokio::test] +async fn pending_empty_queue_returns_zero() { + let state = Arc::new(AiloopAppState::new("default")); + let r: axum::Router = router(Arc::clone(&state), &default_config()).unwrap(); + + let resp = r + .oneshot( + Request::builder() + .uri("/api/v1/pending") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + + assert_eq!(resp.status(), StatusCode::OK); + + let body = axum::body::to_bytes(resp.into_body(), usize::MAX) + .await + .unwrap(); + let json: serde_json::Value = serde_json::from_slice(&body).unwrap(); + + assert_eq!(json["total_count"], 0); + assert!(json["items"].as_array().unwrap().is_empty()); +} + +#[tokio::test] +async fn pending_one_entry_returns_count_one() { + let state = Arc::new(AiloopAppState::new("default")); + + // Register a pending entry directly + let (_rx, _completer) = state + .pending_prompt_registry + .register( + Uuid::new_v4(), + None, + PromptType::Decision, + "ops".to_string(), + "Deploy to production?".to_string(), + ) + .await; + + let r: axum::Router = router(Arc::clone(&state), &default_config()).unwrap(); + + let resp = r + .oneshot( + Request::builder() + .uri("/api/v1/pending") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + + assert_eq!(resp.status(), StatusCode::OK); + + let body = axum::body::to_bytes(resp.into_body(), usize::MAX) + .await + .unwrap(); + let json: serde_json::Value = serde_json::from_slice(&body).unwrap(); + + assert_eq!(json["total_count"], 1); + let items = json["items"].as_array().unwrap(); + assert_eq!(items.len(), 1); + assert_eq!(items[0]["kind"], "decision"); + assert_eq!(items[0]["channel"], "ops"); + assert_eq!(items[0]["position"], 1); + assert_eq!(items[0]["label"], "Deploy to production?"); +} + +#[tokio::test] +async fn pending_channel_filter_works() { + let state = Arc::new(AiloopAppState::new("default")); + + let (_rx1, _c1) = state + .pending_prompt_registry + .register( + Uuid::new_v4(), + None, + PromptType::Decision, + "ops".to_string(), + "Ops decision".to_string(), + ) + .await; + + let (_rx2, _c2) = state + .pending_prompt_registry + .register( + Uuid::new_v4(), + None, + PromptType::Authorization, + "public".to_string(), + "Public action".to_string(), + ) + .await; + + let r: axum::Router = router(Arc::clone(&state), &default_config()).unwrap(); + + // Filter to ops channel only + let resp = r + .clone() + .oneshot( + Request::builder() + .uri("/api/v1/pending?channel=ops") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + + assert_eq!(resp.status(), StatusCode::OK); + let body = axum::body::to_bytes(resp.into_body(), usize::MAX) + .await + .unwrap(); + let json: serde_json::Value = serde_json::from_slice(&body).unwrap(); + assert_eq!(json["total_count"], 1); + assert_eq!(json["items"][0]["channel"], "ops"); + + // No filter — both entries + let resp = r + .oneshot( + Request::builder() + .uri("/api/v1/pending") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + + assert_eq!(resp.status(), StatusCode::OK); + let body = axum::body::to_bytes(resp.into_body(), usize::MAX) + .await + .unwrap(); + let json: serde_json::Value = serde_json::from_slice(&body).unwrap(); + assert_eq!(json["total_count"], 2); +} + +#[tokio::test] +async fn pending_completion_removes_entry() { + let state = Arc::new(AiloopAppState::new("default")); + + let (_rx, completer) = state + .pending_prompt_registry + .register( + Uuid::new_v4(), + None, + PromptType::Authorization, + "ops".to_string(), + "Deploy action".to_string(), + ) + .await; + + // Complete the entry + completer + .complete(ailoop_core::models::MessageContent::Response { + answer: Some("yes".to_string()), + response_type: ailoop_core::models::ResponseType::AuthorizationApproved, + }) + .await; + + let r: axum::Router = router(Arc::clone(&state), &default_config()).unwrap(); + + let resp = r + .oneshot( + Request::builder() + .uri("/api/v1/pending") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + + let body = axum::body::to_bytes(resp.into_body(), usize::MAX) + .await + .unwrap(); + let json: serde_json::Value = serde_json::from_slice(&body).unwrap(); + assert_eq!(json["total_count"], 0); +} + +#[tokio::test] +async fn pending_invalid_channel_returns_400() { + let state = Arc::new(AiloopAppState::new("default")); + let r: axum::Router = router(Arc::clone(&state), &default_config()).unwrap(); + + let resp = r + .oneshot( + Request::builder() + .uri("/api/v1/pending?channel=invalid%20channel%20name!") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + + assert_eq!(resp.status(), StatusCode::BAD_REQUEST); +} + +#[tokio::test] +async fn health_queue_size_reflects_pending_registry() { + let state = Arc::new(AiloopAppState::new("default")); + + let (_rx, _completer) = state + .pending_prompt_registry + .register( + Uuid::new_v4(), + None, + PromptType::Decision, + "public".to_string(), + "Test decision".to_string(), + ) + .await; + + let r: axum::Router = router(Arc::clone(&state), &default_config()).unwrap(); + + let resp = r + .oneshot( + Request::builder() + .uri("/api/v1/health") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + + assert_eq!(resp.status(), StatusCode::OK); + let body = axum::body::to_bytes(resp.into_body(), usize::MAX) + .await + .unwrap(); + let json: serde_json::Value = serde_json::from_slice(&body).unwrap(); + assert_eq!(json["queue_size"], 1); +} diff --git a/ailoop-server/tests/provider_pending_prompt_test.rs b/ailoop-server/tests/provider_pending_prompt_test.rs index a198d2e..02b0e6b 100644 --- a/ailoop-server/tests/provider_pending_prompt_test.rs +++ b/ailoop-server/tests/provider_pending_prompt_test.rs @@ -16,7 +16,13 @@ async fn test_default_prompt_timeout_constant() { async fn test_register_returns_two_tuple() { let registry = PendingPromptRegistry::new(); let (rx, _completer) = registry - .register(Uuid::new_v4(), None, PromptType::Decision) + .register( + Uuid::new_v4(), + None, + PromptType::Decision, + "".into(), + "".into(), + ) .await; drop(rx); } @@ -25,7 +31,13 @@ async fn test_register_returns_two_tuple() { async fn test_submit_reply_oldest_first() { let registry = PendingPromptRegistry::new(); let (rx, _completer) = registry - .register(Uuid::new_v4(), None, PromptType::Decision) + .register( + Uuid::new_v4(), + None, + PromptType::Decision, + "".into(), + "".into(), + ) .await; let matched = registry .submit_reply(None, Some("answer".into()), ResponseType::Text) @@ -49,6 +61,8 @@ async fn test_submit_reply_by_reply_to_message_id() { Uuid::new_v4(), Some(reply_to_id.clone()), PromptType::Decision, + "".into(), + "".into(), ) .await; let matched = registry @@ -67,7 +81,13 @@ async fn test_submit_reply_by_reply_to_message_id() { async fn test_recv_timeout() { let registry = PendingPromptRegistry::new(); let (rx, _completer) = registry - .register(Uuid::new_v4(), None, PromptType::Decision) + .register( + Uuid::new_v4(), + None, + PromptType::Decision, + "".into(), + "".into(), + ) .await; let result = PendingPromptRegistry::recv_with_timeout(rx, Duration::from_millis(10)).await; assert!(matches!(result, Err(RecvTimeoutError::Timeout)));