diff --git a/Cargo.lock b/Cargo.lock index a813ef2b..27c258c1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3428,6 +3428,7 @@ dependencies = [ "hyper-util", "iana-time-zone", "insta", + "ironclaw_common", "ironclaw_safety", "json5", "libsql", @@ -3485,6 +3486,14 @@ dependencies = [ "zip", ] +[[package]] +name = "ironclaw_common" +version = "0.1.0" +dependencies = [ + "serde", + "serde_json", +] + [[package]] name = "ironclaw_safety" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index 99992a40..395e42d3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,5 +1,5 @@ [workspace] -members = [".", "crates/ironclaw_safety"] +members = [".", "crates/ironclaw_common", "crates/ironclaw_safety"] exclude = [ "channels-src/discord", "channels-src/telegram", @@ -100,6 +100,9 @@ tower-http = { version = "0.6", features = ["trace", "cors", "set-header"] } # Cron scheduling for routines cron = "0.13" +# Shared types +ironclaw_common = { path = "crates/ironclaw_common", version = "0.1.0" } + # Safety/sanitization ironclaw_safety = { path = "crates/ironclaw_safety", version = "0.1.0" } regex = "1" diff --git a/crates/ironclaw_common/Cargo.toml b/crates/ironclaw_common/Cargo.toml new file mode 100644 index 00000000..353ab747 --- /dev/null +++ b/crates/ironclaw_common/Cargo.toml @@ -0,0 +1,18 @@ +[package] +name = "ironclaw_common" +version = "0.1.0" +edition = "2024" +rust-version = "1.92" +description = "Shared types and utilities for the IronClaw workspace" +authors = ["NEAR AI "] +license = "MIT OR Apache-2.0" +homepage = "https://github.com/nearai/ironclaw" +repository = "https://github.com/nearai/ironclaw" +publish = false + +[package.metadata.dist] +dist = false + +[dependencies] +serde = { version = "1", features = ["derive"] } +serde_json = "1" diff --git a/crates/ironclaw_common/src/event.rs b/crates/ironclaw_common/src/event.rs new file mode 100644 index 00000000..80cd570a --- /dev/null +++ b/crates/ironclaw_common/src/event.rs @@ -0,0 +1,166 @@ +//! Application-wide event types. +//! +//! `AppEvent` is the real-time event protocol used across the entire +//! application. The web gateway serialises these to SSE / WebSocket +//! frames, but other subsystems (agent loop, orchestrator, extensions) +//! produce and consume them too. + +use serde::Serialize; + +#[derive(Debug, Clone, Serialize)] +#[serde(tag = "type")] +pub enum AppEvent { + #[serde(rename = "response")] + Response { content: String, thread_id: String }, + #[serde(rename = "thinking")] + Thinking { + message: String, + #[serde(skip_serializing_if = "Option::is_none")] + thread_id: Option, + }, + #[serde(rename = "tool_started")] + ToolStarted { + name: String, + #[serde(skip_serializing_if = "Option::is_none")] + thread_id: Option, + }, + #[serde(rename = "tool_completed")] + ToolCompleted { + name: String, + success: bool, + #[serde(skip_serializing_if = "Option::is_none")] + error: Option, + #[serde(skip_serializing_if = "Option::is_none")] + parameters: Option, + #[serde(skip_serializing_if = "Option::is_none")] + thread_id: Option, + }, + #[serde(rename = "tool_result")] + ToolResult { + name: String, + preview: String, + #[serde(skip_serializing_if = "Option::is_none")] + thread_id: Option, + }, + #[serde(rename = "stream_chunk")] + StreamChunk { + content: String, + #[serde(skip_serializing_if = "Option::is_none")] + thread_id: Option, + }, + #[serde(rename = "status")] + Status { + message: String, + #[serde(skip_serializing_if = "Option::is_none")] + thread_id: Option, + }, + #[serde(rename = "job_started")] + JobStarted { + job_id: String, + title: String, + browse_url: String, + }, + #[serde(rename = "approval_needed")] + ApprovalNeeded { + request_id: String, + tool_name: String, + description: String, + parameters: String, + #[serde(skip_serializing_if = "Option::is_none")] + thread_id: Option, + /// Whether the "always" auto-approve option should be shown. + allow_always: bool, + }, + #[serde(rename = "auth_required")] + AuthRequired { + extension_name: String, + #[serde(skip_serializing_if = "Option::is_none")] + instructions: Option, + #[serde(skip_serializing_if = "Option::is_none")] + auth_url: Option, + #[serde(skip_serializing_if = "Option::is_none")] + setup_url: Option, + }, + #[serde(rename = "auth_completed")] + AuthCompleted { + extension_name: String, + success: bool, + message: String, + }, + #[serde(rename = "error")] + Error { + message: String, + #[serde(skip_serializing_if = "Option::is_none")] + thread_id: Option, + }, + #[serde(rename = "heartbeat")] + Heartbeat, + + // Sandbox job streaming events (worker + Claude Code bridge) + #[serde(rename = "job_message")] + JobMessage { + job_id: String, + role: String, + content: String, + }, + #[serde(rename = "job_tool_use")] + JobToolUse { + job_id: String, + tool_name: String, + input: serde_json::Value, + }, + #[serde(rename = "job_tool_result")] + JobToolResult { + job_id: String, + tool_name: String, + output: String, + }, + #[serde(rename = "job_status")] + JobStatus { job_id: String, message: String }, + #[serde(rename = "job_result")] + JobResult { + job_id: String, + status: String, + #[serde(skip_serializing_if = "Option::is_none")] + session_id: Option, + #[serde(skip_serializing_if = "Option::is_none")] + fallback_deliverable: Option, + }, + + /// An image was generated by a tool. + #[serde(rename = "image_generated")] + ImageGenerated { + data_url: String, + #[serde(skip_serializing_if = "Option::is_none")] + path: Option, + #[serde(skip_serializing_if = "Option::is_none")] + thread_id: Option, + }, + + /// Suggested follow-up messages for the user. + #[serde(rename = "suggestions")] + Suggestions { + suggestions: Vec, + #[serde(skip_serializing_if = "Option::is_none")] + thread_id: Option, + }, + + /// Per-turn token usage and cost summary. + #[serde(rename = "turn_cost")] + TurnCost { + input_tokens: u64, + output_tokens: u64, + cost_usd: String, + #[serde(skip_serializing_if = "Option::is_none")] + thread_id: Option, + }, + + /// Extension activation status change (WASM channels). + #[serde(rename = "extension_status")] + ExtensionStatus { + extension_name: String, + status: String, + #[serde(skip_serializing_if = "Option::is_none")] + message: Option, + }, +} diff --git a/crates/ironclaw_common/src/lib.rs b/crates/ironclaw_common/src/lib.rs new file mode 100644 index 00000000..6822bad1 --- /dev/null +++ b/crates/ironclaw_common/src/lib.rs @@ -0,0 +1,7 @@ +//! Shared types and utilities for the IronClaw workspace. + +mod event; +mod util; + +pub use event::AppEvent; +pub use util::truncate_preview; diff --git a/crates/ironclaw_common/src/util.rs b/crates/ironclaw_common/src/util.rs new file mode 100644 index 00000000..4f054671 --- /dev/null +++ b/crates/ironclaw_common/src/util.rs @@ -0,0 +1,100 @@ +//! Shared utility functions. + +/// Truncate a string to at most `max_bytes` bytes at a char boundary, appending "...". +/// +/// If the input is wrapped in `...` and truncation +/// removes the closing tag, the tag is re-appended so downstream XML parsers +/// never see an unclosed element. +pub fn truncate_preview(s: &str, max_bytes: usize) -> String { + if s.len() <= max_bytes { + return s.to_string(); + } + // Walk backwards from max_bytes to find a valid char boundary + let mut end = max_bytes; + while end > 0 && !s.is_char_boundary(end) { + end -= 1; + } + let mut result = format!("{}...", &s[..end]); + + // Re-close if truncation cut through the closing tag. + if s.starts_with("") { + result.push_str("\n"); + } + + result +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_truncate_preview_short_string() { + assert_eq!(truncate_preview("hello", 10), "hello"); + } + + #[test] + fn test_truncate_preview_exact_boundary() { + assert_eq!(truncate_preview("hello", 5), "hello"); + } + + #[test] + fn test_truncate_preview_truncates_ascii() { + assert_eq!(truncate_preview("hello world", 5), "hello..."); + } + + #[test] + fn test_truncate_preview_empty_string() { + assert_eq!(truncate_preview("", 10), ""); + } + + #[test] + fn test_truncate_preview_multibyte_char_boundary() { + let s = "a\u{20AC}b"; + let result = truncate_preview(s, 3); + assert_eq!(result, "a..."); + } + + #[test] + fn test_truncate_preview_emoji() { + let s = "hi\u{1F980}"; + let result = truncate_preview(s, 4); + assert_eq!(result, "hi..."); + } + + #[test] + fn test_truncate_preview_cjk() { + let s = "\u{4F60}\u{597D}\u{4E16}\u{754C}"; + let result = truncate_preview(s, 7); + assert_eq!(result, "\u{4F60}\u{597D}..."); + } + + #[test] + fn test_truncate_preview_zero_max_bytes() { + assert_eq!(truncate_preview("hello", 0), "..."); + } + + #[test] + fn test_truncate_preview_closes_tool_output_tag() { + let s = "\nSome very long content here\n"; + let result = truncate_preview(s, 60); + assert!(result.ends_with("")); + assert!(result.contains("...")); + } + + #[test] + fn test_truncate_preview_no_extra_close_when_intact() { + let s = "\nshort\n"; + let result = truncate_preview(s, 500); + assert_eq!(result, s); + assert_eq!(result.matches("").count(), 1); + } + + #[test] + fn test_truncate_preview_non_xml_unaffected() { + let s = "Just a plain long string that gets truncated"; + let result = truncate_preview(s, 10); + assert_eq!(result, "Just a pla..."); + assert!(!result.contains("")); + } +} diff --git a/src/agent/job_monitor.rs b/src/agent/job_monitor.rs index 02f5e3e2..e102dfbf 100644 --- a/src/agent/job_monitor.rs +++ b/src/agent/job_monitor.rs @@ -21,8 +21,8 @@ use tokio::task::JoinHandle; use uuid::Uuid; use crate::channels::IncomingMessage; -use crate::channels::web::types::SseEvent; use crate::context::{ContextManager, JobState}; +use ironclaw_common::AppEvent; /// Route context for forwarding job monitor events back to the user's channel. #[derive(Debug, Clone)] @@ -36,15 +36,15 @@ pub struct JobMonitorRoute { /// injects assistant messages into the agent loop. /// /// The monitor forwards: -/// - `SseEvent::JobMessage` (assistant role): injected as incoming messages so +/// - `AppEvent::JobMessage` (assistant role): injected as incoming messages so /// the main agent can read and relay to the user. -/// - `SseEvent::JobResult`: injected as a completion notice, then the task exits. +/// - `AppEvent::JobResult`: injected as a completion notice, then the task exits. /// /// Tool use/result and status events are intentionally skipped (too noisy for /// the main agent's context window). pub fn spawn_job_monitor( job_id: Uuid, - event_rx: broadcast::Receiver<(Uuid, String, SseEvent)>, + event_rx: broadcast::Receiver<(Uuid, String, AppEvent)>, inject_tx: mpsc::Sender, route: JobMonitorRoute, ) -> JoinHandle<()> { @@ -56,7 +56,7 @@ pub fn spawn_job_monitor( /// jobs don't stay `InProgress` forever in the `ContextManager`. pub fn spawn_job_monitor_with_context( job_id: Uuid, - mut event_rx: broadcast::Receiver<(Uuid, String, SseEvent)>, + mut event_rx: broadcast::Receiver<(Uuid, String, AppEvent)>, inject_tx: mpsc::Sender, route: JobMonitorRoute, context_manager: Option>, @@ -74,7 +74,7 @@ pub fn spawn_job_monitor_with_context( } match event { - SseEvent::JobMessage { role, content, .. } if role == "assistant" => { + AppEvent::JobMessage { role, content, .. } if role == "assistant" => { let mut msg = IncomingMessage::new( route.channel.clone(), route.user_id.clone(), @@ -92,7 +92,7 @@ pub fn spawn_job_monitor_with_context( break; } } - SseEvent::JobResult { status, .. } => { + AppEvent::JobResult { status, .. } => { // Transition in-memory state so the job frees its // max_jobs slot and query tools show the final state. if let Some(ref cm) = context_manager { @@ -162,7 +162,7 @@ pub fn spawn_job_monitor_with_context( /// inject messages into) but we still need to free the `max_jobs` slot. pub fn spawn_completion_watcher( job_id: Uuid, - mut event_rx: broadcast::Receiver<(Uuid, String, SseEvent)>, + mut event_rx: broadcast::Receiver<(Uuid, String, AppEvent)>, context_manager: Arc, ) -> JoinHandle<()> { let short_id = job_id.to_string()[..8].to_string(); @@ -170,7 +170,7 @@ pub fn spawn_completion_watcher( tokio::spawn(async move { loop { match event_rx.recv().await { - Ok((ev_job_id, _user_id, SseEvent::JobResult { status, .. })) + Ok((ev_job_id, _user_id, AppEvent::JobResult { status, .. })) if ev_job_id == job_id => { let target = if status == "completed" { @@ -229,7 +229,7 @@ mod tests { #[tokio::test] async fn test_monitor_forwards_assistant_messages() { - let (event_tx, _) = broadcast::channel::<(Uuid, String, SseEvent)>(16); + let (event_tx, _) = broadcast::channel::<(Uuid, String, AppEvent)>(16); let (inject_tx, mut inject_rx) = mpsc::channel::(16); let job_id = Uuid::new_v4(); @@ -240,7 +240,7 @@ mod tests { .send(( job_id, "test-user".to_string(), - SseEvent::JobMessage { + AppEvent::JobMessage { job_id: job_id.to_string(), role: "assistant".to_string(), content: "I found a bug".to_string(), @@ -262,7 +262,7 @@ mod tests { #[tokio::test] async fn test_monitor_ignores_other_jobs() { - let (event_tx, _) = broadcast::channel::<(Uuid, String, SseEvent)>(16); + let (event_tx, _) = broadcast::channel::<(Uuid, String, AppEvent)>(16); let (inject_tx, mut inject_rx) = mpsc::channel::(16); let job_id = Uuid::new_v4(); @@ -274,7 +274,7 @@ mod tests { .send(( other_job_id, "test-user".to_string(), - SseEvent::JobMessage { + AppEvent::JobMessage { job_id: other_job_id.to_string(), role: "assistant".to_string(), content: "wrong job".to_string(), @@ -293,7 +293,7 @@ mod tests { #[tokio::test] async fn test_monitor_exits_on_job_result() { - let (event_tx, _) = broadcast::channel::<(Uuid, String, SseEvent)>(16); + let (event_tx, _) = broadcast::channel::<(Uuid, String, AppEvent)>(16); let (inject_tx, mut inject_rx) = mpsc::channel::(16); let job_id = Uuid::new_v4(); @@ -304,7 +304,7 @@ mod tests { .send(( job_id, "test-user".to_string(), - SseEvent::JobResult { + AppEvent::JobResult { job_id: job_id.to_string(), status: "completed".to_string(), session_id: None, @@ -329,7 +329,7 @@ mod tests { #[tokio::test] async fn test_monitor_skips_tool_events() { - let (event_tx, _) = broadcast::channel::<(Uuid, String, SseEvent)>(16); + let (event_tx, _) = broadcast::channel::<(Uuid, String, AppEvent)>(16); let (inject_tx, mut inject_rx) = mpsc::channel::(16); let job_id = Uuid::new_v4(); @@ -340,7 +340,7 @@ mod tests { .send(( job_id, "test-user".to_string(), - SseEvent::JobToolUse { + AppEvent::JobToolUse { job_id: job_id.to_string(), tool_name: "shell".to_string(), input: serde_json::json!({"command": "ls"}), @@ -353,7 +353,7 @@ mod tests { .send(( job_id, "test-user".to_string(), - SseEvent::JobMessage { + AppEvent::JobMessage { job_id: job_id.to_string(), role: "user".to_string(), content: "user prompt".to_string(), @@ -409,7 +409,7 @@ mod tests { .await .unwrap(); - let (event_tx, _) = broadcast::channel::<(Uuid, String, SseEvent)>(16); + let (event_tx, _) = broadcast::channel::<(Uuid, String, AppEvent)>(16); let (inject_tx, mut inject_rx) = mpsc::channel::(16); let handle = spawn_job_monitor_with_context( @@ -425,7 +425,7 @@ mod tests { .send(( job_id, "test-user".to_string(), - SseEvent::JobResult { + AppEvent::JobResult { job_id: job_id.to_string(), status: "completed".to_string(), session_id: None, @@ -458,7 +458,7 @@ mod tests { .await .unwrap(); - let (event_tx, _) = broadcast::channel::<(Uuid, String, SseEvent)>(16); + let (event_tx, _) = broadcast::channel::<(Uuid, String, AppEvent)>(16); let (inject_tx, mut inject_rx) = mpsc::channel::(16); let handle = spawn_job_monitor_with_context( @@ -474,7 +474,7 @@ mod tests { .send(( job_id, "test-user".to_string(), - SseEvent::JobResult { + AppEvent::JobResult { job_id: job_id.to_string(), status: "failed".to_string(), session_id: None, @@ -507,14 +507,14 @@ mod tests { .await .unwrap(); - let (event_tx, _) = broadcast::channel::<(Uuid, String, SseEvent)>(16); + let (event_tx, _) = broadcast::channel::<(Uuid, String, AppEvent)>(16); let handle = spawn_completion_watcher(job_id, event_tx.subscribe(), Arc::clone(&cm)); event_tx .send(( job_id, "test-user".to_string(), - SseEvent::JobResult { + AppEvent::JobResult { job_id: job_id.to_string(), status: "completed".to_string(), session_id: None, diff --git a/src/agent/session.rs b/src/agent/session.rs index 45594922..7ec2023f 100644 --- a/src/agent/session.rs +++ b/src/agent/session.rs @@ -16,8 +16,8 @@ use chrono::{DateTime, TimeDelta, Utc}; use serde::{Deserialize, Serialize}; use uuid::Uuid; -use crate::channels::web::util::truncate_preview; use crate::llm::{ChatMessage, ToolCall, generate_tool_call_id}; +use ironclaw_common::truncate_preview; /// A session containing one or more threads. #[derive(Debug, Clone, Serialize, Deserialize)] diff --git a/src/agent/thread_ops.rs b/src/agent/thread_ops.rs index ddfd0c0f..b2820e7e 100644 --- a/src/agent/thread_ops.rs +++ b/src/agent/thread_ops.rs @@ -16,12 +16,12 @@ use crate::agent::dispatcher::{ }; use crate::agent::session::{MAX_PENDING_MESSAGES, PendingApproval, Session, ThreadState}; use crate::agent::submission::SubmissionResult; -use crate::channels::web::util::truncate_preview; use crate::channels::{IncomingMessage, StatusUpdate}; use crate::context::JobContext; use crate::error::Error; use crate::llm::{ChatMessage, ToolCall}; use crate::tools::redact_params; +use ironclaw_common::truncate_preview; const FORGED_THREAD_ID_ERROR: &str = "Invalid or unauthorized thread ID."; diff --git a/src/channels/web/handlers/chat.rs b/src/channels/web/handlers/chat.rs index 9753c015..de4b3155 100644 --- a/src/channels/web/handlers/chat.rs +++ b/src/channels/web/handlers/chat.rs @@ -175,7 +175,7 @@ pub async fn chat_auth_token_handler( if result.verification.is_some() { state.sse.broadcast_for_user( &user.user_id, - SseEvent::AuthRequired { + AppEvent::AuthRequired { extension_name: req.extension_name.clone(), instructions: Some(result.message), auth_url: None, @@ -187,7 +187,7 @@ pub async fn chat_auth_token_handler( state.sse.broadcast_for_user( &user.user_id, - SseEvent::AuthCompleted { + AppEvent::AuthCompleted { extension_name: req.extension_name.clone(), success: true, message: result.message, @@ -202,7 +202,7 @@ pub async fn chat_auth_token_handler( if matches!(e, crate::extensions::ExtensionError::ValidationFailed(_)) { state.sse.broadcast_for_user( &user.user_id, - SseEvent::AuthRequired { + AppEvent::AuthRequired { extension_name: req.extension_name.clone(), instructions: Some(msg.clone()), auth_url: None, diff --git a/src/channels/web/mod.rs b/src/channels/web/mod.rs index b26a7829..a44d9979 100644 --- a/src/channels/web/mod.rs +++ b/src/channels/web/mod.rs @@ -58,7 +58,7 @@ use self::log_layer::{LogBroadcaster, LogLevelHandle}; use self::auth::MultiAuthState; use self::server::GatewayState; use self::sse::SseManager; -use self::types::SseEvent; +use self::types::AppEvent; /// Web gateway channel implementing the Channel trait. pub struct GatewayChannel { @@ -367,7 +367,7 @@ impl Channel for GatewayChannel { self.state.sse.broadcast_for_user( &msg.user_id, - SseEvent::Response { + AppEvent::Response { content: response.content, thread_id, }, @@ -386,11 +386,11 @@ impl Channel for GatewayChannel { .and_then(|v| v.as_str()) .map(String::from); let event = match status { - StatusUpdate::Thinking(msg) => SseEvent::Thinking { + StatusUpdate::Thinking(msg) => AppEvent::Thinking { message: msg, thread_id: thread_id.clone(), }, - StatusUpdate::ToolStarted { name } => SseEvent::ToolStarted { + StatusUpdate::ToolStarted { name } => AppEvent::ToolStarted { name, thread_id: thread_id.clone(), }, @@ -399,23 +399,23 @@ impl Channel for GatewayChannel { success, error, parameters, - } => SseEvent::ToolCompleted { + } => AppEvent::ToolCompleted { name, success, error, parameters, thread_id: thread_id.clone(), }, - StatusUpdate::ToolResult { name, preview } => SseEvent::ToolResult { + StatusUpdate::ToolResult { name, preview } => AppEvent::ToolResult { name, preview, thread_id: thread_id.clone(), }, - StatusUpdate::StreamChunk(content) => SseEvent::StreamChunk { + StatusUpdate::StreamChunk(content) => AppEvent::StreamChunk { content, thread_id: thread_id.clone(), }, - StatusUpdate::Status(msg) => SseEvent::Status { + StatusUpdate::Status(msg) => AppEvent::Status { message: msg, thread_id: thread_id.clone(), }, @@ -423,7 +423,7 @@ impl Channel for GatewayChannel { job_id, title, browse_url, - } => SseEvent::JobStarted { + } => AppEvent::JobStarted { job_id, title, browse_url, @@ -434,7 +434,7 @@ impl Channel for GatewayChannel { description, parameters, allow_always, - } => SseEvent::ApprovalNeeded { + } => AppEvent::ApprovalNeeded { request_id, tool_name, description, @@ -448,7 +448,7 @@ impl Channel for GatewayChannel { instructions, auth_url, setup_url, - } => SseEvent::AuthRequired { + } => AppEvent::AuthRequired { extension_name, instructions, auth_url, @@ -458,17 +458,17 @@ impl Channel for GatewayChannel { extension_name, success, message, - } => SseEvent::AuthCompleted { + } => AppEvent::AuthCompleted { extension_name, success, message, }, - StatusUpdate::ImageGenerated { data_url, path } => SseEvent::ImageGenerated { + StatusUpdate::ImageGenerated { data_url, path } => AppEvent::ImageGenerated { data_url, path, thread_id: thread_id.clone(), }, - StatusUpdate::Suggestions { suggestions } => SseEvent::Suggestions { + StatusUpdate::Suggestions { suggestions } => AppEvent::Suggestions { suggestions, thread_id, }, @@ -476,7 +476,7 @@ impl Channel for GatewayChannel { input_tokens, output_tokens, cost_usd, - } => SseEvent::TurnCost { + } => AppEvent::TurnCost { input_tokens, output_tokens, cost_usd, @@ -512,7 +512,7 @@ impl Channel for GatewayChannel { }; self.state.sse.broadcast_for_user( user_id, - SseEvent::Response { + AppEvent::Response { content: response.content, thread_id, }, diff --git a/src/channels/web/server.rs b/src/channels/web/server.rs index aaa479fa..f81e703b 100644 --- a/src/channels/web/server.rs +++ b/src/channels/web/server.rs @@ -811,7 +811,7 @@ async fn oauth_callback_handler( if let Some(ref sse) = flow.sse_manager { sse.broadcast_for_user( &flow.user_id, - SseEvent::AuthCompleted { + AppEvent::AuthCompleted { extension_name: flow.extension_name.clone(), success: false, message: "OAuth flow expired. Please try again.".to_string(), @@ -953,7 +953,7 @@ async fn oauth_callback_handler( if let Some(ref sse) = flow.sse_manager { sse.broadcast_for_user( &flow.user_id, - SseEvent::AuthCompleted { + AppEvent::AuthCompleted { extension_name: flow.extension_name, success, message: final_message.clone(), @@ -1203,7 +1203,7 @@ async fn slack_relay_oauth_callback_handler( }; // Broadcast SSE event to notify the web UI - state.sse.broadcast(SseEvent::AuthCompleted { + state.sse.broadcast(AppEvent::AuthCompleted { extension_name: DEFAULT_RELAY_NAME.to_string(), success, message: message.clone(), @@ -1470,7 +1470,7 @@ async fn chat_auth_token_handler( if result.verification.is_some() { state.sse.broadcast_for_user( &user.user_id, - SseEvent::AuthRequired { + AppEvent::AuthRequired { extension_name: req.extension_name.clone(), instructions: Some(result.message), auth_url: None, @@ -1483,7 +1483,7 @@ async fn chat_auth_token_handler( state.sse.broadcast_for_user( &user.user_id, - SseEvent::AuthCompleted { + AppEvent::AuthCompleted { extension_name: req.extension_name.clone(), success: true, message: result.message, @@ -1492,7 +1492,7 @@ async fn chat_auth_token_handler( } else { state.sse.broadcast_for_user( &user.user_id, - SseEvent::AuthCompleted { + AppEvent::AuthCompleted { extension_name: req.extension_name.clone(), success: false, message: result.message, @@ -1508,7 +1508,7 @@ async fn chat_auth_token_handler( if matches!(e, crate::extensions::ExtensionError::ValidationFailed(_)) { state.sse.broadcast_for_user( &user.user_id, - SseEvent::AuthRequired { + AppEvent::AuthRequired { extension_name: req.extension_name.clone(), instructions: Some(msg.clone()), auth_url: None, @@ -2476,7 +2476,7 @@ async fn extensions_setup_submit_handler( // auth card or setup modal that was triggered by tool_auth/tool_activate. state.sse.broadcast_for_user( &user.user_id, - SseEvent::AuthCompleted { + AppEvent::AuthCompleted { extension_name: name.clone(), success: result.activated, message: resp.message.clone(), @@ -3167,7 +3167,7 @@ mod tests { Ok(Ok(scoped)) if matches!( scoped.event, - crate::channels::web::types::SseEvent::AuthRequired { .. } + crate::channels::web::types::AppEvent::AuthRequired { .. } ) => { panic!("verification responses should not emit auth_required SSE events") @@ -3449,7 +3449,7 @@ mod tests { assert_eq!(resp.status(), StatusCode::OK); match receiver.recv().await.expect("auth_completed event").event { - crate::channels::web::types::SseEvent::AuthCompleted { + crate::channels::web::types::AppEvent::AuthCompleted { extension_name, success, message, diff --git a/src/channels/web/sse.rs b/src/channels/web/sse.rs index 46841e19..81d9de9a 100644 --- a/src/channels/web/sse.rs +++ b/src/channels/web/sse.rs @@ -11,7 +11,7 @@ use tokio::sync::broadcast; use tokio_stream::StreamExt; use tokio_stream::wrappers::BroadcastStream; -use crate::channels::web::types::SseEvent; +use crate::channels::web::types::AppEvent; /// Maximum number of concurrent SSE/WebSocket connections. /// Prevents resource exhaustion from connection flooding. @@ -25,7 +25,7 @@ const MAX_CONNECTIONS: u64 = 100; #[derive(Debug, Clone)] pub(crate) struct ScopedEvent { pub(crate) user_id: Option, - pub(crate) event: SseEvent, + pub(crate) event: AppEvent, } /// Manages SSE broadcast to all connected browser tabs. @@ -75,7 +75,7 @@ impl SseManager { } /// Broadcast an event to all connected clients (global/unscoped). - pub fn broadcast(&self, event: SseEvent) { + pub fn broadcast(&self, event: AppEvent) { let _ = self.tx.send(ScopedEvent { user_id: None, event, @@ -86,7 +86,7 @@ impl SseManager { /// /// Only subscribers for this user_id (or unscoped subscribers) will /// receive the event. - pub fn broadcast_for_user(&self, user_id: &str, event: SseEvent) { + pub fn broadcast_for_user(&self, user_id: &str, event: AppEvent) { let _ = self.tx.send(ScopedEvent { user_id: Some(user_id.to_string()), event, @@ -108,7 +108,7 @@ impl SseManager { pub fn subscribe_raw( &self, user_id: Option, - ) -> Option + Send + 'static + use<>> { + ) -> Option + Send + 'static + use<>> { // Atomically increment only if below the limit. This prevents // concurrent callers from overshooting max_connections. let counter = Arc::clone(&self.connection_count); @@ -187,28 +187,28 @@ impl SseManager { } }; let event_type = match &event { - SseEvent::Response { .. } => "response", - SseEvent::Thinking { .. } => "thinking", - SseEvent::ToolStarted { .. } => "tool_started", - SseEvent::ToolCompleted { .. } => "tool_completed", - SseEvent::ToolResult { .. } => "tool_result", - SseEvent::StreamChunk { .. } => "stream_chunk", - SseEvent::Status { .. } => "status", - SseEvent::ApprovalNeeded { .. } => "approval_needed", - SseEvent::AuthRequired { .. } => "auth_required", - SseEvent::AuthCompleted { .. } => "auth_completed", - SseEvent::Error { .. } => "error", - SseEvent::JobStarted { .. } => "job_started", - SseEvent::JobMessage { .. } => "job_message", - SseEvent::JobToolUse { .. } => "job_tool_use", - SseEvent::JobToolResult { .. } => "job_tool_result", - SseEvent::JobStatus { .. } => "job_status", - SseEvent::JobResult { .. } => "job_result", - SseEvent::Heartbeat => "heartbeat", - SseEvent::ImageGenerated { .. } => "image_generated", - SseEvent::Suggestions { .. } => "suggestions", - SseEvent::TurnCost { .. } => "turn_cost", - SseEvent::ExtensionStatus { .. } => "extension_status", + AppEvent::Response { .. } => "response", + AppEvent::Thinking { .. } => "thinking", + AppEvent::ToolStarted { .. } => "tool_started", + AppEvent::ToolCompleted { .. } => "tool_completed", + AppEvent::ToolResult { .. } => "tool_result", + AppEvent::StreamChunk { .. } => "stream_chunk", + AppEvent::Status { .. } => "status", + AppEvent::ApprovalNeeded { .. } => "approval_needed", + AppEvent::AuthRequired { .. } => "auth_required", + AppEvent::AuthCompleted { .. } => "auth_completed", + AppEvent::Error { .. } => "error", + AppEvent::JobStarted { .. } => "job_started", + AppEvent::JobMessage { .. } => "job_message", + AppEvent::JobToolUse { .. } => "job_tool_use", + AppEvent::JobToolResult { .. } => "job_tool_result", + AppEvent::JobStatus { .. } => "job_status", + AppEvent::JobResult { .. } => "job_result", + AppEvent::Heartbeat => "heartbeat", + AppEvent::ImageGenerated { .. } => "image_generated", + AppEvent::Suggestions { .. } => "suggestions", + AppEvent::TurnCost { .. } => "turn_cost", + AppEvent::ExtensionStatus { .. } => "extension_status", }; Some(Ok(Event::default().event(event_type).data(data))) }); @@ -272,7 +272,7 @@ mod tests { fn test_broadcast_without_receivers() { let manager = SseManager::new(); // Should not panic even with no receivers - manager.broadcast(SseEvent::Heartbeat); + manager.broadcast(AppEvent::Heartbeat); } #[tokio::test] @@ -280,14 +280,14 @@ mod tests { let manager = SseManager::new(); let mut stream = Box::pin(manager.subscribe_raw(None).expect("should subscribe")); - manager.broadcast(SseEvent::Status { + manager.broadcast(AppEvent::Status { message: "test".to_string(), thread_id: None, }); let event = stream.next().await.unwrap(); match event { - SseEvent::Status { message, .. } => assert_eq!(message, "test"), + AppEvent::Status { message, .. } => assert_eq!(message, "test"), _ => panic!("unexpected event type"), } } @@ -299,14 +299,14 @@ mod tests { assert_eq!(manager.connection_count(), 1); - manager.broadcast(SseEvent::Thinking { + manager.broadcast(AppEvent::Thinking { message: "working".to_string(), thread_id: None, }); let event = stream.next().await.unwrap(); match event { - SseEvent::Thinking { message, .. } => assert_eq!(message, "working"), + AppEvent::Thinking { message, .. } => assert_eq!(message, "working"), _ => panic!("Expected Thinking event"), } } @@ -329,12 +329,12 @@ mod tests { let mut s2 = Box::pin(manager.subscribe_raw(None).expect("should subscribe")); assert_eq!(manager.connection_count(), 2); - manager.broadcast(SseEvent::Heartbeat); + manager.broadcast(AppEvent::Heartbeat); let e1 = s1.next().await.unwrap(); let e2 = s2.next().await.unwrap(); - assert!(matches!(e1, SseEvent::Heartbeat)); - assert!(matches!(e2, SseEvent::Heartbeat)); + assert!(matches!(e1, AppEvent::Heartbeat)); + assert!(matches!(e2, AppEvent::Heartbeat)); drop(s1); assert_eq!(manager.connection_count(), 1); @@ -373,25 +373,25 @@ mod tests { // Send event scoped to alice manager.broadcast_for_user( "alice", - SseEvent::Status { + AppEvent::Status { message: "alice only".to_string(), thread_id: None, }, ); // Send global event - manager.broadcast(SseEvent::Heartbeat); + manager.broadcast(AppEvent::Heartbeat); // Alice gets her scoped event let e = alice.next().await.unwrap(); - assert!(matches!(e, SseEvent::Status { .. })); + assert!(matches!(e, AppEvent::Status { .. })); // Alice also gets the global heartbeat let e = alice.next().await.unwrap(); - assert!(matches!(e, SseEvent::Heartbeat)); + assert!(matches!(e, AppEvent::Heartbeat)); // Bob only gets the global heartbeat (alice's event was filtered) let e = bob.next().await.unwrap(); // safety: test-only - assert!(matches!(e, SseEvent::Heartbeat)); // safety: test assertion + assert!(matches!(e, AppEvent::Heartbeat)); // safety: test assertion } } diff --git a/src/channels/web/types.rs b/src/channels/web/types.rs index 3ac4163c..be3fcac0 100644 --- a/src/channels/web/types.rs +++ b/src/channels/web/types.rs @@ -114,165 +114,9 @@ pub struct ApprovalRequest { pub thread_id: Option, } -// --- SSE Event Types --- +// --- App Event (re-exported from ironclaw_common) --- -#[derive(Debug, Clone, Serialize)] -#[serde(tag = "type")] -pub enum SseEvent { - #[serde(rename = "response")] - Response { content: String, thread_id: String }, - #[serde(rename = "thinking")] - Thinking { - message: String, - #[serde(skip_serializing_if = "Option::is_none")] - thread_id: Option, - }, - #[serde(rename = "tool_started")] - ToolStarted { - name: String, - #[serde(skip_serializing_if = "Option::is_none")] - thread_id: Option, - }, - #[serde(rename = "tool_completed")] - ToolCompleted { - name: String, - success: bool, - #[serde(skip_serializing_if = "Option::is_none")] - error: Option, - #[serde(skip_serializing_if = "Option::is_none")] - parameters: Option, - #[serde(skip_serializing_if = "Option::is_none")] - thread_id: Option, - }, - #[serde(rename = "tool_result")] - ToolResult { - name: String, - preview: String, - #[serde(skip_serializing_if = "Option::is_none")] - thread_id: Option, - }, - #[serde(rename = "stream_chunk")] - StreamChunk { - content: String, - #[serde(skip_serializing_if = "Option::is_none")] - thread_id: Option, - }, - #[serde(rename = "status")] - Status { - message: String, - #[serde(skip_serializing_if = "Option::is_none")] - thread_id: Option, - }, - #[serde(rename = "job_started")] - JobStarted { - job_id: String, - title: String, - browse_url: String, - }, - #[serde(rename = "approval_needed")] - ApprovalNeeded { - request_id: String, - tool_name: String, - description: String, - parameters: String, - #[serde(skip_serializing_if = "Option::is_none")] - thread_id: Option, - /// Whether the "always" auto-approve option should be shown. - allow_always: bool, - }, - #[serde(rename = "auth_required")] - AuthRequired { - extension_name: String, - #[serde(skip_serializing_if = "Option::is_none")] - instructions: Option, - #[serde(skip_serializing_if = "Option::is_none")] - auth_url: Option, - #[serde(skip_serializing_if = "Option::is_none")] - setup_url: Option, - }, - #[serde(rename = "auth_completed")] - AuthCompleted { - extension_name: String, - success: bool, - message: String, - }, - #[serde(rename = "error")] - Error { - message: String, - #[serde(skip_serializing_if = "Option::is_none")] - thread_id: Option, - }, - #[serde(rename = "heartbeat")] - Heartbeat, - - // Sandbox job streaming events (worker + Claude Code bridge) - #[serde(rename = "job_message")] - JobMessage { - job_id: String, - role: String, - content: String, - }, - #[serde(rename = "job_tool_use")] - JobToolUse { - job_id: String, - tool_name: String, - input: serde_json::Value, - }, - #[serde(rename = "job_tool_result")] - JobToolResult { - job_id: String, - tool_name: String, - output: String, - }, - #[serde(rename = "job_status")] - JobStatus { job_id: String, message: String }, - #[serde(rename = "job_result")] - JobResult { - job_id: String, - status: String, - #[serde(skip_serializing_if = "Option::is_none")] - session_id: Option, - #[serde(skip_serializing_if = "Option::is_none")] - fallback_deliverable: Option, - }, - - /// An image was generated by a tool. - #[serde(rename = "image_generated")] - ImageGenerated { - data_url: String, - #[serde(skip_serializing_if = "Option::is_none")] - path: Option, - #[serde(skip_serializing_if = "Option::is_none")] - thread_id: Option, - }, - - /// Suggested follow-up messages for the user. - #[serde(rename = "suggestions")] - Suggestions { - suggestions: Vec, - #[serde(skip_serializing_if = "Option::is_none")] - thread_id: Option, - }, - - /// Per-turn token usage and cost summary. - #[serde(rename = "turn_cost")] - TurnCost { - input_tokens: u64, - output_tokens: u64, - cost_usd: String, - #[serde(skip_serializing_if = "Option::is_none")] - thread_id: Option, - }, - - /// Extension activation status change (WASM channels). - #[serde(rename = "extension_status")] - ExtensionStatus { - extension_name: String, - status: String, - #[serde(skip_serializing_if = "Option::is_none")] - message: Option, - }, -} +pub use ironclaw_common::AppEvent; // --- Memory --- @@ -784,31 +628,31 @@ pub enum WsServerMessage { } impl WsServerMessage { - /// Create a WsServerMessage from an SseEvent. - pub fn from_sse_event(event: &SseEvent) -> Self { + /// Create a WsServerMessage from an AppEvent. + pub fn from_app_event(event: &AppEvent) -> Self { let event_type = match event { - SseEvent::Response { .. } => "response", - SseEvent::Thinking { .. } => "thinking", - SseEvent::ToolStarted { .. } => "tool_started", - SseEvent::ToolCompleted { .. } => "tool_completed", - SseEvent::ToolResult { .. } => "tool_result", - SseEvent::StreamChunk { .. } => "stream_chunk", - SseEvent::Status { .. } => "status", - SseEvent::JobStarted { .. } => "job_started", - SseEvent::ApprovalNeeded { .. } => "approval_needed", - SseEvent::AuthRequired { .. } => "auth_required", - SseEvent::AuthCompleted { .. } => "auth_completed", - SseEvent::Error { .. } => "error", - SseEvent::Heartbeat => "heartbeat", - SseEvent::JobMessage { .. } => "job_message", - SseEvent::JobToolUse { .. } => "job_tool_use", - SseEvent::JobToolResult { .. } => "job_tool_result", - SseEvent::JobStatus { .. } => "job_status", - SseEvent::JobResult { .. } => "job_result", - SseEvent::ImageGenerated { .. } => "image_generated", - SseEvent::Suggestions { .. } => "suggestions", - SseEvent::TurnCost { .. } => "turn_cost", - SseEvent::ExtensionStatus { .. } => "extension_status", + AppEvent::Response { .. } => "response", + AppEvent::Thinking { .. } => "thinking", + AppEvent::ToolStarted { .. } => "tool_started", + AppEvent::ToolCompleted { .. } => "tool_completed", + AppEvent::ToolResult { .. } => "tool_result", + AppEvent::StreamChunk { .. } => "stream_chunk", + AppEvent::Status { .. } => "status", + AppEvent::JobStarted { .. } => "job_started", + AppEvent::ApprovalNeeded { .. } => "approval_needed", + AppEvent::AuthRequired { .. } => "auth_required", + AppEvent::AuthCompleted { .. } => "auth_completed", + AppEvent::Error { .. } => "error", + AppEvent::Heartbeat => "heartbeat", + AppEvent::JobMessage { .. } => "job_message", + AppEvent::JobToolUse { .. } => "job_tool_use", + AppEvent::JobToolResult { .. } => "job_tool_result", + AppEvent::JobStatus { .. } => "job_status", + AppEvent::JobResult { .. } => "job_result", + AppEvent::ImageGenerated { .. } => "image_generated", + AppEvent::Suggestions { .. } => "suggestions", + AppEvent::TurnCost { .. } => "turn_cost", + AppEvent::ExtensionStatus { .. } => "extension_status", }; let data = serde_json::to_value(event).unwrap_or(serde_json::Value::Null); WsServerMessage::Event { @@ -1102,11 +946,11 @@ mod tests { #[test] fn test_ws_server_from_sse_response() { - let sse = SseEvent::Response { + let sse = AppEvent::Response { content: "hello".to_string(), thread_id: "t1".to_string(), }; - let ws = WsServerMessage::from_sse_event(&sse); + let ws = WsServerMessage::from_app_event(&sse); match ws { WsServerMessage::Event { event_type, data } => { assert_eq!(event_type, "response"); @@ -1119,11 +963,11 @@ mod tests { #[test] fn test_ws_server_from_sse_thinking() { - let sse = SseEvent::Thinking { + let sse = AppEvent::Thinking { message: "reasoning...".to_string(), thread_id: None, }; - let ws = WsServerMessage::from_sse_event(&sse); + let ws = WsServerMessage::from_app_event(&sse); match ws { WsServerMessage::Event { event_type, data } => { assert_eq!(event_type, "thinking"); @@ -1135,7 +979,7 @@ mod tests { #[test] fn test_ws_server_from_sse_approval_needed() { - let sse = SseEvent::ApprovalNeeded { + let sse = AppEvent::ApprovalNeeded { request_id: "r1".to_string(), tool_name: "shell".to_string(), description: "Run ls".to_string(), @@ -1143,7 +987,7 @@ mod tests { thread_id: Some("t1".to_string()), allow_always: true, }; - let ws = WsServerMessage::from_sse_event(&sse); + let ws = WsServerMessage::from_app_event(&sse); match ws { WsServerMessage::Event { event_type, data } => { assert_eq!(event_type, "approval_needed"); @@ -1156,8 +1000,8 @@ mod tests { #[test] fn test_ws_server_from_sse_heartbeat() { - let sse = SseEvent::Heartbeat; - let ws = WsServerMessage::from_sse_event(&sse); + let sse = AppEvent::Heartbeat; + let ws = WsServerMessage::from_app_event(&sse); match ws { WsServerMessage::Event { event_type, .. } => { assert_eq!(event_type, "heartbeat"); @@ -1198,7 +1042,7 @@ mod tests { #[test] fn test_sse_auth_required_serialize() { - let event = SseEvent::AuthRequired { + let event = AppEvent::AuthRequired { extension_name: "notion".to_string(), instructions: Some("Get your token from...".to_string()), auth_url: None, @@ -1215,7 +1059,7 @@ mod tests { #[test] fn test_sse_auth_completed_serialize() { - let event = SseEvent::AuthCompleted { + let event = AppEvent::AuthCompleted { extension_name: "notion".to_string(), success: true, message: "notion authenticated (3 tools loaded)".to_string(), @@ -1229,13 +1073,13 @@ mod tests { #[test] fn test_ws_server_from_sse_auth_required() { - let sse = SseEvent::AuthRequired { + let sse = AppEvent::AuthRequired { extension_name: "openai".to_string(), instructions: Some("Enter API key".to_string()), auth_url: None, setup_url: None, }; - let ws = WsServerMessage::from_sse_event(&sse); + let ws = WsServerMessage::from_app_event(&sse); match ws { WsServerMessage::Event { event_type, data } => { assert_eq!(event_type, "auth_required"); @@ -1247,12 +1091,12 @@ mod tests { #[test] fn test_ws_server_from_sse_auth_completed() { - let sse = SseEvent::AuthCompleted { + let sse = AppEvent::AuthCompleted { extension_name: "slack".to_string(), success: false, message: "Invalid token".to_string(), }; - let ws = WsServerMessage::from_sse_event(&sse); + let ws = WsServerMessage::from_app_event(&sse); match ws { WsServerMessage::Event { event_type, data } => { assert_eq!(event_type, "auth_completed"); diff --git a/src/channels/web/util.rs b/src/channels/web/util.rs index 0debe6a9..ed70c5ce 100644 --- a/src/channels/web/util.rs +++ b/src/channels/web/util.rs @@ -2,29 +2,7 @@ use crate::channels::web::types::{ToolCallInfo, TurnInfo}; -/// Truncate a string to at most `max_bytes` bytes at a char boundary, appending "...". -/// -/// If the input is wrapped in `…` and truncation -/// removes the closing tag, the tag is re-appended so downstream XML parsers -/// never see an unclosed element. -pub fn truncate_preview(s: &str, max_bytes: usize) -> String { - if s.len() <= max_bytes { - return s.to_string(); - } - // Walk backwards from max_bytes to find a valid char boundary - let mut end = max_bytes; - while end > 0 && !s.is_char_boundary(end) { - end -= 1; - } - let mut result = format!("{}...", &s[..end]); - - // Re-close if truncation cut through the closing tag. - if s.starts_with("") { - result.push_str("\n"); - } - - result -} +pub use ironclaw_common::truncate_preview; /// Build TurnInfo pairs from flat DB messages (user/tool_calls/assistant triples). /// @@ -118,88 +96,6 @@ mod tests { use super::*; use uuid::Uuid; - // ---- truncate_preview tests ---- - - #[test] - fn test_truncate_preview_short_string() { - assert_eq!(truncate_preview("hello", 10), "hello"); - } - - #[test] - fn test_truncate_preview_exact_boundary() { - assert_eq!(truncate_preview("hello", 5), "hello"); - } - - #[test] - fn test_truncate_preview_truncates_ascii() { - assert_eq!(truncate_preview("hello world", 5), "hello..."); - } - - #[test] - fn test_truncate_preview_empty_string() { - assert_eq!(truncate_preview("", 10), ""); - } - - #[test] - fn test_truncate_preview_multibyte_char_boundary() { - // '€' is 3 bytes (E2 82 AC). "a€b" = [61, E2, 82, AC, 62] = 5 bytes - // Truncating at max_bytes=3 should not split the euro sign. - let s = "a€b"; - let result = truncate_preview(s, 3); - // max_bytes=3 lands mid-€, so it walks back to byte 1 ("a") - assert_eq!(result, "a..."); - } - - #[test] - fn test_truncate_preview_emoji() { - // '🦀' is 4 bytes. "hi🦀" = 6 bytes - let s = "hi🦀"; - let result = truncate_preview(s, 4); - // max_bytes=4 lands mid-🦀, walks back to byte 2 ("hi") - assert_eq!(result, "hi..."); - } - - #[test] - fn test_truncate_preview_cjk() { - // CJK characters are 3 bytes each. "你好世界" = 12 bytes - let s = "你好世界"; - let result = truncate_preview(s, 7); - // max_bytes=7 lands mid-character (byte 7 is inside 世), walks back to 6 ("你好") - assert_eq!(result, "你好..."); - } - - #[test] - fn test_truncate_preview_zero_max_bytes() { - assert_eq!(truncate_preview("hello", 0), "..."); - } - - #[test] - fn test_truncate_preview_closes_tool_output_tag() { - let s = "\nSome very long content here\n"; - // Truncate so it cuts before the closing tag - let result = truncate_preview(s, 60); - assert!(result.ends_with("")); - assert!(result.contains("...")); - } - - #[test] - fn test_truncate_preview_no_extra_close_when_intact() { - let s = "\nshort\n"; - // The string is short enough not to be truncated - let result = truncate_preview(s, 500); - assert_eq!(result, s); - // Should not have a duplicate closing tag - assert_eq!(result.matches("").count(), 1); - } - - #[test] - fn test_truncate_preview_non_xml_unaffected() { - let s = "Just a plain long string that gets truncated"; - let result = truncate_preview(s, 10); - assert_eq!(result, "Just a pla..."); - assert!(!result.contains("")); - } - // ---- build_turns_from_db_messages tests ---- fn make_msg(role: &str, content: &str, offset_ms: i64) -> crate::history::ConversationMessage { diff --git a/src/channels/web/ws.rs b/src/channels/web/ws.rs index 3a601679..e209b4db 100644 --- a/src/channels/web/ws.rs +++ b/src/channels/web/ws.rs @@ -97,7 +97,7 @@ pub async fn handle_ws_connection( let msg = tokio::select! { event = event_stream.next() => { match event { - Some(sse_event) => WsServerMessage::from_sse_event(&sse_event), + Some(sse_event) => WsServerMessage::from_app_event(&sse_event), None => break, // Broadcast channel closed } } @@ -275,7 +275,7 @@ async fn handle_client_message( if result.verification.is_some() { state.sse.broadcast_for_user( user_id, - crate::channels::web::types::SseEvent::AuthRequired { + crate::channels::web::types::AppEvent::AuthRequired { extension_name: extension_name.clone(), instructions: Some(result.message), auth_url: None, @@ -286,7 +286,7 @@ async fn handle_client_message( crate::channels::web::server::clear_auth_mode(state, user_id).await; state.sse.broadcast_for_user( user_id, - crate::channels::web::types::SseEvent::AuthCompleted { + crate::channels::web::types::AppEvent::AuthCompleted { extension_name, success: true, message: result.message, @@ -299,7 +299,7 @@ async fn handle_client_message( if matches!(e, crate::extensions::ExtensionError::ValidationFailed(_)) { state.sse.broadcast_for_user( user_id, - crate::channels::web::types::SseEvent::AuthRequired { + crate::channels::web::types::AppEvent::AuthRequired { extension_name: extension_name.clone(), instructions: Some(msg.clone()), auth_url: None, diff --git a/src/extensions/manager.rs b/src/extensions/manager.rs index 7da9e980..b53cc257 100644 --- a/src/extensions/manager.rs +++ b/src/extensions/manager.rs @@ -1131,7 +1131,7 @@ impl ExtensionManager { /// Broadcast an extension status change to the web UI via SSE. async fn broadcast_extension_status(&self, name: &str, status: &str, message: Option<&str>) { if let Some(ref sse) = *self.sse_manager.read().await { - sse.broadcast(crate::channels::web::types::SseEvent::ExtensionStatus { + sse.broadcast(ironclaw_common::AppEvent::ExtensionStatus { extension_name: name.to_string(), status: status.to_string(), message: message.map(|m| m.to_string()), @@ -3310,7 +3310,7 @@ impl ExtensionManager { } if let Some(ref sse) = sse_manager { - sse.broadcast(crate::channels::web::types::SseEvent::AuthCompleted { + sse.broadcast(ironclaw_common::AppEvent::AuthCompleted { extension_name: ext_name, success, message, diff --git a/src/orchestrator/api.rs b/src/orchestrator/api.rs index 00f8a4da..dd9aba89 100644 --- a/src/orchestrator/api.rs +++ b/src/orchestrator/api.rs @@ -14,7 +14,6 @@ use serde::{Deserialize, Serialize}; use tokio::sync::{Mutex, broadcast}; use uuid::Uuid; -use crate::channels::web::types::SseEvent; use crate::db::Database; use crate::llm::{CompletionRequest, LlmProvider, ToolCompletionRequest}; use crate::orchestrator::auth::{TokenStore, worker_auth_middleware}; @@ -25,6 +24,7 @@ use crate::worker::api::{ CompletionReport, CredentialResponse, JobDescription, ProxyCompletionRequest, ProxyCompletionResponse, ProxyToolCompletionRequest, ProxyToolCompletionResponse, StatusUpdate, }; +use ironclaw_common::AppEvent; /// A follow-up prompt queued for a Claude Code bridge. #[derive(Debug, Clone, Serialize, Deserialize)] @@ -41,7 +41,7 @@ pub struct OrchestratorState { pub token_store: TokenStore, /// Broadcast channel for job events (consumed by the web gateway SSE). /// Tuple: (job_id, user_id, event). - pub job_event_tx: Option>, + pub job_event_tx: Option>, /// Buffered follow-up prompts for sandbox jobs, keyed by job_id. pub prompt_queue: Arc>>>, /// Database handle for persisting job events. @@ -280,7 +280,7 @@ async fn job_event_handler( // Convert to SSE event and broadcast let job_id_str = job_id.to_string(); let sse_event = match payload.event_type.as_str() { - "message" => SseEvent::JobMessage { + "message" => AppEvent::JobMessage { job_id: job_id_str, role: payload .data @@ -295,7 +295,7 @@ async fn job_event_handler( .unwrap_or("") .to_string(), }, - "tool_use" => SseEvent::JobToolUse { + "tool_use" => AppEvent::JobToolUse { job_id: job_id_str, tool_name: payload .data @@ -309,7 +309,7 @@ async fn job_event_handler( .cloned() .unwrap_or(serde_json::Value::Null), }, - "tool_result" => SseEvent::JobToolResult { + "tool_result" => AppEvent::JobToolResult { job_id: job_id_str, tool_name: payload .data @@ -324,7 +324,7 @@ async fn job_event_handler( .unwrap_or("") .to_string(), }, - "result" => SseEvent::JobResult { + "result" => AppEvent::JobResult { job_id: job_id_str, status: payload .data @@ -344,7 +344,7 @@ async fn job_event_handler( // gain context/memory tracking capabilities. fallback_deliverable: payload.data.get("fallback_deliverable").cloned(), }, - _ => SseEvent::JobStatus { + _ => AppEvent::JobStatus { job_id: job_id_str, message: payload .data @@ -817,7 +817,7 @@ mod tests { // No store configured, so user_id falls back to empty string. assert_eq!(recv_uid, ""); match event { - SseEvent::JobMessage { + AppEvent::JobMessage { job_id: jid, role, content, @@ -872,7 +872,7 @@ mod tests { let (_recv_id, _recv_uid, event) = rx.recv().await.unwrap(); match event { - SseEvent::JobToolUse { tool_name, .. } => { + AppEvent::JobToolUse { tool_name, .. } => { assert_eq!(tool_name, "shell"); } other => panic!("Expected JobToolUse, got {:?}", other), @@ -918,7 +918,7 @@ mod tests { let (_recv_id, _recv_uid, event) = rx.recv().await.unwrap(); // Unknown event types fall through to JobStatus - assert!(matches!(event, SseEvent::JobStatus { .. })); + assert!(matches!(event, AppEvent::JobStatus { .. })); } // -- Status update test -- diff --git a/src/orchestrator/mod.rs b/src/orchestrator/mod.rs index 896b5648..8d09dc53 100644 --- a/src/orchestrator/mod.rs +++ b/src/orchestrator/mod.rs @@ -46,10 +46,10 @@ use std::sync::Arc; use tokio::sync::{Mutex, broadcast}; use uuid::Uuid; -use crate::channels::web::types::SseEvent; use crate::db::Database; use crate::llm::LlmProvider; use crate::secrets::SecretsStore; +use ironclaw_common::AppEvent; /// Resolve the orchestrator port from the `ORCHESTRATOR_PORT` environment /// variable, falling back to 50051. @@ -63,7 +63,7 @@ fn resolve_orchestrator_port() -> u16 { /// Result of orchestrator setup, containing all handles needed by the agent. pub struct OrchestratorSetup { pub container_job_manager: Option>, - pub job_event_tx: Option>, + pub job_event_tx: Option>, pub prompt_queue: Arc>>>, pub docker_status: crate::sandbox::DockerStatus, } diff --git a/src/tools/builtin/job.rs b/src/tools/builtin/job.rs index 86d7e44d..4c711e69 100644 --- a/src/tools/builtin/job.rs +++ b/src/tools/builtin/job.rs @@ -17,7 +17,6 @@ use uuid::Uuid; use crate::bootstrap::ironclaw_base_dir; use crate::channels::IncomingMessage; -use crate::channels::web::types::SseEvent; use crate::context::{ContextManager, JobContext, JobState}; use crate::db::Database; use crate::history::SandboxJobRecord; @@ -25,6 +24,7 @@ use crate::orchestrator::auth::CredentialGrant; use crate::orchestrator::job_manager::{ContainerJobManager, JobMode}; use crate::secrets::SecretsStore; use crate::tools::tool::{ApprovalRequirement, Tool, ToolError, ToolOutput, require_str}; +use ironclaw_common::AppEvent; /// Lazy scheduler reference, filled after Agent::new creates the Scheduler. /// @@ -85,7 +85,7 @@ pub struct CreateJobTool { job_manager: Option>, store: Option>, /// Broadcast sender for job events (used to subscribe a monitor). - event_tx: Option>, + event_tx: Option>, /// Injection channel for pushing messages into the agent loop. inject_tx: Option>, /// Encrypted secrets store for validating credential grants. @@ -120,7 +120,7 @@ impl CreateJobTool { /// monitor that forwards Claude Code output to the main agent loop. pub fn with_monitor_deps( mut self, - event_tx: tokio::sync::broadcast::Sender<(Uuid, String, SseEvent)>, + event_tx: tokio::sync::broadcast::Sender<(Uuid, String, AppEvent)>, inject_tx: tokio::sync::mpsc::Sender, ) -> Self { self.event_tx = Some(event_tx); diff --git a/src/tools/registry.rs b/src/tools/registry.rs index bc3be144..8c08633b 100644 --- a/src/tools/registry.rs +++ b/src/tools/registry.rs @@ -383,11 +383,7 @@ impl ToolRegistry { job_manager: Option>, store: Option>, job_event_tx: Option< - tokio::sync::broadcast::Sender<( - uuid::Uuid, - String, - crate::channels::web::types::SseEvent, - )>, + tokio::sync::broadcast::Sender<(uuid::Uuid, String, ironclaw_common::AppEvent)>, >, inject_tx: Option>, prompt_queue: Option, diff --git a/src/worker/job.rs b/src/worker/job.rs index b2e3f7e6..f1843300 100644 --- a/src/worker/job.rs +++ b/src/worker/job.rs @@ -18,7 +18,6 @@ use crate::agent::agentic_loop::{ }; use crate::agent::scheduler::WorkerMessage; use crate::agent::task::TaskOutput; -use crate::channels::web::types::SseEvent; use crate::context::{ContextManager, JobState}; use crate::db::Database; use crate::error::Error; @@ -33,6 +32,7 @@ use crate::tools::rate_limiter::RateLimitResult; use crate::tools::{ ApprovalContext, ToolRegistry, autonomous_unavailable_error, prepare_tool_params, redact_params, }; +use ironclaw_common::AppEvent; /// Shared dependencies for worker execution. /// @@ -48,7 +48,7 @@ pub struct WorkerDeps { pub hooks: Arc, pub timeout: Duration, pub use_planning: bool, - /// SSE manager for live job event streaming to the web gateway. + /// SSE broadcast sender for live job event streaming to the web gateway. pub sse_tx: Option>, /// Approval context for tool execution. When `None`, all non-`Never` tools are /// blocked (legacy behavior). When `Some`, the context determines which tools @@ -141,7 +141,7 @@ impl Worker { if let Some(ref sse) = self.deps.sse_tx { let job_id_str = job_id.to_string(); let event = match event_type { - "message" => Some(SseEvent::JobMessage { + "message" => Some(AppEvent::JobMessage { job_id: job_id_str, role: data .get("role") @@ -154,7 +154,7 @@ impl Worker { .unwrap_or("") .to_string(), }), - "tool_use" => Some(SseEvent::JobToolUse { + "tool_use" => Some(AppEvent::JobToolUse { job_id: job_id_str, tool_name: data .get("tool_name") @@ -166,7 +166,7 @@ impl Worker { .cloned() .unwrap_or(serde_json::Value::Null), }), - "tool_result" => Some(SseEvent::JobToolResult { + "tool_result" => Some(AppEvent::JobToolResult { job_id: job_id_str, tool_name: data .get("tool_name") @@ -179,7 +179,7 @@ impl Worker { .unwrap_or("") .to_string(), }), - "status" => Some(SseEvent::JobStatus { + "status" => Some(AppEvent::JobStatus { job_id: job_id_str, message: data .get("message") @@ -187,7 +187,7 @@ impl Worker { .unwrap_or("") .to_string(), }), - "result" => Some(SseEvent::JobResult { + "result" => Some(AppEvent::JobResult { job_id: job_id_str, status: data .get("status") diff --git a/tests/multi_tenant_integration.rs b/tests/multi_tenant_integration.rs index 02eb60e8..76e28f3c 100644 --- a/tests/multi_tenant_integration.rs +++ b/tests/multi_tenant_integration.rs @@ -301,7 +301,7 @@ fn per_user_rate_limiter_single_user_mode() { #[tokio::test] async fn sse_scoped_event_only_delivered_to_target_user() { - use ironclaw::channels::web::types::SseEvent; + use ironclaw_common::AppEvent; use tokio_stream::StreamExt; let manager = SseManager::new(); @@ -319,34 +319,34 @@ async fn sse_scoped_event_only_delivered_to_target_user() { // Send event scoped to alice manager.broadcast_for_user( ALICE_USER_ID, - SseEvent::Status { + AppEvent::Status { message: "alice's event".to_string(), thread_id: None, }, ); // Send global heartbeat (both should get it) - manager.broadcast(SseEvent::Heartbeat); + manager.broadcast(AppEvent::Heartbeat); // Alice gets her scoped event first let e = alice_stream.next().await.unwrap(); match &e { - SseEvent::Status { message, .. } => assert_eq!(message, "alice's event"), + AppEvent::Status { message, .. } => assert_eq!(message, "alice's event"), _ => panic!("Expected Status, got {:?}", e), } // Alice also gets heartbeat let e = alice_stream.next().await.unwrap(); - assert!(matches!(e, SseEvent::Heartbeat)); + assert!(matches!(e, AppEvent::Heartbeat)); // Bob only gets the heartbeat (alice's event was filtered) let e = bob_stream.next().await.unwrap(); - assert!(matches!(e, SseEvent::Heartbeat)); + assert!(matches!(e, AppEvent::Heartbeat)); } #[tokio::test] async fn sse_global_event_delivered_to_all_users() { - use ironclaw::channels::web::types::SseEvent; + use ironclaw_common::AppEvent; use tokio_stream::StreamExt; let manager = SseManager::new(); @@ -361,7 +361,7 @@ async fn sse_global_event_delivered_to_all_users() { .expect("subscribe"), ); - manager.broadcast(SseEvent::Status { + manager.broadcast(AppEvent::Status { message: "global announcement".to_string(), thread_id: None, }); @@ -369,7 +369,7 @@ async fn sse_global_event_delivered_to_all_users() { let ea = alice.next().await.unwrap(); let eb = bob.next().await.unwrap(); match (&ea, &eb) { - (SseEvent::Status { message: a, .. }, SseEvent::Status { message: b, .. }) => { + (AppEvent::Status { message: a, .. }, AppEvent::Status { message: b, .. }) => { assert_eq!(a, "global announcement"); assert_eq!(b, "global announcement"); } @@ -379,7 +379,7 @@ async fn sse_global_event_delivered_to_all_users() { #[tokio::test] async fn sse_user_b_event_not_visible_to_user_a() { - use ironclaw::channels::web::types::SseEvent; + use ironclaw_common::AppEvent; use tokio_stream::StreamExt; let manager = SseManager::new(); @@ -392,19 +392,19 @@ async fn sse_user_b_event_not_visible_to_user_a() { // Send event for bob only manager.broadcast_for_user( BOB_USER_ID, - SseEvent::Response { + AppEvent::Response { content: "bob's secret".to_string(), thread_id: "t1".to_string(), }, ); // Send heartbeat so alice has something to receive - manager.broadcast(SseEvent::Heartbeat); + manager.broadcast(AppEvent::Heartbeat); // Alice should only get heartbeat, not bob's response let e = alice.next().await.unwrap(); assert!( - matches!(e, SseEvent::Heartbeat), + matches!(e, AppEvent::Heartbeat), "Expected Heartbeat, got {:?}", e ); @@ -412,7 +412,7 @@ async fn sse_user_b_event_not_visible_to_user_a() { #[tokio::test] async fn sse_unscoped_subscriber_receives_all_events() { - use ironclaw::channels::web::types::SseEvent; + use ironclaw_common::AppEvent; use tokio_stream::StreamExt; let manager = SseManager::new(); @@ -421,19 +421,19 @@ async fn sse_unscoped_subscriber_receives_all_events() { manager.broadcast_for_user( ALICE_USER_ID, - SseEvent::Status { + AppEvent::Status { message: "alice only".to_string(), thread_id: None, }, ); manager.broadcast_for_user( BOB_USER_ID, - SseEvent::Status { + AppEvent::Status { message: "bob only".to_string(), thread_id: None, }, ); - manager.broadcast(SseEvent::Heartbeat); + manager.broadcast(AppEvent::Heartbeat); // Unscoped subscriber gets ALL three events let e1 = stream.next().await.unwrap(); @@ -441,14 +441,14 @@ async fn sse_unscoped_subscriber_receives_all_events() { let e3 = stream.next().await.unwrap(); match &e1 { - SseEvent::Status { message, .. } => assert_eq!(message, "alice only"), + AppEvent::Status { message, .. } => assert_eq!(message, "alice only"), _ => panic!("Expected alice's Status"), } match &e2 { - SseEvent::Status { message, .. } => assert_eq!(message, "bob only"), + AppEvent::Status { message, .. } => assert_eq!(message, "bob only"), _ => panic!("Expected bob's Status"), } - assert!(matches!(e3, SseEvent::Heartbeat)); + assert!(matches!(e3, AppEvent::Heartbeat)); } // =========================================================================== @@ -767,7 +767,7 @@ async fn full_server_jobs_endpoint_rejected_without_auth() { #[tokio::test] async fn full_server_ws_multi_user_event_isolation() { use futures::StreamExt; - use ironclaw::channels::web::types::SseEvent; + use ironclaw_common::AppEvent; use tokio_tungstenite::tungstenite::Message; use tokio_tungstenite::tungstenite::client::IntoClientRequest; @@ -800,14 +800,14 @@ async fn full_server_ws_multi_user_event_isolation() { // Broadcast an event scoped to Alice only state.sse.broadcast_for_user( ALICE_USER_ID, - SseEvent::Status { + AppEvent::Status { message: "alice-only-event".to_string(), thread_id: None, }, ); // Broadcast a global heartbeat so Bob has something to receive - state.sse.broadcast(SseEvent::Heartbeat); + state.sse.broadcast(AppEvent::Heartbeat); // Alice should get her scoped event let alice_msg = tokio::time::timeout(Duration::from_secs(2), alice_ws.next()) diff --git a/tests/ws_gateway_integration.rs b/tests/ws_gateway_integration.rs index 43277389..9c4b8e76 100644 --- a/tests/ws_gateway_integration.rs +++ b/tests/ws_gateway_integration.rs @@ -22,8 +22,8 @@ use tokio_tungstenite::tungstenite::client::IntoClientRequest; use ironclaw::channels::IncomingMessage; use ironclaw::channels::web::server::{GatewayState, start_server}; use ironclaw::channels::web::sse::SseManager; -use ironclaw::channels::web::types::SseEvent; use ironclaw::channels::web::ws::WsConnectionTracker; +use ironclaw_common::AppEvent; const AUTH_TOKEN: &str = "test-token-12345"; const TIMEOUT: Duration = Duration::from_secs(5); @@ -164,7 +164,7 @@ async fn test_ws_broadcast_event_received() { tokio::time::sleep(Duration::from_millis(50)).await; // Broadcast an SSE event (simulates agent sending a response) - state.sse.broadcast(SseEvent::Response { + state.sse.broadcast(AppEvent::Response { content: "agent says hi".to_string(), thread_id: "t1".to_string(), }); @@ -185,7 +185,7 @@ async fn test_ws_thinking_event() { let mut ws = connect_ws(addr).await; tokio::time::sleep(Duration::from_millis(50)).await; - state.sse.broadcast(SseEvent::Thinking { + state.sse.broadcast(AppEvent::Thinking { message: "analyzing...".to_string(), thread_id: None, }); @@ -310,22 +310,22 @@ async fn test_ws_multiple_events_in_sequence() { tokio::time::sleep(Duration::from_millis(50)).await; // Broadcast multiple events rapidly - state.sse.broadcast(SseEvent::Thinking { + state.sse.broadcast(AppEvent::Thinking { message: "step 1".to_string(), thread_id: None, }); - state.sse.broadcast(SseEvent::ToolStarted { + state.sse.broadcast(AppEvent::ToolStarted { name: "shell".to_string(), thread_id: None, }); - state.sse.broadcast(SseEvent::ToolCompleted { + state.sse.broadcast(AppEvent::ToolCompleted { name: "shell".to_string(), success: true, error: None, parameters: None, thread_id: None, }); - state.sse.broadcast(SseEvent::Response { + state.sse.broadcast(AppEvent::Response { content: "done".to_string(), thread_id: "t1".to_string(), });