feat: early ag-ui implementation with test along

Author: Alex Holmberg

.cargo/config.toml

@@ -1,8 +1,7 @@
 [build]
 rustflags = ["-C", "target-cpu=native"]
 
-# On macOS, the default linker (ld64) is already quite fast
-# For Linux, you could use mold: rustflags = ["-C", "link-arg=-fuse-ld=mold"]
+# Note: mold linker doesn't support macOS. Use default ld64 (already fast on Mac)
 
 [alias]
 b = "build"

Cargo.lock

@@ -88,6 +88,7 @@ rig-core = { version = "0.28", features = ["derive", "image"] }
 ag-ui-core = { path = "../ag-ui-sdk/crates/ag-ui-core" }
 ag-ui-server = { path = "../ag-ui-sdk/crates/ag-ui-server" }
 axum = { version = "0.8", features = ["ws"] }
+tower-http = { version = "0.6", features = ["cors"] }
 tokio-stream = { version = "0.1", features = ["sync"] }
 
 # AWS Bedrock dependencies (inlined bedrock module with extended thinking fixes)

Cargo.toml

@@ -204,18 +204,35 @@ fn get_system_prompt(project_path: &Path, query: Option<&str>, plan_mode: PlanMo
 /// * `host` - Host address to bind to
 /// * `port` - Port number to listen on
 pub async fn run_agent_server(
-    _project_path: &Path,
-    _provider: ProviderType,
-    _model: Option<String>,
+    project_path: &Path,
+    provider: ProviderType,
+    model: Option<String>,
     host: &str,
     port: u16,
 ) -> AgentResult<()> {
-    use crate::server::{AgUiConfig, AgUiServer};
-
-    let config = AgUiConfig::new().port(port).host(host);
+    use crate::server::{AgUiConfig, AgUiServer, ProcessorConfig};
+
+    // Configure the agent processor with provider, model, and project path
+    // Use regional model IDs (no global. prefix) for wider availability
+    let default_model = match provider {
+        // Claude 3.5 Sonnet v2 is widely available across regions
+        ProviderType::Bedrock => "anthropic.claude-3-5-sonnet-20241022-v2:0".to_string(),
+        ProviderType::Anthropic => "claude-3-5-sonnet-20241022".to_string(),
+        ProviderType::OpenAI => "gpt-4o".to_string(),
+    };
+    let processor_config = ProcessorConfig::new()
+        .with_provider(&provider.to_string())
+        .with_model(&model.unwrap_or(default_model))
+        .with_project_path(project_path);
+
+    let config = AgUiConfig::new()
+        .port(port)
+        .host(host)
+        .with_processor_config(processor_config);
     let server = AgUiServer::new(config);
 
     println!("AG-UI agent server listening on http://{}:{}", host, port);
+    println!("Project path: {}", project_path.display());
     println!("Connect frontends via SSE (/sse) or WebSocket (/ws)");
     println!("Press Ctrl+C to stop the server");
 

src/agent/mod.rs

@@ -47,6 +47,7 @@ use std::sync::Arc;
 
 use ag_ui_core::{Event, JsonValue, RunId, ThreadId};
 use axum::{routing::{get, post}, Router};
+use tower_http::cors::{Any, CorsLayer};
 use tokio::sync::{broadcast, mpsc, RwLock};
 
 pub use bridge::EventBridge;
@@ -255,12 +256,20 @@ impl AgUiServer {
             }
         }
 
+        // Configure CORS to allow requests from any origin (for development)
+        let cors = CorsLayer::new()
+            .allow_origin(Any)
+            .allow_methods(Any)
+            .allow_headers(Any);
+
         let app = Router::new()
-            .route("/", get(routes::health))
+            .route("/", get(routes::health).post(routes::post_message))
+            .route("/info", get(routes::info))
             .route("/sse", get(routes::sse_handler))
             .route("/ws", get(routes::ws_handler))
             .route("/message", post(routes::post_message))
             .route("/health", get(routes::health))
+            .layer(cors)
             .with_state(self.state);
 
         println!("AG-UI server listening on http://{}", addr);
@@ -439,4 +448,212 @@ mod tests {
             handle
         ).await;
     }
+
+    // =============================================================================
+    // E2E Integration Tests (Phase 25)
+    // =============================================================================
+
+    /// Helper to collect events until RunFinished or RunError
+    async fn collect_until_finished(rx: &mut tokio::sync::broadcast::Receiver<ag_ui_core::Event>) -> Vec<ag_ui_core::Event> {
+        use ag_ui_core::Event;
+        let mut events = Vec::new();
+        loop {
+            match tokio::time::timeout(std::time::Duration::from_secs(5), rx.recv()).await {
+                Ok(Ok(event)) => {
+                    let is_finished = matches!(&event, Event::RunFinished(_) | Event::RunError(_));
+                    events.push(event);
+                    if is_finished { break; }
+                }
+                _ => break,
+            }
+        }
+        events
+    }
+
+    /// Helper to drain events until run is finished
+    async fn drain_events_until_run_finished(rx: &mut tokio::sync::broadcast::Receiver<ag_ui_core::Event>) {
+        use ag_ui_core::Event;
+        loop {
+            match tokio::time::timeout(std::time::Duration::from_secs(30), rx.recv()).await {
+                Ok(Ok(Event::RunFinished(_))) => break,
+                Ok(Ok(Event::RunError(_))) => break,
+                Ok(Ok(_)) => continue,
+                _ => panic!("Timeout or error waiting for RunFinished"),
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn test_multi_turn_conversation() {
+        use ag_ui_core::types::{Message, RunAgentInput};
+
+        // Create state and components
+        let state = ServerState::new();
+        let msg_tx = state.message_sender();
+        let mut event_rx = state.subscribe();
+        let msg_rx = state.take_message_receiver().await.expect("Should get receiver");
+
+        // Create processor
+        let event_bridge = state.event_sender();
+        let mut processor = AgentProcessor::with_defaults(msg_rx, event_bridge);
+
+        let handle = tokio::spawn(async move {
+            processor.run().await;
+        });
+
+        let thread_id = ThreadId::random();
+
+        // Send first message
+        let input1 = RunAgentInput::new(thread_id.clone(), RunId::random())
+            .with_messages(vec![Message::new_user("Hello")]);
+        msg_tx.send(AgentMessage::new(input1)).await.expect("Should send");
+
+        // Wait for first response
+        drain_events_until_run_finished(&mut event_rx).await;
+
+        // Send follow-up message (same thread)
+        let input2 = RunAgentInput::new(thread_id.clone(), RunId::random())
+            .with_messages(vec![Message::new_user("Follow up message")]);
+        msg_tx.send(AgentMessage::new(input2)).await.expect("Should send");
+
+        // Verify second run completes
+        drain_events_until_run_finished(&mut event_rx).await;
+
+        drop(msg_tx);
+        let _ = tokio::time::timeout(std::time::Duration::from_millis(200), handle).await;
+    }
+
+    #[tokio::test]
+    async fn test_event_sequence() {
+        use ag_ui_core::types::{Message, RunAgentInput};
+        use ag_ui_core::Event;
+
+        // Setup server state
+        let state = ServerState::new();
+        let msg_tx = state.message_sender();
+        let mut event_rx = state.subscribe();
+        let msg_rx = state.take_message_receiver().await.expect("receiver");
+        let event_bridge = state.event_sender();
+        let mut processor = AgentProcessor::with_defaults(msg_rx, event_bridge);
+
+        tokio::spawn(async move { processor.run().await; });
+
+        // Send message
+        let thread_id = ThreadId::random();
+        let input = RunAgentInput::new(thread_id, RunId::random())
+            .with_messages(vec![Message::new_user("Test event sequence")]);
+        msg_tx.send(AgentMessage::new(input)).await.unwrap();
+
+        // Collect events
+        let events = collect_until_finished(&mut event_rx).await;
+
+        // Verify sequence
+        assert!(!events.is_empty(), "Should receive at least one event");
+        assert!(matches!(events[0], Event::RunStarted(_)), "First event should be RunStarted");
+
+        // Should end with RunFinished or RunError
+        assert!(
+            matches!(events.last(), Some(Event::RunFinished(_) | Event::RunError(_))),
+            "Last event should be RunFinished or RunError"
+        );
+
+        // When successful (API key available), we expect at least:
+        // RunStarted -> StepStarted -> StepFinished -> TextMessageStart -> TextMessageContent* -> TextMessageEnd -> RunFinished
+        // Without API key, we get: RunStarted -> StepStarted -> StepFinished -> RunError
+        // Either way, verify we have multiple events
+        assert!(events.len() >= 2, "Should have at least RunStarted and terminal event");
+
+        drop(msg_tx);
+    }
+
+    #[tokio::test]
+    async fn test_empty_message_error() {
+        use ag_ui_core::types::RunAgentInput;
+        use ag_ui_core::Event;
+
+        let state = ServerState::new();
+        let msg_tx = state.message_sender();
+        let mut event_rx = state.subscribe();
+        let msg_rx = state.take_message_receiver().await.expect("receiver");
+        let event_bridge = state.event_sender();
+        let mut processor = AgentProcessor::with_defaults(msg_rx, event_bridge);
+
+        tokio::spawn(async move { processor.run().await; });
+
+        // Send message with no user content
+        let input = RunAgentInput::new(ThreadId::random(), RunId::random());
+        msg_tx.send(AgentMessage::new(input)).await.unwrap();
+
+        // Collect events
+        let events = collect_until_finished(&mut event_rx).await;
+
+        // Should get RunStarted then RunError
+        assert!(matches!(events[0], Event::RunStarted(_)), "First should be RunStarted");
+        assert!(
+            matches!(events.last(), Some(Event::RunError(_))),
+            "Should end with RunError for empty message"
+        );
+
+        drop(msg_tx);
+    }
+
+    #[tokio::test]
+    async fn test_invalid_provider_error() {
+        use ag_ui_core::types::{Message, RunAgentInput};
+        use ag_ui_core::Event;
+
+        let state = ServerState::new();
+        let msg_tx = state.message_sender();
+        let mut event_rx = state.subscribe();
+        let msg_rx = state.take_message_receiver().await.expect("receiver");
+        let event_bridge = state.event_sender();
+
+        // Configure with invalid provider
+        let config = ProcessorConfig::new().with_provider("invalid_provider_xyz");
+        let mut processor = AgentProcessor::new(msg_rx, event_bridge, config);
+
+        tokio::spawn(async move { processor.run().await; });
+
+        let input = RunAgentInput::new(ThreadId::random(), RunId::random())
+            .with_messages(vec![Message::new_user("Test invalid provider")]);
+        msg_tx.send(AgentMessage::new(input)).await.unwrap();
+
+        // Collect events
+        let events = collect_until_finished(&mut event_rx).await;
+
+        // Should error due to unsupported provider
+        assert!(
+            matches!(events.last(), Some(Event::RunError(_))),
+            "Should end with RunError for invalid provider"
+        );
+
+        drop(msg_tx);
+    }
+
+    #[tokio::test]
+    async fn test_custom_system_prompt() {
+        use ag_ui_core::types::{Message, RunAgentInput};
+
+        let state = ServerState::new();
+        let msg_tx = state.message_sender();
+        let mut event_rx = state.subscribe();
+        let msg_rx = state.take_message_receiver().await.expect("receiver");
+        let event_bridge = state.event_sender();
+
+        // Configure with custom system prompt
+        let config = ProcessorConfig::new()
+            .with_system_prompt("You are a DevOps assistant. Always respond with deployment advice.");
+        let mut processor = AgentProcessor::new(msg_rx, event_bridge, config);
+
+        tokio::spawn(async move { processor.run().await; });
+
+        let input = RunAgentInput::new(ThreadId::random(), RunId::random())
+            .with_messages(vec![Message::new_user("Hello")]);
+        msg_tx.send(AgentMessage::new(input)).await.unwrap();
+
+        // Should complete (may error without API key, but should not panic)
+        drain_events_until_run_finished(&mut event_rx).await;
+
+        drop(msg_tx);
+    }
 }