mirror of
https://github.com/outbackdingo/optimclaw.git
synced 2026-09-01 17:19:24 +00:00
refactor: extract AppEvent to crates/ironclaw_common
SseEvent was defined in src/channels/web/types.rs but imported by 12+ modules across agent, orchestrator, worker, tools, and extensions — it had become the application-wide event protocol, not a web transport concern. Create crates/ironclaw_common as a shared workspace crate and move the enum there as AppEvent. Also move the truncate_preview utility which was similarly leaked from the web gateway into agent modules. - New crate: crates/ironclaw_common (AppEvent, truncate_preview) - Rename SseEvent → AppEvent, from_sse_event → from_app_event - web/types.rs re-exports AppEvent for internal gateway use - web/util.rs re-exports truncate_preview - Wire format unchanged (serde renames are on variants, not the enum) Aligned with the event bus direction on refactor/architectural-hardening where DomainEvent (≡ AppEvent) is wrapped in a SystemEvent envelope. Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>
This commit is contained in:
@@ -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())
|
||||
|
||||
@@ -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(),
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user