refactor: extract AppEvent to crates/ironclaw_common (#1615)

* 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]>

* refactor: add AppEvent::event_type() helper, deduplicate match blocks

Address Gemini review: extract the variant→string match into a single
method on AppEvent, replacing the duplicated 22-arm matches in sse.rs
and types.rs.

Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>

* refactor: rename leftover sse vars/tests to match AppEvent rename

Address Copilot review: rename sse_event vars to app_event in
orchestrator/api.rs and ws.rs, rename test functions from
test_ws_server_from_sse_* to test_ws_server_from_app_event_*, and
update stale SSE comments.

Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>

* refactor: add Deserialize to AppEvent, round-trip test, fix stale comments

Address zmanian review:
- Add Deserialize derive to AppEvent so downstream consumers can
  deserialize incoming events
- Add event_type_matches_serde_type_field test that round-trips every
  variant through serde and asserts event_type() matches the serialized
  "type" field — catches drift between serde renames and the manual match
- Add round_trip_deserialize test for basic Serialize/Deserialize parity
- Update remaining "SSE" references in comments across server.rs,
  manager.rs, ws_gateway_integration.rs, and worker/job.rs

Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>

---------

Co-authored-by: Claude Opus 4.6 (1M context) <[email protected]>
This commit is contained in:
Illia Polosukhin
2026-03-24 23:02:46 -07:00
committed by GitHub
co-authored by Claude Opus 4.6
parent 656151783c
commit 706c3a1b47
24 changed files with 646 additions and 481 deletions
+23 -23
View File
@@ -307,7 +307,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();
@@ -325,34 +325,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();
@@ -367,7 +367,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,
});
@@ -375,7 +375,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");
}
@@ -385,7 +385,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();
@@ -398,19 +398,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
);
@@ -418,7 +418,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();
@@ -427,19 +427,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();
@@ -447,14 +447,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));
}
// ===========================================================================
@@ -881,7 +881,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;
@@ -914,14 +914,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())
+9 -9
View File
@@ -5,7 +5,7 @@
//! - WebSocket upgrade with auth
//! - Ping/pong
//! - Client message → agent msg_tx
//! - Broadcast SSE event → WebSocket client
//! - Broadcast AppEvent → WebSocket client
//! - Connection tracking (counter increment/decrement)
//! - Gateway status endpoint
@@ -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,8 +164,8 @@ async fn test_ws_broadcast_event_received() {
// Give the connection a moment to fully establish
tokio::time::sleep(Duration::from_millis(50)).await;
// Broadcast an SSE event (simulates agent sending a response)
state.sse.broadcast(SseEvent::Response {
// Broadcast an event (simulates agent sending a response)
state.sse.broadcast(AppEvent::Response {
content: "agent says hi".to_string(),
thread_id: "t1".to_string(),
});
@@ -186,7 +186,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,
});
@@ -311,22 +311,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(),
});