gitops/optimclaw
1
0
Fork 0
mirror of https://github.com/outbackdingo/optimclaw.git synced 2026-09-01 17:19:24 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/staging' into feat/multi-tenant-isolation-phases-2-4

Browse Source 
# Conflicts:
#	src/agent/agent_loop.rs
This commit is contained in:
[email protected]
2026-03-25 16:43:03 -07:00
parent 8f1e0df0cd 86d1143064
commit 3a945a81e7
72 changed files with 3879 additions and 748 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/channels/channel.rs
+16
View File
@@ -265,6 +265,15 @@ impl OutgoingResponse {
}
}
/// A single tool decision within a reasoning update.
#[derive(Debug, Clone)]
pub struct ToolDecision {
/// Tool name.
pub tool_name: String,
/// Agent's reasoning for choosing this tool.
pub rationale: String,
}
/// Status update types for showing agent activity.
#[derive(Debug, Clone)]
pub enum StatusUpdate {
@@ -333,6 +342,13 @@ pub enum StatusUpdate {
},
/// Suggested follow-up messages for the user.
Suggestions { suggestions: Vec<String> },
/// Agent reasoning update (why it chose specific tools).
ReasoningUpdate {
/// Human-readable summary of the agent's decision.
narrative: String,
/// Per-tool decisions.
decisions: Vec<ToolDecision>,
},
/// Per-turn token usage and cost summary (shown as subtle metadata).
TurnCost {
input_tokens: u64,
src/channels/mod.rs
+1 -1
View File
@@ -39,7 +39,7 @@ mod webhook_server;
pub use channel::{
AttachmentKind, Channel, ChannelSecretUpdater, IncomingAttachment, IncomingMessage,
MessageStream, OutgoingResponse, StatusUpdate, routing_target_from_metadata,
MessageStream, OutgoingResponse, StatusUpdate, ToolDecision, routing_target_from_metadata,
};
pub use http::{HttpChannel, HttpChannelState};
pub use manager::ChannelManager;
src/channels/repl.rs
+65 -9
View File
@@ -75,6 +75,7 @@ const SLASH_COMMANDS: &[&str] = &[
"/suggest",
"/thread",
"/resume",
"/reasoning",
];
/// Rustyline helper for slash-command tab completion.
@@ -430,6 +431,18 @@ impl ReplChannel {
let _ = execute!(stderr, terminal::Clear(terminal::ClearType::FromCursorDown));
}
}
async fn finish_single_message_turn(&self) {
if self.single_message.is_none() {
return;
}
let tx = self.msg_tx.lock().ok().and_then(|mut guard| guard.take());
if let Some(tx) = tx {
let msg = IncomingMessage::new("repl", &self.user_id, "/quit");
let _ = tx.send(msg).await;
}
}
}
impl Default for ReplChannel {
@@ -479,7 +492,9 @@ impl Channel for ReplChannel {
async fn start(&self) -> Result<MessageStream, ChannelError> {
let (tx, rx) = mpsc::channel(32);
// Store tx so send_status can inject approval responses directly
// Approval prompts inject responses back through this sender.
// In single-message mode we keep it until the turn finishes, then
// drop it after enqueuing /quit so the receiver stream can close.
if let Ok(mut guard) = self.msg_tx.lock() {
*guard = Some(tx.clone());
}
@@ -495,11 +510,10 @@ impl Channel for ReplChannel {
// Single message mode: send it and return
if let Some(msg) = single_message {
let incoming = IncomingMessage::new("repl", &user_id, &msg).with_timezone(&sys_tz);
let incoming = IncomingMessage::new("repl", &user_id, &msg)
.with_metadata(serde_json::json!({ "single_message_mode": true }))
.with_timezone(&sys_tz);
let _ = tx.blocking_send(incoming);
// Ensure the agent exits after handling exactly one turn in -m mode,
// even when other channels (gateway/http) are enabled.
let _ = tx.blocking_send(IncomingMessage::new("repl", &user_id, "/quit"));
return;
}
@@ -662,6 +676,7 @@ impl Channel for ReplChannel {
println!();
println!();
self.stdin_locked.store(false, Ordering::Relaxed);
self.finish_single_message_turn().await;
return Ok(());
}
@@ -680,6 +695,7 @@ impl Channel for ReplChannel {
println!();
// Unlock stdin so readline can resume
self.stdin_locked.store(false, Ordering::Relaxed);
self.finish_single_message_turn().await;
Ok(())
}
@@ -779,6 +795,7 @@ impl Channel for ReplChannel {
let msg_tx = Arc::clone(&self.msg_tx);
let user_id = self.user_id.clone();
let lock_flag = Arc::clone(&self.stdin_locked);
let single_message_mode = self.single_message.is_some();
tokio::task::spawn_blocking(move || {
let action = run_approval_selector(allow_always).unwrap_or("n");
// Unlock stdin so readline can resume after approval
@@ -787,7 +804,12 @@ impl Channel for ReplChannel {
return;
};
if let Some(tx) = guard.as_ref() {
let msg = IncomingMessage::new("repl", &user_id, action);
let msg = if single_message_mode {
IncomingMessage::new("repl", &user_id, action)
.with_metadata(serde_json::json!({ "single_message_mode": true }))
} else {
IncomingMessage::new("repl", &user_id, action)
};
let _ = tx.blocking_send(msg);
}
});
@@ -841,6 +863,19 @@ impl Channel for ReplChannel {
StatusUpdate::Suggestions { .. } => {
// Suggestions are only rendered by the web gateway
}
StatusUpdate::ReasoningUpdate {
narrative,
decisions,
} => {
if !narrative.is_empty() {
let display = truncate_for_preview(&narrative, CLI_STATUS_MAX);
eprintln!(" \x1b[94m\u{25B6} {display}\x1b[0m");
}
for d in &decisions {
let display = truncate_for_preview(&d.rationale, CLI_STATUS_MAX);
eprintln!(" \x1b[90m\u{2192} {}: {display}\x1b[0m", d.tool_name);
}
}
StatusUpdate::TurnCost { .. } => {
// Cost display is handled by the TUI channel
}
@@ -875,6 +910,7 @@ impl Channel for ReplChannel {
#[cfg(test)]
mod tests {
use futures::StreamExt;
use tokio::time::{Duration, timeout};
use super::*;
@@ -883,16 +919,36 @@ mod tests {
let repl = ReplChannel::with_message("hi".to_string());
let mut stream = repl.start().await.expect("repl start should succeed");
let first = stream.next().await.expect("first message missing");
let first = timeout(Duration::from_secs(1), stream.next())
.await
.expect("timed out waiting for first message")
.expect("first message missing");
assert_eq!(first.channel, "repl");
assert_eq!(first.content, "hi");
let second = stream.next().await.expect("quit message missing");
assert!(
timeout(Duration::from_millis(100), stream.next())
.await
.is_err(),
"single-message mode should wait for the turn to finish before quitting"
);
repl.respond(&first, OutgoingResponse::text("done"))
.await
.expect("respond should succeed");
let second = timeout(Duration::from_secs(1), stream.next())
.await
.expect("timed out waiting for quit message")
.expect("quit message missing");
assert_eq!(second.channel, "repl");
assert_eq!(second.content, "/quit");
assert!(
stream.next().await.is_none(),
timeout(Duration::from_secs(1), stream.next())
.await
.expect("timed out waiting for stream to close")
.is_none(),
"stream should end after /quit"
);
}
src/channels/wasm/wrapper.rs
+14
View File
@@ -3061,6 +3061,20 @@ fn status_to_wit(
},
// Suggestions and turn cost are web-gateway-only; skip for WASM channels
StatusUpdate::Suggestions { .. } | StatusUpdate::TurnCost { .. } => return None,
StatusUpdate::ReasoningUpdate {
narrative,
decisions,
} => {
let mut msg = narrative.clone();
for d in decisions {
msg.push_str(&format!("\n → {}: {}", d.tool_name, d.rationale));
}
wit_channel::StatusUpdate {
status: wit_channel::StatusType::Status,
message: msg,
metadata_json,
}
}
})
}
src/channels/web/handlers/chat.rs
+5 -3
View File
@@ -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,
@@ -398,8 +398,10 @@ pub async fn chat_history_handler(
truncate_preview(&s, 500)
}),
error: tc.error.clone(),
rationale: tc.rationale.clone(),
})
.collect(),
narrative: t.narrative.clone(),
})
.collect();
src/channels/web/mod.rs
+52 -19
View File
@@ -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 {
@@ -98,7 +98,8 @@ impl GatewayChannel {
job_manager: None,
prompt_queue: None,
scheduler: None,
default_user_id: config.user_id.clone(),
owner_id: config.user_id.clone(),
default_sender_id: config.user_id.clone(),
shutdown_tx: tokio::sync::RwLock::new(None),
ws_tracker: Some(Arc::new(ws::WsConnectionTracker::new())),
llm_provider: None,
@@ -121,6 +122,22 @@ impl GatewayChannel {
}
}
/// Rebind the single-user auth identity to the durable owner scope while
/// preserving the configured gateway sender/routing identity.
pub fn with_owner_scope(mut self, owner_id: impl Into<String>) -> Self {
let owner_id = owner_id.into();
let single_user_token = if self.config.user_tokens.is_none() {
self.auth.first_token().map(ToOwned::to_owned)
} else {
None
};
if let Some(token) = single_user_token {
self.auth = MultiAuthState::single(token, owner_id.clone());
}
self.rebuild_state(|s| s.owner_id = owner_id);
self
}
/// Create a gateway channel with a pre-built multi-user auth state.
pub fn new_multi_auth(config: GatewayConfig, auth: MultiAuthState) -> Self {
let state = Arc::new(GatewayState {
@@ -137,7 +154,8 @@ impl GatewayChannel {
job_manager: None,
prompt_queue: None,
scheduler: None,
default_user_id: config.user_id.clone(),
owner_id: config.user_id.clone(),
default_sender_id: config.user_id.clone(),
shutdown_tx: tokio::sync::RwLock::new(None),
ws_tracker: Some(Arc::new(ws::WsConnectionTracker::new())),
llm_provider: None,
@@ -177,7 +195,8 @@ impl GatewayChannel {
job_manager: self.state.job_manager.clone(),
prompt_queue: self.state.prompt_queue.clone(),
scheduler: self.state.scheduler.clone(),
default_user_id: self.state.default_user_id.clone(),
owner_id: self.state.owner_id.clone(),
default_sender_id: self.state.default_sender_id.clone(),
shutdown_tx: tokio::sync::RwLock::new(None),
ws_tracker: self.state.ws_tracker.clone(),
llm_provider: self.state.llm_provider.clone(),
@@ -367,7 +386,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 +405,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 +418,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 +442,7 @@ impl Channel for GatewayChannel {
job_id,
title,
browse_url,
} => SseEvent::JobStarted {
} => AppEvent::JobStarted {
job_id,
title,
browse_url,
@@ -434,7 +453,7 @@ impl Channel for GatewayChannel {
description,
parameters,
allow_always,
} => SseEvent::ApprovalNeeded {
} => AppEvent::ApprovalNeeded {
request_id,
tool_name,
description,
@@ -448,7 +467,7 @@ impl Channel for GatewayChannel {
instructions,
auth_url,
setup_url,
} => SseEvent::AuthRequired {
} => AppEvent::AuthRequired {
extension_name,
instructions,
auth_url,
@@ -458,25 +477,39 @@ 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: thread_id.clone(),
},
StatusUpdate::ReasoningUpdate {
narrative,
decisions,
} => AppEvent::ReasoningUpdate {
narrative,
decisions: decisions
.into_iter()
.map(|d| crate::channels::web::types::ToolDecisionDto {
tool_name: d.tool_name,
rationale: d.rationale,
})
.collect(),
thread_id,
},
StatusUpdate::TurnCost {
input_tokens,
output_tokens,
cost_usd,
} => SseEvent::TurnCost {
} => AppEvent::TurnCost {
input_tokens,
output_tokens,
cost_usd,
@@ -512,7 +545,7 @@ impl Channel for GatewayChannel {
};
self.state.sse.broadcast_for_user(
user_id,
SseEvent::Response {
AppEvent::Response {
content: response.content,
thread_id,
},
src/channels/web/openai_compat.rs
+2
View File
@@ -231,6 +231,7 @@ pub fn convert_messages(messages: &[OpenAiMessage]) -> Result<Vec<ChatMessage>,
name: tc.function.name.clone(),
arguments: serde_json::from_str(&tc.function.arguments)
.unwrap_or(serde_json::Value::Object(Default::default())),
reasoning: None,
})
.collect();
Ok(ChatMessage::assistant_with_tool_calls(
@@ -954,6 +955,7 @@ mod tests {
id: "call_abc".to_string(),
name: "search".to_string(),
arguments: serde_json::json!({"query": "rust"}),
reasoning: None,
}];
let converted = convert_tool_calls_to_openai(&calls);
src/channels/web/server.rs
+31 -27
View File
@@ -345,8 +345,10 @@ pub struct GatewayState {
pub job_manager: Option<Arc<ContainerJobManager>>,
/// Prompt queue for Claude Code follow-up prompts.
pub prompt_queue: Option<PromptQueue>,
/// Default user ID (fallback for non-request contexts like heartbeat/routines).
pub default_user_id: String,
/// Durable owner scope for persistence and unauthenticated callback flows.
pub owner_id: String,
/// Default sender/routing identity for gateway-originated messages.
pub default_sender_id: String,
/// Shutdown signal sender.
pub shutdown_tx: tokio::sync::RwLock<Option<oneshot::Sender<()>>>,
/// WebSocket connection tracker.
@@ -780,7 +782,7 @@ async fn oauth_callback_handler(
error = %error,
"OAuth callback received with malformed state"
);
clear_auth_mode(&state, &state.default_user_id).await;
clear_auth_mode(&state, &state.owner_id).await;
return oauth_error_page("IronClaw");
}
};
@@ -816,7 +818,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(),
@@ -954,11 +956,11 @@ async fn oauth_callback_handler(
message
};
// Broadcast SSE event to notify the web UI
// Broadcast event to notify the web UI
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(),
@@ -1141,7 +1143,7 @@ async fn slack_relay_oauth_callback_handler(
let state_key = format!("relay:{}:oauth_state", DEFAULT_RELAY_NAME);
let stored_state = match ext_mgr
.secrets()
.get_decrypted(&state.default_user_id, &state_key)
.get_decrypted(&state.owner_id, &state_key)
.await
{
Ok(secret) => secret.expose().to_string(),
@@ -1165,10 +1167,7 @@ async fn slack_relay_oauth_callback_handler(
}
// Delete the nonce (one-time use)
let _ = ext_mgr
.secrets()
.delete(&state.default_user_id, &state_key)
.await;
let _ = ext_mgr.secrets().delete(&state.owner_id, &state_key).await;
let result: Result<(), String> = async {
let store = state.store.as_ref().ok_or_else(|| {
@@ -1179,16 +1178,12 @@ async fn slack_relay_oauth_callback_handler(
// Store team_id in settings
let team_id_key = format!("relay:{}:team_id", DEFAULT_RELAY_NAME);
let _ = store
.set_setting(
&state.default_user_id,
&team_id_key,
&serde_json::json!(team_id),
)
.set_setting(&state.owner_id, &team_id_key, &serde_json::json!(team_id))
.await;
// Activate the relay channel
ext_mgr
.activate_stored_relay(DEFAULT_RELAY_NAME, &state.default_user_id)
.activate_stored_relay(DEFAULT_RELAY_NAME, &state.owner_id)
.await
.map_err(|e| format!("Failed to activate relay channel: {}", e))?;
@@ -1207,8 +1202,8 @@ async fn slack_relay_oauth_callback_handler(
}
};
// Broadcast SSE event to notify the web UI
state.sse.broadcast(SseEvent::AuthCompleted {
// Broadcast event to notify the web UI
state.sse.broadcast(AppEvent::AuthCompleted {
extension_name: DEFAULT_RELAY_NAME.to_string(),
success,
message: message.clone(),
@@ -1308,6 +1303,9 @@ async fn chat_send_handler(
}
let mut msg = IncomingMessage::new("gateway", &user.user_id, &req.content);
if state.owner_id != state.default_sender_id && user.user_id == state.owner_id {
msg = msg.with_sender_id(&state.default_sender_id);
}
// Prefer timezone from JSON body, fall back to X-Timezone header
let tz = req
.timezone
@@ -1409,6 +1407,9 @@ async fn chat_approval_handler(
})?;
let mut msg = IncomingMessage::new("gateway", &user.user_id, content);
if state.owner_id != state.default_sender_id && user.user_id == state.owner_id {
msg = msg.with_sender_id(&state.default_sender_id);
}
if let Some(ref thread_id) = req.thread_id {
msg = msg.with_thread(thread_id);
@@ -1475,7 +1476,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,
@@ -1488,7 +1489,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,
@@ -1497,7 +1498,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,
@@ -1513,7 +1514,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,
@@ -1729,8 +1730,10 @@ async fn chat_history_handler(
truncate_preview(&s, 500)
}),
error: tc.error.clone(),
rationale: tc.rationale.clone(),
})
.collect(),
narrative: t.narrative.clone(),
})
.collect();
@@ -2481,7 +2484,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(),
@@ -2981,7 +2984,8 @@ mod tests {
store: None,
job_manager: None,
prompt_queue: None,
default_user_id: "test".to_string(),
owner_id: "test".to_string(),
default_sender_id: "test".to_string(),
shutdown_tx: tokio::sync::RwLock::new(None),
ws_tracker: None,
llm_provider: None,
@@ -3172,7 +3176,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")
@@ -3454,7 +3458,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,
src/channels/web/sse.rs
+19 -42
View File
@@ -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<String>,
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<String>,
) -> Option<impl Stream<Item = SseEvent> + Send + 'static + use<>> {
) -> Option<impl Stream<Item = AppEvent> + 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);
@@ -186,30 +186,7 @@ impl SseManager {
return None;
}
};
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",
};
let event_type = event.event_type();
Some(Ok(Event::default().event(event_type).data(data)))
});
@@ -272,7 +249,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 +257,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 +276,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 +306,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 +350,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
}
}
src/channels/web/test_helpers.rs
+2 -1
View File
@@ -76,7 +76,8 @@ impl TestGatewayBuilder {
store: None,
job_manager: None,
prompt_queue: None,
default_user_id: self.user_id,
owner_id: self.user_id.clone(),
default_sender_id: self.user_id,
shutdown_tx: tokio::sync::RwLock::new(None),
ws_tracker: Some(Arc::new(WsConnectionTracker::new())),
llm_provider: self.llm_provider,
src/channels/web/tests/multi_tenant.rs
+38 -1
View File
@@ -16,6 +16,7 @@ use axum::routing::{delete, get, post};
use tower::ServiceExt;
use uuid::Uuid;
use crate::channels::web::GatewayChannel;
use crate::channels::web::auth::{
AuthenticatedUser, MultiAuthState, UserIdentity, auth_middleware,
};
@@ -23,6 +24,7 @@ use crate::channels::web::server::{
ActiveConfigSnapshot, GatewayState, PerUserRateLimiter, PromptQueue, RateLimiter, WorkspacePool,
};
use crate::channels::web::sse::SseManager;
use crate::config::GatewayConfig;
// ── Helpers ────────────────────────────────────────────────────────────
@@ -64,7 +66,8 @@ fn build_state(
store,
job_manager: None,
prompt_queue,
default_user_id: "test".to_string(),
owner_id: "test".to_string(),
default_sender_id: "test".to_string(),
shutdown_tx: tokio::sync::RwLock::new(None),
ws_tracker: None,
llm_provider: None,
@@ -82,6 +85,40 @@ fn build_state(
})
}
fn gateway_config() -> GatewayConfig {
GatewayConfig {
host: "127.0.0.1".to_string(),
port: 3000,
auth_token: Some("gateway-auth".to_string()),
user_id: "gateway-sender".to_string(),
workspace_read_scopes: Vec::new(),
memory_layers: Vec::new(),
user_tokens: None,
}
}
#[test]
fn with_owner_scope_updates_gateway_owner_scope_in_multi_user_mode() {
let mut gateway = GatewayChannel::new(gateway_config());
gateway.auth = two_user_auth();
gateway.config.user_tokens = Some(HashMap::new());
let gateway = gateway.with_owner_scope("owner-scope");
assert_eq!(gateway.state.owner_id, "owner-scope");
assert_eq!(gateway.state.default_sender_id, "gateway-sender");
let alice = gateway
.auth
.authenticate("tok-alice")
.expect("alice token should remain valid");
let bob = gateway
.auth
.authenticate("tok-bob")
.expect("bob token should remain valid");
assert_eq!(alice.user_id, "alice");
assert_eq!(bob.user_id, "bob");
}
/// Create a libSQL-backed test database in a temporary directory.
///
/// Returns the database and a `TempDir` guard — the database file is
src/channels/web/types.rs
+33 -206
View File
@@ -63,6 +63,9 @@ pub struct TurnInfo {
pub started_at: String,
pub completed_at: Option<String>,
pub tool_calls: Vec<ToolCallInfo>,
/// Agent's reasoning narrative for this turn.
#[serde(skip_serializing_if = "Option::is_none")]
pub narrative: Option<String>,
}
#[derive(Debug, Serialize)]
@@ -74,6 +77,9 @@ pub struct ToolCallInfo {
pub result_preview: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub error: Option<String>,
/// Agent's reasoning for choosing this tool.
#[serde(skip_serializing_if = "Option::is_none")]
pub rationale: Option<String>,
}
#[derive(Debug, Serialize)]
@@ -114,165 +120,9 @@ pub struct ApprovalRequest {
pub thread_id: Option<String>,
}
// --- 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<String>,
},
#[serde(rename = "tool_started")]
ToolStarted {
name: String,
#[serde(skip_serializing_if = "Option::is_none")]
thread_id: Option<String>,
},
#[serde(rename = "tool_completed")]
ToolCompleted {
name: String,
success: bool,
#[serde(skip_serializing_if = "Option::is_none")]
error: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
parameters: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
thread_id: Option<String>,
},
#[serde(rename = "tool_result")]
ToolResult {
name: String,
preview: String,
#[serde(skip_serializing_if = "Option::is_none")]
thread_id: Option<String>,
},
#[serde(rename = "stream_chunk")]
StreamChunk {
content: String,
#[serde(skip_serializing_if = "Option::is_none")]
thread_id: Option<String>,
},
#[serde(rename = "status")]
Status {
message: String,
#[serde(skip_serializing_if = "Option::is_none")]
thread_id: Option<String>,
},
#[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<String>,
/// 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<String>,
#[serde(skip_serializing_if = "Option::is_none")]
auth_url: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
setup_url: Option<String>,
},
#[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<String>,
},
#[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<String>,
#[serde(skip_serializing_if = "Option::is_none")]
fallback_deliverable: Option<serde_json::Value>,
},
/// An image was generated by a tool.
#[serde(rename = "image_generated")]
ImageGenerated {
data_url: String,
#[serde(skip_serializing_if = "Option::is_none")]
path: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
thread_id: Option<String>,
},
/// Suggested follow-up messages for the user.
#[serde(rename = "suggestions")]
Suggestions {
suggestions: Vec<String>,
#[serde(skip_serializing_if = "Option::is_none")]
thread_id: Option<String>,
},
/// 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<String>,
},
/// 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<String>,
},
}
pub use ironclaw_common::{AppEvent, ToolDecisionDto};
// --- Memory ---
@@ -784,32 +634,9 @@ pub enum WsServerMessage {
}
impl WsServerMessage {
/// Create a WsServerMessage from an SseEvent.
pub fn from_sse_event(event: &SseEvent) -> 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",
};
/// Create a WsServerMessage from an AppEvent.
pub fn from_app_event(event: &AppEvent) -> Self {
let event_type = event.event_type();
let data = serde_json::to_value(event).unwrap_or(serde_json::Value::Null);
WsServerMessage::Event {
event_type: event_type.to_string(),
@@ -1101,12 +928,12 @@ mod tests {
}
#[test]
fn test_ws_server_from_sse_response() {
let sse = SseEvent::Response {
fn test_ws_server_from_app_event_response() {
let event = AppEvent::Response {
content: "hello".to_string(),
thread_id: "t1".to_string(),
};
let ws = WsServerMessage::from_sse_event(&sse);
let ws = WsServerMessage::from_app_event(&event);
match ws {
WsServerMessage::Event { event_type, data } => {
assert_eq!(event_type, "response");
@@ -1118,12 +945,12 @@ mod tests {
}
#[test]
fn test_ws_server_from_sse_thinking() {
let sse = SseEvent::Thinking {
fn test_ws_server_from_app_event_thinking() {
let event = AppEvent::Thinking {
message: "reasoning...".to_string(),
thread_id: None,
};
let ws = WsServerMessage::from_sse_event(&sse);
let ws = WsServerMessage::from_app_event(&event);
match ws {
WsServerMessage::Event { event_type, data } => {
assert_eq!(event_type, "thinking");
@@ -1134,8 +961,8 @@ mod tests {
}
#[test]
fn test_ws_server_from_sse_approval_needed() {
let sse = SseEvent::ApprovalNeeded {
fn test_ws_server_from_app_event_approval_needed() {
let event = AppEvent::ApprovalNeeded {
request_id: "r1".to_string(),
tool_name: "shell".to_string(),
description: "Run ls".to_string(),
@@ -1143,7 +970,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(&event);
match ws {
WsServerMessage::Event { event_type, data } => {
assert_eq!(event_type, "approval_needed");
@@ -1155,9 +982,9 @@ mod tests {
}
#[test]
fn test_ws_server_from_sse_heartbeat() {
let sse = SseEvent::Heartbeat;
let ws = WsServerMessage::from_sse_event(&sse);
fn test_ws_server_from_app_event_heartbeat() {
let event = AppEvent::Heartbeat;
let ws = WsServerMessage::from_app_event(&event);
match ws {
WsServerMessage::Event { event_type, .. } => {
assert_eq!(event_type, "heartbeat");
@@ -1197,8 +1024,8 @@ mod tests {
}
#[test]
fn test_sse_auth_required_serialize() {
let event = SseEvent::AuthRequired {
fn test_app_event_auth_required_serialize() {
let event = AppEvent::AuthRequired {
extension_name: "notion".to_string(),
instructions: Some("Get your token from...".to_string()),
auth_url: None,
@@ -1214,8 +1041,8 @@ mod tests {
}
#[test]
fn test_sse_auth_completed_serialize() {
let event = SseEvent::AuthCompleted {
fn test_app_event_auth_completed_serialize() {
let event = AppEvent::AuthCompleted {
extension_name: "notion".to_string(),
success: true,
message: "notion authenticated (3 tools loaded)".to_string(),
@@ -1228,14 +1055,14 @@ mod tests {
}
#[test]
fn test_ws_server_from_sse_auth_required() {
let sse = SseEvent::AuthRequired {
fn test_ws_server_from_app_event_auth_required() {
let event = 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(&event);
match ws {
WsServerMessage::Event { event_type, data } => {
assert_eq!(event_type, "auth_required");
@@ -1246,13 +1073,13 @@ mod tests {
}
#[test]
fn test_ws_server_from_sse_auth_completed() {
let sse = SseEvent::AuthCompleted {
fn test_ws_server_from_app_event_auth_completed() {
let event = 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(&event);
match ws {
WsServerMessage::Event { event_type, data } => {
assert_eq!(event_type, "auth_completed");
src/channels/web/util.rs
+86 -115
View File
@@ -2,28 +2,21 @@
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 `<tool_output …>…</tool_output>` 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]);
pub use ironclaw_common::truncate_preview;
// Re-close <tool_output> if truncation cut through the closing tag.
if s.starts_with("<tool_output") && !result.ends_with("</tool_output>") {
result.push_str("\n</tool_output>");
}
result
/// Parse tool call summary JSON objects into `ToolCallInfo` structs.
fn parse_tool_call_infos(calls: &[serde_json::Value]) -> Vec<ToolCallInfo> {
calls
.iter()
.map(|c| ToolCallInfo {
name: c["name"].as_str().unwrap_or("unknown").to_string(),
has_result: c.get("result_preview").is_some_and(|v| !v.is_null()),
has_error: c.get("error").is_some_and(|v| !v.is_null()),
result_preview: c["result_preview"].as_str().map(String::from),
error: c["error"].as_str().map(String::from),
rationale: c["rationale"].as_str().map(String::from),
})
.collect()
}
/// Build TurnInfo pairs from flat DB messages (user/tool_calls/assistant triples).
@@ -49,6 +42,7 @@ pub fn build_turns_from_db_messages(
started_at: msg.created_at.to_rfc3339(),
completed_at: None,
tool_calls: Vec::new(),
narrative: None,
};
// Check if next message is a tool_calls record
@@ -56,18 +50,28 @@ pub fn build_turns_from_db_messages(
&& next.role == "tool_calls"
{
let tc_msg = iter.next().expect("peeked");
match serde_json::from_str::<Vec<serde_json::Value>>(&tc_msg.content) {
Ok(calls) => {
turn.tool_calls = calls
.iter()
.map(|c| ToolCallInfo {
name: c["name"].as_str().unwrap_or("unknown").to_string(),
has_result: c.get("result_preview").is_some(),
has_error: c.get("error").is_some(),
result_preview: c["result_preview"].as_str().map(String::from),
error: c["error"].as_str().map(String::from),
})
.collect();
// Parse tool_calls JSON — supports two formats:
// safety: no byte-index slicing; comment describes JSON shape
match serde_json::from_str::<serde_json::Value>(&tc_msg.content) {
Ok(serde_json::Value::Array(calls)) => {
// Old format: plain array
turn.tool_calls = parse_tool_call_infos(&calls);
}
Ok(serde_json::Value::Object(obj)) => {
// New wrapped format with narrative
turn.narrative = obj
.get("narrative")
.and_then(|v| v.as_str())
.map(String::from);
if let Some(serde_json::Value::Array(calls)) = obj.get("calls") {
turn.tool_calls = parse_tool_call_infos(calls);
}
}
Ok(_) => {
tracing::warn!(
message_id = %tc_msg.id,
"Unexpected tool_calls JSON shape in DB, skipping"
);
}
Err(e) => {
tracing::warn!(
@@ -105,6 +109,7 @@ pub fn build_turns_from_db_messages(
started_at: msg.created_at.to_rfc3339(),
completed_at: Some(msg.created_at.to_rfc3339()),
tool_calls: Vec::new(),
narrative: None,
});
turn_number += 1;
}
@@ -118,88 +123,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 = "<tool_output name=\"search\">\nSome very long content here\n</tool_output>";
// Truncate so it cuts before the closing tag
let result = truncate_preview(s, 60);
assert!(result.ends_with("</tool_output>"));
assert!(result.contains("..."));
}
#[test]
fn test_truncate_preview_no_extra_close_when_intact() {
let s = "<tool_output name=\"echo\">\nshort\n</tool_output>";
// 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("</tool_output>").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("</tool_output>"));
}
// ---- build_turns_from_db_messages tests ----
fn make_msg(role: &str, content: &str, offset_ms: i64) -> crate::history::ConversationMessage {
@@ -305,4 +228,52 @@ mod tests {
assert!(turns[0].tool_calls.is_empty());
assert_eq!(turns[0].state, "Completed");
}
#[test]
fn test_build_turns_with_wrapped_tool_calls_format() {
let tc_json = serde_json::json!({
"narrative": "Searching memory for context before proceeding.",
"calls": [
{"name": "memory_search", "result_preview": "found 3 items", "rationale": "consult prior context"},
{"name": "shell", "error": "permission denied"}
]
});
let messages = vec![
make_msg("user", "Find info", 0),
make_msg("tool_calls", &tc_json.to_string(), 500),
make_msg("assistant", "Here's what I found", 1000),
];
let turns = build_turns_from_db_messages(&messages);
assert_eq!(turns.len(), 1);
assert_eq!(
turns[0].narrative.as_deref(),
Some("Searching memory for context before proceeding.")
);
assert_eq!(turns[0].tool_calls.len(), 2);
assert_eq!(turns[0].tool_calls[0].name, "memory_search");
assert_eq!(
turns[0].tool_calls[0].rationale.as_deref(),
Some("consult prior context")
);
assert!(turns[0].tool_calls[0].has_result);
assert_eq!(turns[0].tool_calls[1].name, "shell");
assert!(turns[0].tool_calls[1].has_error);
assert_eq!(turns[0].response.as_deref(), Some("Here's what I found"));
}
#[test]
fn test_build_turns_wrapped_format_without_narrative() {
let tc_json = serde_json::json!({
"calls": [{"name": "echo", "result_preview": "hello"}]
});
let messages = vec![
make_msg("user", "Say hi", 0),
make_msg("tool_calls", &tc_json.to_string(), 500),
make_msg("assistant", "Done", 1000),
];
let turns = build_turns_from_db_messages(&messages);
assert_eq!(turns.len(), 1);
assert!(turns[0].narrative.is_none());
assert_eq!(turns[0].tool_calls.len(), 1);
}
}
src/channels/web/ws.rs
+6 -5
View File
@@ -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(app_event) => WsServerMessage::from_app_event(&app_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,
@@ -520,7 +520,8 @@ mod tests {
job_manager: None,
prompt_queue: None,
scheduler: None,
default_user_id: "test".to_string(),
owner_id: "test".to_string(),
default_sender_id: "test".to_string(),
shutdown_tx: tokio::sync::RwLock::new(None),
ws_tracker: Some(Arc::new(WsConnectionTracker::new())),
llm_provider: None,