gitops/optimclaw
1
0
Fork 0
mirror of https://github.com/outbackdingo/optimclaw.git synced 2026-08-26 15:40:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: multi-tenant auth with per-user scoping

Browse Source 
Multi-user authentication and authorization for IronClaw gateway:
- Token-based auth mapping tokens to user IDs via GATEWAY_USER_TOKENS
- Per-user SSE broadcast scoping
- Per-user rate limiting with poisoned lock recovery
- Handler auth and ownership checks for jobs, settings, routines
- Extension secrets scoped per-user
- Chat handlers use authenticated identity
- Reverse proxy deployment documentation
- Comprehensive integration tests for auth, SSE, rate limiting, and job isolation
This commit is contained in:
Andrew Preece
2026-03-23 18:01:00 +00:00
parent 485d1568c4
commit b394c0cfdd
31 changed files with 3167 additions and 696 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/agent/job_monitor.rs
+20 -12
View File
@@ -44,7 +44,7 @@ pub struct JobMonitorRoute {
/// the main agent's context window).
pub fn spawn_job_monitor(
job_id: Uuid,
event_rx: broadcast::Receiver<(Uuid, SseEvent)>,
event_rx: broadcast::Receiver<(Uuid, String, SseEvent)>,
inject_tx: mpsc::Sender<IncomingMessage>,
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, SseEvent)>,
mut event_rx: broadcast::Receiver<(Uuid, String, SseEvent)>,
inject_tx: mpsc::Sender<IncomingMessage>,
route: JobMonitorRoute,
context_manager: Option<Arc<ContextManager>>,
@@ -68,7 +68,7 @@ pub fn spawn_job_monitor_with_context(
loop {
match event_rx.recv().await {
Ok((ev_job_id, event)) => {
Ok((ev_job_id, _user_id, event)) => {
if ev_job_id != job_id {
continue;
}
@@ -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, SseEvent)>,
mut event_rx: broadcast::Receiver<(Uuid, String, SseEvent)>,
context_manager: Arc<ContextManager>,
) -> 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, SseEvent::JobResult { status, .. })) if ev_job_id == job_id => {
Ok((ev_job_id, _user_id, SseEvent::JobResult { status, .. })) if ev_job_id == job_id => {
let target = if status == "completed" {
JobState::Completed
} else {
@@ -227,7 +227,7 @@ mod tests {
#[tokio::test]
async fn test_monitor_forwards_assistant_messages() {
let (event_tx, _) = broadcast::channel::<(Uuid, SseEvent)>(16);
let (event_tx, _) = broadcast::channel::<(Uuid, String, SseEvent)>(16);
let (inject_tx, mut inject_rx) = mpsc::channel::<IncomingMessage>(16);
let job_id = Uuid::new_v4();
@@ -237,6 +237,7 @@ mod tests {
event_tx
.send((
job_id,
"test-user".to_string(),
SseEvent::JobMessage {
job_id: job_id.to_string(),
role: "assistant".to_string(),
@@ -259,7 +260,7 @@ mod tests {
#[tokio::test]
async fn test_monitor_ignores_other_jobs() {
let (event_tx, _) = broadcast::channel::<(Uuid, SseEvent)>(16);
let (event_tx, _) = broadcast::channel::<(Uuid, String, SseEvent)>(16);
let (inject_tx, mut inject_rx) = mpsc::channel::<IncomingMessage>(16);
let job_id = Uuid::new_v4();
@@ -270,6 +271,7 @@ mod tests {
event_tx
.send((
other_job_id,
"test-user".to_string(),
SseEvent::JobMessage {
job_id: other_job_id.to_string(),
role: "assistant".to_string(),
@@ -289,7 +291,7 @@ mod tests {
#[tokio::test]
async fn test_monitor_exits_on_job_result() {
let (event_tx, _) = broadcast::channel::<(Uuid, SseEvent)>(16);
let (event_tx, _) = broadcast::channel::<(Uuid, String, SseEvent)>(16);
let (inject_tx, mut inject_rx) = mpsc::channel::<IncomingMessage>(16);
let job_id = Uuid::new_v4();
@@ -299,6 +301,7 @@ mod tests {
event_tx
.send((
job_id,
"test-user".to_string(),
SseEvent::JobResult {
job_id: job_id.to_string(),
status: "completed".to_string(),
@@ -324,7 +327,7 @@ mod tests {
#[tokio::test]
async fn test_monitor_skips_tool_events() {
let (event_tx, _) = broadcast::channel::<(Uuid, SseEvent)>(16);
let (event_tx, _) = broadcast::channel::<(Uuid, String, SseEvent)>(16);
let (inject_tx, mut inject_rx) = mpsc::channel::<IncomingMessage>(16);
let job_id = Uuid::new_v4();
@@ -334,6 +337,7 @@ mod tests {
event_tx
.send((
job_id,
"test-user".to_string(),
SseEvent::JobToolUse {
job_id: job_id.to_string(),
tool_name: "shell".to_string(),
@@ -346,6 +350,7 @@ mod tests {
event_tx
.send((
job_id,
"test-user".to_string(),
SseEvent::JobMessage {
job_id: job_id.to_string(),
role: "user".to_string(),
@@ -402,7 +407,7 @@ mod tests {
.await
.unwrap();
let (event_tx, _) = broadcast::channel::<(Uuid, SseEvent)>(16);
let (event_tx, _) = broadcast::channel::<(Uuid, String, SseEvent)>(16);
let (inject_tx, mut inject_rx) = mpsc::channel::<IncomingMessage>(16);
let handle = spawn_job_monitor_with_context(
@@ -417,6 +422,7 @@ mod tests {
event_tx
.send((
job_id,
"test-user".to_string(),
SseEvent::JobResult {
job_id: job_id.to_string(),
status: "completed".to_string(),
@@ -450,7 +456,7 @@ mod tests {
.await
.unwrap();
let (event_tx, _) = broadcast::channel::<(Uuid, SseEvent)>(16);
let (event_tx, _) = broadcast::channel::<(Uuid, String, SseEvent)>(16);
let (inject_tx, mut inject_rx) = mpsc::channel::<IncomingMessage>(16);
let handle = spawn_job_monitor_with_context(
@@ -465,6 +471,7 @@ mod tests {
event_tx
.send((
job_id,
"test-user".to_string(),
SseEvent::JobResult {
job_id: job_id.to_string(),
status: "failed".to_string(),
@@ -498,12 +505,13 @@ mod tests {
.await
.unwrap();
let (event_tx, _) = broadcast::channel::<(Uuid, SseEvent)>(16);
let (event_tx, _) = broadcast::channel::<(Uuid, String, SseEvent)>(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 {
job_id: job_id.to_string(),
status: "completed".to_string(),
src/agent/thread_ops.rs
+1 -1
View File
@@ -1646,7 +1646,7 @@ impl Agent {
};
match ext_mgr
.configure_token(&pending.extension_name, token)
.configure_token(&pending.extension_name, token, &message.user_id)
.await
{
Ok(result) if result.activated => {
src/channels/web/auth.rs
+360 -22
View File
@@ -1,17 +1,107 @@
//! Bearer token authentication middleware for the web gateway.
//!
//! Supports multi-user mode: each token maps to a `UserIdentity` that carries
//! the user_id. The identity is inserted into request extensions so downstream
//! handlers can extract it via `AuthenticatedUser`.
use std::collections::HashMap;
use axum::{
extract::{Request, State},
http::{HeaderMap, Method, StatusCode},
extract::{FromRequestParts, Request, State},
http::{HeaderMap, Method, StatusCode, request::Parts},
middleware::Next,
response::{IntoResponse, Response},
};
use subtle::ConstantTimeEq;
/// Shared auth state injected via axum middleware state.
/// Identity resolved from a bearer token.
#[derive(Debug, Clone)]
pub struct UserIdentity {
pub user_id: String,
/// Additional user scopes this identity can read from.
pub workspace_read_scopes: Vec<String>,
}
/// Multi-user auth state: maps tokens to user identities.
///
/// In single-user mode (the default), contains exactly one entry.
#[derive(Clone)]
pub struct AuthState {
pub token: String,
pub struct MultiAuthState {
tokens: HashMap<String, UserIdentity>,
}
impl MultiAuthState {
/// Create a single-user auth state (backwards compatible).
pub fn single(token: String, user_id: String) -> Self {
let mut tokens = HashMap::new();
tokens.insert(
token,
UserIdentity {
user_id,
workspace_read_scopes: Vec::new(),
},
);
Self { tokens }
}
/// Create a multi-user auth state from a map of tokens to identities.
pub fn multi(tokens: HashMap<String, UserIdentity>) -> Self {
Self { tokens }
}
/// Authenticate a token, returning the associated identity if valid.
///
/// Uses constant-time comparison (`subtle::ConstantTimeEq`) to prevent
/// timing side-channels that could leak token information. Iterates all
/// entries regardless of match to avoid early-exit timing differences.
/// O(n) in the number of configured users — negligible for typical
/// deployments (< 10 users).
pub fn authenticate(&self, candidate: &str) -> Option<&UserIdentity> {
let candidate_bytes = candidate.as_bytes();
let mut matched: Option<&UserIdentity> = None;
for (token, identity) in &self.tokens {
let token_bytes = token.as_bytes();
// ct_eq requires equal lengths; pad comparison to avoid length leak
if candidate_bytes.len() == token_bytes.len()
&& bool::from(candidate_bytes.ct_eq(token_bytes))
{
matched = Some(identity);
}
}
matched
}
/// Get the first token (for backwards-compatible printing at startup).
pub fn first_token(&self) -> Option<&str> {
self.tokens.keys().next().map(|s| s.as_str())
}
/// Get the first user identity (for single-user fallback).
pub fn first_identity(&self) -> Option<&UserIdentity> {
self.tokens.values().next()
}
}
/// Axum extractor that provides the authenticated user identity.
///
/// Only available on routes behind `auth_middleware`. Extracts the
/// `UserIdentity` that the middleware inserted into request extensions.
pub struct AuthenticatedUser(pub UserIdentity);
impl<S> FromRequestParts<S> for AuthenticatedUser
where
S: Send + Sync,
{
type Rejection = (StatusCode, &'static str);
async fn from_request_parts(parts: &mut Parts, _state: &S) -> Result<Self, Self::Rejection> {
parts
.extensions
.get::<UserIdentity>()
.cloned()
.map(AuthenticatedUser)
.ok_or((StatusCode::UNAUTHORIZED, "Not authenticated"))
}
}
/// Whether query-string token auth is allowed for this request.
@@ -51,47 +141,101 @@ fn query_token(request: &Request) -> Option<String> {
/// Auth middleware that validates bearer token from header or query param.
///
/// SSE connections can't set headers from `EventSource`, so we also accept
/// `?token=xxx` as a query parameter, but only on SSE endpoints.
/// `?token=xxx` as a query parameter, but only on SSE/WS endpoints.
///
/// On successful authentication, inserts the matching `UserIdentity` into
/// request extensions for downstream extraction via `AuthenticatedUser`.
pub async fn auth_middleware(
State(auth): State<AuthState>,
State(auth): State<MultiAuthState>,
headers: HeaderMap,
request: Request,
mut request: Request,
next: Next,
) -> Response {
// Try Authorization header first (constant-time comparison).
// Try Authorization header first.
// RFC 6750 Section 2.1: auth-scheme comparison is case-insensitive.
if let Some(auth_header) = headers.get("authorization")
&& let Ok(value) = auth_header.to_str()
&& value.len() > 7
&& value[..7].eq_ignore_ascii_case("Bearer ")
&& bool::from(value.as_bytes()[7..].ct_eq(auth.token.as_bytes()))
&& let Some(identity) = auth.authenticate(&value[7..])
{
request.extensions_mut().insert(identity.clone());
return next.run(request).await;
}
// Fall back to query parameter, but only for SSE endpoints (constant-time comparison).
// Fall back to query parameter, but only for SSE/WS endpoints.
if allows_query_token_auth(&request)
&& let Some(token) = query_token(&request)
&& bool::from(token.as_bytes().ct_eq(auth.token.as_bytes()))
&& let Some(identity) = auth.authenticate(&token)
{
request.extensions_mut().insert(identity.clone());
return next.run(request).await;
}
(StatusCode::UNAUTHORIZED, "Invalid or missing auth token").into_response()
}
// Keep the old type as an alias for any external references during migration.
pub type AuthState = MultiAuthState;
#[cfg(test)]
mod tests {
use super::*;
use crate::testing::credentials::{TEST_AUTH_SECRET_TOKEN, TEST_BEARER_TOKEN};
use crate::testing::credentials::TEST_AUTH_SECRET_TOKEN;
#[test]
fn test_auth_state_clone() {
let state = AuthState {
token: TEST_BEARER_TOKEN.to_string(),
};
let cloned = state.clone();
assert_eq!(cloned.token, TEST_BEARER_TOKEN);
fn test_multi_auth_state_single() {
let state = MultiAuthState::single("tok-123".to_string(), "alice".to_string());
let identity = state.authenticate("tok-123");
assert!(identity.is_some());
assert_eq!(identity.unwrap().user_id, "alice");
}
#[test]
fn test_multi_auth_state_reject_wrong_token() {
let state = MultiAuthState::single("tok-123".to_string(), "alice".to_string());
assert!(state.authenticate("wrong-token").is_none());
}
#[test]
fn test_multi_auth_state_multi_users() {
let mut tokens = HashMap::new();
tokens.insert(
"tok-alice".to_string(),
UserIdentity {
user_id: "alice".to_string(),
workspace_read_scopes: Vec::new(),
},
);
tokens.insert(
"tok-bob".to_string(),
UserIdentity {
user_id: "bob".to_string(),
workspace_read_scopes: Vec::new(),
},
);
let state = MultiAuthState::multi(tokens);
let alice = state.authenticate("tok-alice").unwrap();
assert_eq!(alice.user_id, "alice");
let bob = state.authenticate("tok-bob").unwrap();
assert_eq!(bob.user_id, "bob");
assert!(state.authenticate("tok-charlie").is_none());
}
#[test]
fn test_multi_auth_state_first_token() {
let state = MultiAuthState::single("my-token".to_string(), "user1".to_string());
assert_eq!(state.first_token(), Some("my-token"));
}
#[test]
fn test_multi_auth_state_first_identity() {
let state = MultiAuthState::single("my-token".to_string(), "user1".to_string());
let identity = state.first_identity().unwrap();
assert_eq!(identity.user_id, "user1");
}
use axum::Router;
@@ -107,9 +251,7 @@ mod tests {
/// Router with streaming endpoints (query auth allowed) and regular
/// endpoints (query auth rejected).
fn test_app(token: &str) -> Router {
let state = AuthState {
token: token.to_string(),
};
let state = MultiAuthState::single(token.to_string(), "test-user".to_string());
Router::new()
.route("/api/chat/events", get(dummy_handler))
.route("/api/logs/events", get(dummy_handler))
@@ -306,4 +448,200 @@ mod tests {
let resp = app.oneshot(req).await.unwrap();
assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
}
// --- Multi-tenant auth integration tests ---
/// Handler that extracts `AuthenticatedUser` and returns the resolved user_id.
async fn identity_handler(AuthenticatedUser(identity): AuthenticatedUser) -> String {
identity.user_id
}
/// Handler that extracts `AuthenticatedUser` and returns workspace_read_scopes as JSON.
async fn scopes_handler(AuthenticatedUser(identity): AuthenticatedUser) -> String {
serde_json::to_string(&identity.workspace_read_scopes).unwrap()
}
/// Build a multi-user router where each token maps to a distinct identity.
fn multi_user_app(tokens: HashMap<String, UserIdentity>) -> Router {
let state = MultiAuthState::multi(tokens);
Router::new()
.route("/api/chat/events", get(identity_handler))
.route("/api/chat/send", post(identity_handler))
.route("/api/scopes", get(scopes_handler))
.layer(middleware::from_fn_with_state(state, auth_middleware))
}
fn two_user_tokens() -> HashMap<String, UserIdentity> {
let mut tokens = HashMap::new();
tokens.insert(
"tok-alice".to_string(),
UserIdentity {
user_id: "alice".to_string(),
workspace_read_scopes: vec!["shared".to_string()],
},
);
tokens.insert(
"tok-bob".to_string(),
UserIdentity {
user_id: "bob".to_string(),
workspace_read_scopes: vec!["shared".to_string(), "alice".to_string()],
},
);
tokens
}
#[tokio::test]
async fn test_multi_user_alice_token_resolves_to_alice() {
let app = multi_user_app(two_user_tokens());
let req = Request::builder()
.uri("/api/chat/events")
.header("Authorization", "Bearer tok-alice")
.body(Body::empty())
.unwrap();
let resp = app.oneshot(req).await.unwrap();
assert_eq!(resp.status(), StatusCode::OK);
let body = axum::body::to_bytes(resp.into_body(), 1024).await.unwrap();
assert_eq!(body, "alice");
}
#[tokio::test]
async fn test_multi_user_bob_token_resolves_to_bob() {
let app = multi_user_app(two_user_tokens());
let req = Request::builder()
.uri("/api/chat/events")
.header("Authorization", "Bearer tok-bob")
.body(Body::empty())
.unwrap();
let resp = app.oneshot(req).await.unwrap();
assert_eq!(resp.status(), StatusCode::OK);
let body = axum::body::to_bytes(resp.into_body(), 1024).await.unwrap();
assert_eq!(body, "bob");
}
#[tokio::test]
async fn test_multi_user_sequential_tokens_resolve_independently() {
// Send both alice and bob tokens sequentially and verify each gets
// the correct identity — guards against token map corruption.
let tokens = two_user_tokens();
let app1 = multi_user_app(tokens.clone());
let req = Request::builder()
.uri("/api/chat/events")
.header("Authorization", "Bearer tok-alice")
.body(Body::empty())
.unwrap();
let resp = app1.oneshot(req).await.unwrap();
let body = axum::body::to_bytes(resp.into_body(), 1024).await.unwrap();
assert_eq!(body, "alice");
let app2 = multi_user_app(tokens);
let req = Request::builder()
.uri("/api/chat/events")
.header("Authorization", "Bearer tok-bob")
.body(Body::empty())
.unwrap();
let resp = app2.oneshot(req).await.unwrap();
let body = axum::body::to_bytes(resp.into_body(), 1024).await.unwrap();
assert_eq!(body, "bob");
}
#[tokio::test]
async fn test_multi_user_unknown_token_rejected() {
let app = multi_user_app(two_user_tokens());
let req = Request::builder()
.uri("/api/chat/events")
.header("Authorization", "Bearer tok-charlie")
.body(Body::empty())
.unwrap();
let resp = app.oneshot(req).await.unwrap();
assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
}
#[tokio::test]
async fn test_multi_user_workspace_read_scopes_propagated() {
let app = multi_user_app(two_user_tokens());
// Alice has ["shared"]
let req = Request::builder()
.uri("/api/scopes")
.header("Authorization", "Bearer tok-alice")
.body(Body::empty())
.unwrap();
let resp = app.oneshot(req).await.unwrap();
let body = axum::body::to_bytes(resp.into_body(), 1024).await.unwrap();
let scopes: Vec<String> = serde_json::from_slice(&body).unwrap();
assert_eq!(scopes, vec!["shared"]);
}
#[tokio::test]
async fn test_multi_user_bob_has_two_scopes() {
let app = multi_user_app(two_user_tokens());
// Bob has ["shared", "alice"]
let req = Request::builder()
.uri("/api/scopes")
.header("Authorization", "Bearer tok-bob")
.body(Body::empty())
.unwrap();
let resp = app.oneshot(req).await.unwrap();
let body = axum::body::to_bytes(resp.into_body(), 1024).await.unwrap();
let scopes: Vec<String> = serde_json::from_slice(&body).unwrap();
assert_eq!(scopes, vec!["shared", "alice"]);
}
#[tokio::test]
async fn test_multi_user_query_param_resolves_correct_identity() {
let app = multi_user_app(two_user_tokens());
let req = Request::builder()
.uri("/api/chat/events?token=tok-bob")
.body(Body::empty())
.unwrap();
let resp = app.oneshot(req).await.unwrap();
assert_eq!(resp.status(), StatusCode::OK);
let body = axum::body::to_bytes(resp.into_body(), 1024).await.unwrap();
assert_eq!(body, "bob");
}
#[tokio::test]
async fn test_multi_user_post_with_bearer_resolves_identity() {
let app = multi_user_app(two_user_tokens());
let req = Request::builder()
.method(Method::POST)
.uri("/api/chat/send")
.header("Authorization", "Bearer tok-alice")
.body(Body::empty())
.unwrap();
let resp = app.oneshot(req).await.unwrap();
assert_eq!(resp.status(), StatusCode::OK);
let body = axum::body::to_bytes(resp.into_body(), 1024).await.unwrap();
assert_eq!(body, "alice");
}
#[tokio::test]
async fn test_multi_user_empty_scopes_for_single_user() {
// Single-user mode creates identity with empty workspace_read_scopes.
let state = MultiAuthState::single("tok-only".to_string(), "solo".to_string());
let app = Router::new()
.route("/api/scopes", get(scopes_handler))
.layer(middleware::from_fn_with_state(state, auth_middleware));
let req = Request::builder()
.uri("/api/scopes")
.header("Authorization", "Bearer tok-only")
.body(Body::empty())
.unwrap();
let resp = app.oneshot(req).await.unwrap();
let body = axum::body::to_bytes(resp.into_body(), 1024).await.unwrap();
let scopes: Vec<String> = serde_json::from_slice(&body).unwrap();
assert!(scopes.is_empty());
}
#[tokio::test]
async fn test_prefix_and_extension_tokens_rejected() {
// Verifies that prefix/suffix variants of valid tokens are rejected.
// Note: the constant-time property is enforced structurally by use of
// subtle::ConstantTimeEq and cannot be verified via outcome testing.
let state = MultiAuthState::single("long-secret-token".to_string(), "user".to_string());
assert!(state.authenticate("long-secret").is_none());
assert!(state.authenticate("long-secret-token-extra").is_none());
}
}
src/channels/web/handlers/chat.rs
+30 -18
View File
@@ -12,22 +12,24 @@ use serde::Deserialize;
use uuid::Uuid;
use crate::channels::IncomingMessage;
use crate::channels::web::auth::AuthenticatedUser;
use crate::channels::web::server::GatewayState;
use crate::channels::web::types::*;
use crate::channels::web::util::{build_turns_from_db_messages, truncate_preview};
pub async fn chat_send_handler(
State(state): State<Arc<GatewayState>>,
AuthenticatedUser(identity): AuthenticatedUser,
Json(req): Json<SendMessageRequest>,
) -> Result<(StatusCode, Json<SendMessageResponse>), (StatusCode, String)> {
if !state.chat_rate_limiter.check() {
if !state.chat_rate_limiter.check(&identity.user_id) {
return Err((
StatusCode::TOO_MANY_REQUESTS,
"Rate limit exceeded. Try again shortly.".to_string(),
));
}
let mut msg = IncomingMessage::new("gateway", &state.user_id, &req.content);
let mut msg = IncomingMessage::new("gateway", &identity.user_id, &req.content);
if let Some(ref thread_id) = req.thread_id {
msg = msg.with_thread(thread_id);
@@ -74,6 +76,7 @@ pub async fn chat_send_handler(
pub async fn chat_approval_handler(
State(state): State<Arc<GatewayState>>,
AuthenticatedUser(identity): AuthenticatedUser,
Json(req): Json<ApprovalRequest>,
) -> Result<(StatusCode, Json<SendMessageResponse>), (StatusCode, String)> {
let (approved, always) = match req.action.as_str() {
@@ -109,7 +112,7 @@ pub async fn chat_approval_handler(
)
})?;
let mut msg = IncomingMessage::new("gateway", &state.user_id, content);
let mut msg = IncomingMessage::new("gateway", &identity.user_id, content);
if let Some(ref thread_id) = req.thread_id {
msg = msg.with_thread(thread_id);
@@ -150,6 +153,7 @@ pub async fn chat_approval_handler(
/// The token never touches the LLM, chat history, or SSE stream.
pub async fn chat_auth_token_handler(
State(state): State<Arc<GatewayState>>,
AuthenticatedUser(user): AuthenticatedUser,
Json(req): Json<AuthTokenRequest>,
) -> Result<Json<ActionResponse>, (StatusCode, String)> {
let ext_mgr = state.extension_manager.as_ref().ok_or((
@@ -158,7 +162,7 @@ pub async fn chat_auth_token_handler(
))?;
match ext_mgr
.configure_token(&req.extension_name, &req.token)
.configure_token(&req.extension_name, &req.token, &user.user_id)
.await
{
Ok(result) => {
@@ -176,7 +180,7 @@ pub async fn chat_auth_token_handler(
setup_url: None,
});
} else {
clear_auth_mode(&state).await;
clear_auth_mode(&state, &user.user_id).await;
state.sse.broadcast(SseEvent::AuthCompleted {
extension_name: req.extension_name.clone(),
@@ -205,16 +209,17 @@ pub async fn chat_auth_token_handler(
/// Cancel an in-progress auth flow.
pub async fn chat_auth_cancel_handler(
State(state): State<Arc<GatewayState>>,
AuthenticatedUser(identity): AuthenticatedUser,
Json(_req): Json<AuthCancelRequest>,
) -> Result<Json<ActionResponse>, (StatusCode, String)> {
clear_auth_mode(&state).await;
clear_auth_mode(&state, &identity.user_id).await;
Ok(Json(ActionResponse::ok("Auth cancelled")))
}
/// Clear pending auth mode on the active thread.
pub async fn clear_auth_mode(state: &GatewayState) {
pub async fn clear_auth_mode(state: &GatewayState, user_id: &str) {
if let Some(ref sm) = state.session_manager {
let session = sm.get_or_create_session(&state.user_id).await;
let session = sm.get_or_create_session(user_id).await;
let mut sess = session.lock().await;
if let Some(thread_id) = sess.active_thread
&& let Some(thread) = sess.threads.get_mut(&thread_id)
@@ -226,8 +231,9 @@ pub async fn clear_auth_mode(state: &GatewayState) {
pub async fn chat_events_handler(
State(state): State<Arc<GatewayState>>,
AuthenticatedUser(user): AuthenticatedUser,
) -> Result<impl IntoResponse, (StatusCode, String)> {
state.sse.subscribe().ok_or((
state.sse.subscribe(Some(user.user_id)).ok_or((
StatusCode::SERVICE_UNAVAILABLE,
"Too many connections".to_string(),
))
@@ -237,6 +243,7 @@ pub async fn chat_ws_handler(
headers: axum::http::HeaderMap,
ws: WebSocketUpgrade,
State(state): State<Arc<GatewayState>>,
AuthenticatedUser(identity): AuthenticatedUser,
) -> Result<impl IntoResponse, (StatusCode, String)> {
// Validate Origin header to prevent cross-site WebSocket hijacking.
let origin = headers
@@ -262,7 +269,9 @@ pub async fn chat_ws_handler(
"WebSocket origin not allowed".to_string(),
));
}
Ok(ws.on_upgrade(move |socket| crate::channels::web::ws::handle_ws_connection(socket, state)))
Ok(ws.on_upgrade(move |socket| {
crate::channels::web::ws::handle_ws_connection(socket, state, identity)
}))
}
#[derive(Deserialize)]
@@ -274,6 +283,7 @@ pub struct HistoryQuery {
pub async fn chat_history_handler(
State(state): State<Arc<GatewayState>>,
AuthenticatedUser(identity): AuthenticatedUser,
Query(query): Query<HistoryQuery>,
) -> Result<Json<HistoryResponse>, (StatusCode, String)> {
let session_manager = state.session_manager.as_ref().ok_or((
@@ -281,7 +291,7 @@ pub async fn chat_history_handler(
"Session manager not available".to_string(),
))?;
let session = session_manager.get_or_create_session(&state.user_id).await;
let session = session_manager.get_or_create_session(&identity.user_id).await;
let limit = query.limit.unwrap_or(50);
let before_cursor = query
@@ -314,7 +324,7 @@ pub async fn chat_history_handler(
&& let Some(ref store) = state.store
{
let owned = store
.conversation_belongs_to_user(thread_id, &state.user_id)
.conversation_belongs_to_user(thread_id, &identity.user_id)
.await
.unwrap_or(false);
if !owned {
@@ -434,24 +444,25 @@ pub async fn chat_history_handler(
pub async fn chat_threads_handler(
State(state): State<Arc<GatewayState>>,
AuthenticatedUser(identity): AuthenticatedUser,
) -> Result<Json<ThreadListResponse>, (StatusCode, String)> {
let session_manager = state.session_manager.as_ref().ok_or((
StatusCode::SERVICE_UNAVAILABLE,
"Session manager not available".to_string(),
))?;
let session = session_manager.get_or_create_session(&state.user_id).await;
let session = session_manager.get_or_create_session(&identity.user_id).await;
// Try DB first for persistent thread list
if let Some(ref store) = state.store {
// Auto-create assistant thread if it doesn't exist
let assistant_id = store
.get_or_create_assistant_conversation(&state.user_id, "gateway")
.get_or_create_assistant_conversation(&identity.user_id, "gateway")
.await
.map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
if let Ok(summaries) = store
.list_conversations_all_channels(&state.user_id, 50)
.list_conversations_all_channels(&identity.user_id, 50)
.await
{
let mut assistant_thread = None;
@@ -534,13 +545,14 @@ pub async fn chat_threads_handler(
pub async fn chat_new_thread_handler(
State(state): State<Arc<GatewayState>>,
AuthenticatedUser(identity): AuthenticatedUser,
) -> Result<Json<ThreadInfo>, (StatusCode, String)> {
let session_manager = state.session_manager.as_ref().ok_or((
StatusCode::SERVICE_UNAVAILABLE,
"Session manager not available".to_string(),
))?;
let session = session_manager.get_or_create_session(&state.user_id).await;
let session = session_manager.get_or_create_session(&identity.user_id).await;
let (thread_id, info) = {
let mut sess = session.lock().await;
let thread = sess.create_thread();
@@ -562,12 +574,12 @@ pub async fn chat_new_thread_handler(
// so that the subsequent loadThreads() call from the frontend sees it.
if let Some(ref store) = state.store {
match store
.ensure_conversation(thread_id, "gateway", &state.user_id, None)
.ensure_conversation(thread_id, "gateway", &identity.user_id, None)
.await
{
Ok(true) => {}
Ok(false) => tracing::warn!(
user = %state.user_id,
user = %identity.user_id,
thread_id = %thread_id,
"Skipped persisting new thread due to ownership/channel conflict"
),
src/channels/web/handlers/extensions.rs
+7 -3
View File
@@ -8,11 +8,13 @@ use axum::{
http::StatusCode,
};
use crate::channels::web::auth::AuthenticatedUser;
use crate::channels::web::server::GatewayState;
use crate::channels::web::types::*;
pub async fn extensions_list_handler(
State(state): State<Arc<GatewayState>>,
AuthenticatedUser(user): AuthenticatedUser,
) -> Result<Json<ExtensionListResponse>, (StatusCode, String)> {
let ext_mgr = state.extension_manager.as_ref().ok_or((
StatusCode::NOT_IMPLEMENTED,
@@ -20,7 +22,7 @@ pub async fn extensions_list_handler(
))?;
let installed = ext_mgr
.list(None, false)
.list(None, false, &user.user_id)
.await
.map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
@@ -100,6 +102,7 @@ pub async fn extensions_tools_handler(
pub async fn extensions_install_handler(
State(state): State<Arc<GatewayState>>,
AuthenticatedUser(user): AuthenticatedUser,
Json(req): Json<InstallExtensionRequest>,
) -> Result<Json<ActionResponse>, (StatusCode, String)> {
let ext_mgr = state.extension_manager.as_ref().ok_or((
@@ -116,7 +119,7 @@ pub async fn extensions_install_handler(
});
match ext_mgr
.install(&req.name, req.url.as_deref(), kind_hint)
.install(&req.name, req.url.as_deref(), kind_hint, &user.user_id)
.await
{
Ok(result) => Ok(Json(ActionResponse::ok(result.message))),
@@ -126,6 +129,7 @@ pub async fn extensions_install_handler(
pub async fn extensions_remove_handler(
State(state): State<Arc<GatewayState>>,
AuthenticatedUser(user): AuthenticatedUser,
Path(name): Path<String>,
) -> Result<Json<ActionResponse>, (StatusCode, String)> {
let ext_mgr = state.extension_manager.as_ref().ok_or((
@@ -133,7 +137,7 @@ pub async fn extensions_remove_handler(
"Extension manager not available (secrets store required)".to_string(),
))?;
match ext_mgr.remove(&name).await {
match ext_mgr.remove(&name, &user.user_id).await {
Ok(message) => Ok(Json(ActionResponse::ok(message))),
Err(e) => Ok(Json(ActionResponse::fail(e.to_string()))),
}
src/channels/web/handlers/jobs.rs
+241 -136
View File
@@ -11,11 +11,13 @@ use axum::{
use serde::Deserialize;
use uuid::Uuid;
use crate::channels::web::auth::AuthenticatedUser;
use crate::channels::web::server::GatewayState;
use crate::channels::web::types::*;
pub async fn jobs_list_handler(
State(state): State<Arc<GatewayState>>,
AuthenticatedUser(user): AuthenticatedUser,
) -> Result<Json<JobListResponse>, (StatusCode, String)> {
let store = state.store.as_ref().ok_or((
StatusCode::SERVICE_UNAVAILABLE,
@@ -25,8 +27,8 @@ pub async fn jobs_list_handler(
let mut jobs: Vec<JobInfo> = Vec::new();
let mut seen_ids: HashSet<Uuid> = HashSet::new();
// Fetch sandbox jobs from database.
match store.list_sandbox_jobs().await {
// Fetch sandbox jobs scoped to this user.
match store.list_sandbox_jobs_for_user(&user.user_id).await {
Ok(sandbox_jobs) => {
for j in &sandbox_jobs {
let ui_state = match j.status.as_str() {
@@ -54,6 +56,9 @@ pub async fn jobs_list_handler(
match store.list_agent_jobs().await {
Ok(agent_jobs) => {
for j in &agent_jobs {
if j.user_id != user.user_id {
continue;
}
if seen_ids.contains(&j.id) {
continue;
}
@@ -80,6 +85,7 @@ pub async fn jobs_list_handler(
pub async fn jobs_summary_handler(
State(state): State<Arc<GatewayState>>,
AuthenticatedUser(_user): AuthenticatedUser,
) -> Result<Json<JobSummaryResponse>, (StatusCode, String)> {
let store = state.store.as_ref().ok_or((
StatusCode::SERVICE_UNAVAILABLE,
@@ -134,6 +140,7 @@ pub async fn jobs_summary_handler(
pub async fn jobs_detail_handler(
State(state): State<Arc<GatewayState>>,
AuthenticatedUser(user): AuthenticatedUser,
Path(id): Path<String>,
) -> Result<Json<JobDetailResponse>, (StatusCode, String)> {
let store = state.store.as_ref().ok_or((
@@ -145,169 +152,213 @@ pub async fn jobs_detail_handler(
.map_err(|_| (StatusCode::BAD_REQUEST, "Invalid job ID".to_string()))?;
// Try sandbox job from DB first.
if let Ok(Some(job)) = store.get_sandbox_job(job_id).await {
let browse_id = std::path::Path::new(&job.project_dir)
.file_name()
.map(|n| n.to_string_lossy().to_string())
.unwrap_or_else(|| job.id.to_string());
match store.get_sandbox_job(job_id).await {
Ok(Some(job)) => {
if job.user_id != user.user_id {
return Err((StatusCode::NOT_FOUND, "Job not found".to_string()));
}
let browse_id = std::path::Path::new(&job.project_dir)
.file_name()
.map(|n| n.to_string_lossy().to_string())
.unwrap_or_else(|| job.id.to_string());
let ui_state = match job.status.as_str() {
"creating" => "pending",
"running" => "in_progress",
s => s,
};
let ui_state = match job.status.as_str() {
"creating" => "pending",
"running" => "in_progress",
s => s,
};
let elapsed_secs = job.started_at.map(|start| {
let end = job.completed_at.unwrap_or_else(chrono::Utc::now);
(end - start).num_seconds().max(0) as u64
});
// Synthesize transitions from timestamps.
let mut transitions = Vec::new();
if let Some(started) = job.started_at {
transitions.push(TransitionInfo {
from: "creating".to_string(),
to: "running".to_string(),
timestamp: started.to_rfc3339(),
reason: None,
let elapsed_secs = job.started_at.map(|start| {
let end = job.completed_at.unwrap_or_else(chrono::Utc::now);
(end - start).num_seconds().max(0) as u64
});
}
if let Some(completed) = job.completed_at {
transitions.push(TransitionInfo {
from: "running".to_string(),
to: job.status.clone(),
timestamp: completed.to_rfc3339(),
reason: job.failure_reason.clone(),
});
}
let mode = store.get_sandbox_job_mode(job.id).await.ok().flatten();
let is_claude_code = mode.as_deref() == Some("claude_code");
// Synthesize transitions from timestamps.
let mut transitions = Vec::new();
if let Some(started) = job.started_at {
transitions.push(TransitionInfo {
from: "creating".to_string(),
to: "running".to_string(),
timestamp: started.to_rfc3339(),
reason: None,
});
}
if let Some(completed) = job.completed_at {
transitions.push(TransitionInfo {
from: "running".to_string(),
to: job.status.clone(),
timestamp: completed.to_rfc3339(),
reason: job.failure_reason.clone(),
});
}
return Ok(Json(JobDetailResponse {
id: job.id,
title: job.task.clone(),
description: String::new(),
state: ui_state.to_string(),
user_id: job.user_id.clone(),
created_at: job.created_at.to_rfc3339(),
started_at: job.started_at.map(|dt| dt.to_rfc3339()),
completed_at: job.completed_at.map(|dt| dt.to_rfc3339()),
elapsed_secs,
project_dir: Some(job.project_dir.clone()),
browse_url: Some(format!("/projects/{}/", browse_id)),
job_mode: mode.filter(|m| m != "worker"),
transitions,
can_restart: state.job_manager.is_some(),
can_prompt: is_claude_code && state.prompt_queue.is_some(),
job_kind: Some("sandbox".to_string()),
}));
let mode = store.get_sandbox_job_mode(job.id).await.ok().flatten();
let is_claude_code = mode.as_deref() == Some("claude_code");
return Ok(Json(JobDetailResponse {
id: job.id,
title: job.task.clone(),
description: String::new(),
state: ui_state.to_string(),
user_id: job.user_id.clone(),
created_at: job.created_at.to_rfc3339(),
started_at: job.started_at.map(|dt| dt.to_rfc3339()),
completed_at: job.completed_at.map(|dt| dt.to_rfc3339()),
elapsed_secs,
project_dir: Some(job.project_dir.clone()),
browse_url: Some(format!("/projects/{}/", browse_id)),
job_mode: mode.filter(|m| m != "worker"),
transitions,
can_restart: state.job_manager.is_some(),
can_prompt: is_claude_code && state.prompt_queue.is_some(),
job_kind: Some("sandbox".to_string()),
}));
}
Ok(None) => {}
Err(e) => {
return Err((
StatusCode::INTERNAL_SERVER_ERROR,
format!("Database error: {}", e),
));
}
}
// Fall back to agent job from DB.
if let Ok(Some(ctx)) = store.get_job(job_id).await {
let elapsed_secs = ctx.started_at.map(|start| {
let end = ctx.completed_at.unwrap_or_else(chrono::Utc::now);
(end - start).num_seconds().max(0) as u64
});
match store.get_job(job_id).await {
Ok(Some(ctx)) => {
if ctx.user_id != user.user_id {
return Err((StatusCode::NOT_FOUND, "Job not found".to_string()));
}
let elapsed_secs = ctx.started_at.map(|start| {
let end = ctx.completed_at.unwrap_or_else(chrono::Utc::now);
(end - start).num_seconds().max(0) as u64
});
// Only show prompt bar for jobs that have a running worker (Pending/InProgress).
// Stuck jobs have no active worker loop, so messages would be silently dropped.
let is_promptable = matches!(
ctx.state,
crate::context::JobState::Pending | crate::context::JobState::InProgress
);
return Ok(Json(JobDetailResponse {
id: ctx.job_id,
title: ctx.title.clone(),
description: ctx.description.clone(),
state: ctx.state.to_string(),
user_id: ctx.user_id.clone(),
created_at: ctx.created_at.to_rfc3339(),
started_at: ctx.started_at.map(|dt| dt.to_rfc3339()),
completed_at: ctx.completed_at.map(|dt| dt.to_rfc3339()),
elapsed_secs,
project_dir: None,
browse_url: None,
job_mode: None,
transitions: Vec::new(),
can_restart: state.scheduler.is_some(),
can_prompt: is_promptable && state.scheduler.is_some(),
job_kind: Some("agent".to_string()),
}));
// Only show prompt bar for jobs that have a running worker (Pending/InProgress).
// Stuck jobs have no active worker loop, so messages would be silently dropped.
let is_promptable = matches!(
ctx.state,
crate::context::JobState::Pending | crate::context::JobState::InProgress
);
Ok(Json(JobDetailResponse {
id: ctx.job_id,
title: ctx.title.clone(),
description: ctx.description.clone(),
state: ctx.state.to_string(),
user_id: ctx.user_id.clone(),
created_at: ctx.created_at.to_rfc3339(),
started_at: ctx.started_at.map(|dt| dt.to_rfc3339()),
completed_at: ctx.completed_at.map(|dt| dt.to_rfc3339()),
elapsed_secs,
project_dir: None,
browse_url: None,
job_mode: None,
transitions: Vec::new(),
can_restart: state.scheduler.is_some(),
can_prompt: is_promptable && state.scheduler.is_some(),
job_kind: Some("agent".to_string()),
}))
}
Ok(None) => Err((StatusCode::NOT_FOUND, "Job not found".to_string())),
Err(e) => Err((
StatusCode::INTERNAL_SERVER_ERROR,
format!("Database error: {}", e),
)),
}
Err((StatusCode::NOT_FOUND, "Job not found".to_string()))
}
pub async fn jobs_cancel_handler(
State(state): State<Arc<GatewayState>>,
AuthenticatedUser(user): AuthenticatedUser,
Path(id): Path<String>,
) -> Result<Json<serde_json::Value>, (StatusCode, String)> {
let job_id = Uuid::parse_str(&id)
.map_err(|_| (StatusCode::BAD_REQUEST, "Invalid job ID".to_string()))?;
// Try sandbox job cancellation.
if let Some(ref store) = state.store
&& let Ok(Some(job)) = store.get_sandbox_job(job_id).await
{
if job.status == "running" || job.status == "creating" {
// Stop the container if we have a job manager.
if let Some(ref jm) = state.job_manager
&& let Err(e) = jm.stop_job(job_id).await
{
tracing::warn!(job_id = %job_id, error = %e, "Failed to stop container during cancellation");
if let Some(ref store) = state.store {
match store.get_sandbox_job(job_id).await {
Ok(Some(job)) => {
if job.user_id != user.user_id {
return Err((StatusCode::NOT_FOUND, "Job not found".to_string()));
}
if job.status == "running" || job.status == "creating" {
if let Some(ref jm) = state.job_manager
&& let Err(e) = jm.stop_job(job_id).await
{
tracing::warn!(job_id = %job_id, error = %e, "Failed to stop container during cancellation");
}
store
.update_sandbox_job_status(
job_id,
"failed",
Some(false),
Some("Cancelled by user"),
None,
Some(chrono::Utc::now()),
)
.await
.map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
}
return Ok(Json(serde_json::json!({
"status": "cancelled",
"job_id": job_id,
})));
}
Ok(None) => {}
Err(e) => {
return Err((
StatusCode::INTERNAL_SERVER_ERROR,
format!("Database error: {}", e),
));
}
store
.update_sandbox_job_status(
job_id,
"failed",
Some(false),
Some("Cancelled by user"),
None,
Some(chrono::Utc::now()),
)
.await
.map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
}
return Ok(Json(serde_json::json!({
"status": "cancelled",
"job_id": job_id,
})));
}
// Fall back to agent job cancellation: stop the worker via the scheduler
// (which updates the in-memory ContextManager AND aborts the task handle),
// then persist the status to the DB as a fallback.
if let Some(ref store) = state.store
&& let Ok(Some(job)) = store.get_job(job_id).await
{
if job.state.is_active() {
// Try to stop via scheduler (aborts the worker task + updates
// in-memory ContextManager). This is best-effort — the job may
// not be in the scheduler map if it already finished.
if let Some(ref slot) = state.scheduler
&& let Some(ref scheduler) = *slot.read().await
{
let _ = scheduler.stop(job_id).await;
}
if let Some(ref store) = state.store {
match store.get_job(job_id).await {
Ok(Some(job)) => {
if job.user_id != user.user_id {
return Err((StatusCode::NOT_FOUND, "Job not found".to_string()));
}
if job.state.is_active() {
// Try to stop via scheduler (aborts the worker task + updates
// in-memory ContextManager). This is best-effort — the job may
// not be in the scheduler map if it already finished.
if let Some(ref slot) = state.scheduler
&& let Some(ref scheduler) = *slot.read().await
{
let _ = scheduler.stop(job_id).await;
}
// Always persist cancellation to the DB so the state is
// consistent even if the scheduler wasn't available or the
// job wasn't in its in-memory map.
store
.update_job_status(
job_id,
crate::context::JobState::Cancelled,
Some("Cancelled by user"),
)
.await
.map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
// Always persist cancellation to the DB so the state is
// consistent even if the scheduler wasn't available or the
// job wasn't in its in-memory map.
store
.update_job_status(
job_id,
crate::context::JobState::Cancelled,
Some("Cancelled by user"),
)
.await
.map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
}
return Ok(Json(serde_json::json!({
"status": "cancelled",
"job_id": job_id,
})));
}
Ok(None) => {}
Err(e) => {
return Err((
StatusCode::INTERNAL_SERVER_ERROR,
format!("Database error: {}", e),
));
}
}
return Ok(Json(serde_json::json!({
"status": "cancelled",
"job_id": job_id,
})));
}
Err((StatusCode::NOT_FOUND, "Job not found".to_string()))
@@ -315,6 +366,7 @@ pub async fn jobs_cancel_handler(
pub async fn jobs_restart_handler(
State(state): State<Arc<GatewayState>>,
AuthenticatedUser(user): AuthenticatedUser,
Path(id): Path<String>,
) -> Result<Json<serde_json::Value>, (StatusCode, String)> {
let store = state.store.as_ref().ok_or((
@@ -327,6 +379,9 @@ pub async fn jobs_restart_handler(
// Try sandbox job restart first.
if let Ok(Some(old_job)) = store.get_sandbox_job(old_job_id).await {
if old_job.user_id != user.user_id {
return Err((StatusCode::NOT_FOUND, "Job not found".to_string()));
}
if old_job.status != "interrupted" && old_job.status != "failed" {
return Err((
StatusCode::CONFLICT,
@@ -476,6 +531,7 @@ pub async fn jobs_restart_handler(
/// - Worker-mode sandbox jobs → not supported (no mechanism to inject)
pub async fn jobs_prompt_handler(
State(state): State<Arc<GatewayState>>,
AuthenticatedUser(user): AuthenticatedUser,
Path(id): Path<String>,
Json(body): Json<serde_json::Value>,
) -> Result<Json<serde_json::Value>, (StatusCode, String)> {
@@ -483,6 +539,26 @@ pub async fn jobs_prompt_handler(
.parse()
.map_err(|_| (StatusCode::BAD_REQUEST, "Invalid job ID".to_string()))?;
// Verify ownership before queuing a prompt.
if let Some(ref store) = state.store {
match store.get_sandbox_job(job_id).await {
Ok(Some(job)) => {
if job.user_id != user.user_id {
return Err((StatusCode::NOT_FOUND, "Job not found".to_string()));
}
}
Ok(None) => {
return Err((StatusCode::NOT_FOUND, "Job not found".to_string()));
}
Err(e) => {
return Err((
StatusCode::INTERNAL_SERVER_ERROR,
format!("Database error: {}", e),
));
}
}
}
let content = body
.get("content")
.and_then(|v| v.as_str())
@@ -550,6 +626,7 @@ pub async fn jobs_prompt_handler(
/// Load persisted job events for a job (for history replay on page open).
pub async fn jobs_events_handler(
State(state): State<Arc<GatewayState>>,
AuthenticatedUser(user): AuthenticatedUser,
Path(id): Path<String>,
) -> Result<Json<serde_json::Value>, (StatusCode, String)> {
let store = state.store.as_ref().ok_or((
@@ -561,6 +638,24 @@ pub async fn jobs_events_handler(
.parse()
.map_err(|_| (StatusCode::BAD_REQUEST, "Invalid job ID".to_string()))?;
// Verify ownership before returning events.
match store.get_sandbox_job(job_id).await {
Ok(Some(job)) => {
if job.user_id != user.user_id {
return Err((StatusCode::NOT_FOUND, "Job not found".to_string()));
}
}
Ok(None) => {
return Err((StatusCode::NOT_FOUND, "Job not found".to_string()));
}
Err(e) => {
return Err((
StatusCode::INTERNAL_SERVER_ERROR,
format!("Database error: {}", e),
));
}
}
let events = store
.list_job_events(job_id, None)
.await
@@ -593,6 +688,7 @@ pub struct FilePathQuery {
pub async fn job_files_list_handler(
State(state): State<Arc<GatewayState>>,
AuthenticatedUser(user): AuthenticatedUser,
Path(id): Path<String>,
Query(query): Query<FilePathQuery>,
) -> Result<Json<ProjectFilesResponse>, (StatusCode, String)> {
@@ -610,6 +706,10 @@ pub async fn job_files_list_handler(
.map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?
.ok_or((StatusCode::NOT_FOUND, "Job not found".to_string()))?;
if job.user_id != user.user_id {
return Err((StatusCode::NOT_FOUND, "Job not found".to_string()));
}
let base = std::path::PathBuf::from(&job.project_dir);
let rel_path = query.path.as_deref().unwrap_or("");
let target = base.join(rel_path);
@@ -656,6 +756,7 @@ pub async fn job_files_list_handler(
pub async fn job_files_read_handler(
State(state): State<Arc<GatewayState>>,
AuthenticatedUser(user): AuthenticatedUser,
Path(id): Path<String>,
Query(query): Query<FilePathQuery>,
) -> Result<Json<ProjectFileReadResponse>, (StatusCode, String)> {
@@ -673,6 +774,10 @@ pub async fn job_files_read_handler(
.map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?
.ok_or((StatusCode::NOT_FOUND, "Job not found".to_string()))?;
if job.user_id != user.user_id {
return Err((StatusCode::NOT_FOUND, "Job not found".to_string()));
}
let path = query.path.as_deref().ok_or((
StatusCode::BAD_REQUEST,
"path parameter required".to_string(),
src/channels/web/handlers/mod.rs
+1 -2
View File
@@ -8,6 +8,7 @@
//! The remaining modules are in-progress migrations from inline server.rs
//! handlers; their functions are not yet wired up, hence the `dead_code` allow.
pub mod jobs;
pub mod skills;
// Modules not yet wired into server.rs router -- suppress dead_code until
@@ -17,8 +18,6 @@ pub mod chat;
#[allow(dead_code)]
pub mod extensions;
#[allow(dead_code)]
pub mod jobs;
#[allow(dead_code)]
pub mod memory;
#[allow(dead_code)]
pub mod routines;
src/channels/web/handlers/routines.rs
+2 -1
View File
@@ -137,6 +137,7 @@ pub async fn routines_detail_handler(
pub async fn routines_trigger_handler(
State(state): State<Arc<GatewayState>>,
crate::channels::web::auth::AuthenticatedUser(user): crate::channels::web::auth::AuthenticatedUser,
Path(id): Path<String>,
) -> Result<Json<serde_json::Value>, (StatusCode, String)> {
// Clone the Arc out of the lock to avoid holding the RwLock across .await.
@@ -152,7 +153,7 @@ pub async fn routines_trigger_handler(
.map_err(|_| (StatusCode::BAD_REQUEST, "Invalid routine ID".to_string()))?;
let run_id = engine
.fire_manual(routine_id, Some(&state.user_id))
.fire_manual(routine_id, Some(&user.user_id))
.await
.map_err(|e| (routine_error_status(&e), e.to_string()))?;
src/channels/web/handlers/settings.rs
+13 -6
View File
@@ -8,17 +8,19 @@ use axum::{
http::StatusCode,
};
use crate::channels::web::auth::AuthenticatedUser;
use crate::channels::web::server::GatewayState;
use crate::channels::web::types::*;
pub async fn settings_list_handler(
State(state): State<Arc<GatewayState>>,
AuthenticatedUser(user): AuthenticatedUser,
) -> Result<Json<SettingsListResponse>, StatusCode> {
let store = state
.store
.as_ref()
.ok_or(StatusCode::SERVICE_UNAVAILABLE)?;
let rows = store.list_settings(&state.user_id).await.map_err(|e| {
let rows = store.list_settings(&user.user_id).await.map_err(|e| {
tracing::error!("Failed to list settings: {}", e);
StatusCode::INTERNAL_SERVER_ERROR
})?;
@@ -37,6 +39,7 @@ pub async fn settings_list_handler(
pub async fn settings_get_handler(
State(state): State<Arc<GatewayState>>,
AuthenticatedUser(user): AuthenticatedUser,
Path(key): Path<String>,
) -> Result<Json<SettingResponse>, StatusCode> {
let store = state
@@ -44,7 +47,7 @@ pub async fn settings_get_handler(
.as_ref()
.ok_or(StatusCode::SERVICE_UNAVAILABLE)?;
let row = store
.get_setting_full(&state.user_id, &key)
.get_setting_full(&user.user_id, &key)
.await
.map_err(|e| {
tracing::error!("Failed to get setting '{}': {}", key, e);
@@ -61,6 +64,7 @@ pub async fn settings_get_handler(
pub async fn settings_set_handler(
State(state): State<Arc<GatewayState>>,
AuthenticatedUser(user): AuthenticatedUser,
Path(key): Path<String>,
Json(body): Json<SettingWriteRequest>,
) -> Result<StatusCode, StatusCode> {
@@ -69,7 +73,7 @@ pub async fn settings_set_handler(
.as_ref()
.ok_or(StatusCode::SERVICE_UNAVAILABLE)?;
store
.set_setting(&state.user_id, &key, &body.value)
.set_setting(&user.user_id, &key, &body.value)
.await
.map_err(|e| {
tracing::error!("Failed to set setting '{}': {}", key, e);
@@ -81,6 +85,7 @@ pub async fn settings_set_handler(
pub async fn settings_delete_handler(
State(state): State<Arc<GatewayState>>,
AuthenticatedUser(user): AuthenticatedUser,
Path(key): Path<String>,
) -> Result<StatusCode, StatusCode> {
let store = state
@@ -88,7 +93,7 @@ pub async fn settings_delete_handler(
.as_ref()
.ok_or(StatusCode::SERVICE_UNAVAILABLE)?;
store
.delete_setting(&state.user_id, &key)
.delete_setting(&user.user_id, &key)
.await
.map_err(|e| {
tracing::error!("Failed to delete setting '{}': {}", key, e);
@@ -100,12 +105,13 @@ pub async fn settings_delete_handler(
pub async fn settings_export_handler(
State(state): State<Arc<GatewayState>>,
AuthenticatedUser(user): AuthenticatedUser,
) -> Result<Json<SettingsExportResponse>, StatusCode> {
let store = state
.store
.as_ref()
.ok_or(StatusCode::SERVICE_UNAVAILABLE)?;
let settings = store.get_all_settings(&state.user_id).await.map_err(|e| {
let settings = store.get_all_settings(&user.user_id).await.map_err(|e| {
tracing::error!("Failed to export settings: {}", e);
StatusCode::INTERNAL_SERVER_ERROR
})?;
@@ -115,6 +121,7 @@ pub async fn settings_export_handler(
pub async fn settings_import_handler(
State(state): State<Arc<GatewayState>>,
AuthenticatedUser(user): AuthenticatedUser,
Json(body): Json<SettingsImportRequest>,
) -> Result<StatusCode, StatusCode> {
let store = state
@@ -122,7 +129,7 @@ pub async fn settings_import_handler(
.as_ref()
.ok_or(StatusCode::SERVICE_UNAVAILABLE)?;
store
.set_all_settings(&state.user_id, &body.settings)
.set_all_settings(&user.user_id, &body.settings)
.await
.map_err(|e| {
tracing::error!("Failed to import settings: {}", e);
src/channels/web/mod.rs
+87 -22
View File
@@ -52,6 +52,7 @@ use crate::workspace::Workspace;
use self::log_layer::{LogBroadcaster, LogLevelHandle};
use self::auth::MultiAuthState;
use self::server::GatewayState;
use self::sse::SseManager;
use self::types::SseEvent;
@@ -60,14 +61,15 @@ use self::types::SseEvent;
pub struct GatewayChannel {
config: GatewayConfig,
state: Arc<GatewayState>,
/// The actual auth token in use (generated or from config).
auth_token: String,
/// Multi-user auth state (replaces bare auth_token).
auth: MultiAuthState,
}
impl GatewayChannel {
/// Create a new gateway channel.
///
/// If no auth token is configured, generates a random one and prints it.
/// Builds a single-user `MultiAuthState` from the config.
pub fn new(config: GatewayConfig) -> Self {
let auth_token = config.auth_token.clone().unwrap_or_else(|| {
use rand::RngCore;
@@ -77,10 +79,13 @@ impl GatewayChannel {
bytes.iter().map(|b| format!("{b:02x}")).collect()
});
let auth = MultiAuthState::single(auth_token, config.user_id.clone());
let state = Arc::new(GatewayState {
msg_tx: tokio::sync::RwLock::new(None),
sse: SseManager::new(),
sse: Arc::new(SseManager::new()),
workspace: None,
workspace_pool: None,
session_manager: None,
log_broadcaster: None,
log_level_handle: None,
@@ -90,13 +95,13 @@ impl GatewayChannel {
job_manager: None,
prompt_queue: None,
scheduler: None,
user_id: config.user_id.clone(),
default_user_id: config.user_id.clone(),
shutdown_tx: tokio::sync::RwLock::new(None),
ws_tracker: Some(Arc::new(ws::WsConnectionTracker::new())),
llm_provider: None,
skill_registry: None,
skill_catalog: None,
chat_rate_limiter: server::RateLimiter::new(30, 60),
chat_rate_limiter: server::PerUserRateLimiter::new(30, 60),
oauth_rate_limiter: server::RateLimiter::new(10, 60),
webhook_rate_limiter: server::RateLimiter::new(10, 60),
registry_entries: Vec::new(),
@@ -109,7 +114,46 @@ impl GatewayChannel {
Self {
config,
state,
auth_token,
auth,
}
}
/// 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 {
msg_tx: tokio::sync::RwLock::new(None),
sse: Arc::new(SseManager::new()),
workspace: None,
workspace_pool: None,
session_manager: None,
log_broadcaster: None,
log_level_handle: None,
extension_manager: None,
tool_registry: None,
store: None,
job_manager: None,
prompt_queue: None,
scheduler: None,
default_user_id: config.user_id.clone(),
shutdown_tx: tokio::sync::RwLock::new(None),
ws_tracker: Some(Arc::new(ws::WsConnectionTracker::new())),
llm_provider: None,
skill_registry: None,
skill_catalog: None,
chat_rate_limiter: server::PerUserRateLimiter::new(30, 60),
oauth_rate_limiter: server::RateLimiter::new(10, 60),
registry_entries: Vec::new(),
cost_guard: None,
routine_engine: Arc::new(tokio::sync::RwLock::new(None)),
startup_time: std::time::Instant::now(),
webhook_rate_limiter: server::RateLimiter::new(10, 60),
active_config: server::ActiveConfigSnapshot::default(),
});
Self {
config,
state,
auth,
}
}
@@ -118,8 +162,9 @@ impl GatewayChannel {
let mut new_state = GatewayState {
msg_tx: tokio::sync::RwLock::new(None),
// Preserve the existing broadcast channel so sender handles remain valid.
sse: SseManager::from_sender(self.state.sse.sender()),
sse: Arc::new(SseManager::from_sender(self.state.sse.sender())),
workspace: self.state.workspace.clone(),
workspace_pool: self.state.workspace_pool.clone(),
session_manager: self.state.session_manager.clone(),
log_broadcaster: self.state.log_broadcaster.clone(),
log_level_handle: self.state.log_level_handle.clone(),
@@ -129,13 +174,13 @@ impl GatewayChannel {
job_manager: self.state.job_manager.clone(),
prompt_queue: self.state.prompt_queue.clone(),
scheduler: self.state.scheduler.clone(),
user_id: self.state.user_id.clone(),
default_user_id: self.state.default_user_id.clone(),
shutdown_tx: tokio::sync::RwLock::new(None),
ws_tracker: self.state.ws_tracker.clone(),
llm_provider: self.state.llm_provider.clone(),
skill_registry: self.state.skill_registry.clone(),
skill_catalog: self.state.skill_catalog.clone(),
chat_rate_limiter: server::RateLimiter::new(30, 60),
chat_rate_limiter: server::PerUserRateLimiter::new(30, 60),
oauth_rate_limiter: server::RateLimiter::new(10, 60),
webhook_rate_limiter: server::RateLimiter::new(10, 60),
registry_entries: self.state.registry_entries.clone(),
@@ -260,9 +305,15 @@ impl GatewayChannel {
self
}
/// Get the auth token (for printing to console on startup).
/// Inject the per-user workspace pool for multi-user mode.
pub fn with_workspace_pool(mut self, pool: Arc<server::WorkspacePool>) -> Self {
self.rebuild_state(|s| s.workspace_pool = Some(pool));
self
}
/// Get the first auth token (for printing to console on startup).
pub fn auth_token(&self) -> &str {
&self.auth_token
self.auth.first_token().unwrap_or("")
}
/// Get a reference to the shared gateway state (for the agent to push SSE events).
@@ -291,7 +342,7 @@ impl Channel for GatewayChannel {
),
})?;
server::start_server(addr, self.state.clone(), self.auth_token.clone()).await?;
server::start_server(addr, self.state.clone(), self.auth.clone()).await?;
Ok(Box::pin(ReceiverStream::new(rx)))
}
@@ -311,10 +362,13 @@ impl Channel for GatewayChannel {
}
};
self.state.sse.broadcast(SseEvent::Response {
content: response.content,
thread_id,
});
self.state.sse.broadcast_for_user(
&msg.user_id,
SseEvent::Response {
content: response.content,
thread_id,
},
);
Ok(())
}
@@ -427,13 +481,21 @@ impl Channel for GatewayChannel {
},
};
self.state.sse.broadcast(event);
// Scope events to the user when user_id is available in metadata.
// When user_id is missing (heartbeat, routines), events go to all
// subscribers. In multi-tenant mode this leaks status across users.
if let Some(uid) = metadata.get("user_id").and_then(|v| v.as_str()) {
self.state.sse.broadcast_for_user(uid, event);
} else {
tracing::debug!("Status event missing user_id in metadata; broadcasting globally");
self.state.sse.broadcast(event);
}
Ok(())
}
async fn broadcast(
&self,
_user_id: &str,
user_id: &str,
response: OutgoingResponse,
) -> Result<(), ChannelError> {
let thread_id = match response.thread_id {
@@ -445,10 +507,13 @@ impl Channel for GatewayChannel {
return Ok(());
}
};
self.state.sse.broadcast(SseEvent::Response {
content: response.content,
thread_id,
});
self.state.sse.broadcast_for_user(
user_id,
SseEvent::Response {
content: response.content,
thread_id,
},
);
Ok(())
}
src/channels/web/openai_compat.rs
+2 -1
View File
@@ -463,9 +463,10 @@ fn build_tool_request(
pub async fn chat_completions_handler(
State(state): State<Arc<GatewayState>>,
super::auth::AuthenticatedUser(user): super::auth::AuthenticatedUser,
Json(req): Json<OpenAiChatRequest>,
) -> Result<impl IntoResponse, (StatusCode, Json<OpenAiErrorResponse>)> {
if !state.chat_rate_limiter.check() {
if !state.chat_rate_limiter.check(&user.user_id) {
return Err(openai_error(
StatusCode::TOO_MANY_REQUESTS,
"Rate limit exceeded. Please try again later.",
src/channels/web/server.rs
+398 -136
View File
File diff suppressed because it is too large Load Diff
src/channels/web/sse.rs
+120 -26
View File
@@ -17,9 +17,25 @@ use crate::channels::web::types::SseEvent;
/// Prevents resource exhaustion from connection flooding.
const MAX_CONNECTIONS: u64 = 100;
/// Envelope for broadcast events: carries an optional user scope.
///
/// `user_id = None` means the event is global (e.g. Heartbeat) and delivered
/// to all subscribers. `user_id = Some(id)` means the event is only delivered
/// to subscribers that match that user_id.
#[derive(Debug, Clone)]
pub(crate) struct ScopedEvent {
pub(crate) user_id: Option<String>,
pub(crate) event: SseEvent,
}
/// Manages SSE broadcast to all connected browser tabs.
///
/// In multi-user mode, events are scoped by user_id so that each subscriber
/// only receives events intended for their user (plus global events like
/// Heartbeat). In single-user mode, all events are delivered to all subscribers
/// (backwards compatible).
pub struct SseManager {
tx: broadcast::Sender<SseEvent>,
tx: broadcast::Sender<ScopedEvent>,
connection_count: Arc<AtomicU64>,
max_connections: u64,
}
@@ -45,7 +61,7 @@ impl SseManager {
/// only be called before the server starts accepting connections (i.e.,
/// during startup wiring). Calling it after connections are established
/// will break connection tracking and allow exceeding `MAX_CONNECTIONS`.
pub fn from_sender(tx: broadcast::Sender<SseEvent>) -> Self {
pub(crate) fn from_sender(tx: broadcast::Sender<ScopedEvent>) -> Self {
Self {
tx,
connection_count: Arc::new(AtomicU64::new(0)),
@@ -53,15 +69,28 @@ impl SseManager {
}
}
/// Broadcast an event to all connected clients.
pub fn broadcast(&self, event: SseEvent) {
// Ignore send errors (no receivers is fine)
let _ = self.tx.send(event);
/// Get a clone of the broadcast sender for use by other components.
pub(crate) fn sender(&self) -> broadcast::Sender<ScopedEvent> {
self.tx.clone()
}
/// Get a clone of the broadcast sender for use by other components.
pub fn sender(&self) -> broadcast::Sender<SseEvent> {
self.tx.clone()
/// Broadcast an event to all connected clients (global/unscoped).
pub fn broadcast(&self, event: SseEvent) {
let _ = self.tx.send(ScopedEvent {
user_id: None,
event,
});
}
/// Broadcast an event scoped to a specific user.
///
/// Only subscribers for this user_id (or unscoped subscribers) will
/// receive the event.
pub fn broadcast_for_user(&self, user_id: &str, event: SseEvent) {
let _ = self.tx.send(ScopedEvent {
user_id: Some(user_id.to_string()),
event,
});
}
/// Get current number of active connections.
@@ -71,11 +100,15 @@ impl SseManager {
/// Create a raw broadcast subscription for non-SSE consumers (e.g. WebSocket).
///
/// Returns a stream of `SseEvent` values and increments/decrements the
/// connection counter on creation/drop, just like `subscribe()` does for SSE.
/// When `user_id` is `Some`, only events scoped to that user (or global
/// events) are delivered. When `None`, all events are delivered (single-user
/// backwards compatibility).
///
/// Returns `None` if the maximum connection limit has been reached.
pub fn subscribe_raw(&self) -> Option<impl Stream<Item = SseEvent> + Send + 'static + use<>> {
pub fn subscribe_raw(
&self,
user_id: Option<String>,
) -> Option<impl Stream<Item = SseEvent> + 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);
@@ -91,7 +124,19 @@ impl SseManager {
.ok()?;
let rx = self.tx.subscribe();
let stream = BroadcastStream::new(rx).filter_map(|result| result.ok());
let stream = BroadcastStream::new(rx).filter_map(move |result| match result {
Ok(scoped) => {
// Global events (user_id=None) always pass through.
// Scoped events only pass if the subscriber matches (or subscriber is unscoped).
match (&user_id, &scoped.user_id) {
(_, None) => Some(scoped.event), // global -> all
(None, _) => Some(scoped.event), // unscoped subscriber -> all
(Some(sub), Some(ev)) if sub == ev => Some(scoped.event), // match
_ => None, // different user -> skip
}
}
Err(_) => None,
});
Some(CountedStream {
inner: stream,
@@ -101,9 +146,13 @@ impl SseManager {
/// Create a new SSE stream for a client connection.
///
/// When `user_id` is `Some`, only events for that user (or global events)
/// are delivered. When `None`, all events are delivered.
///
/// Returns `None` if the maximum connection limit has been reached.
pub fn subscribe(
&self,
user_id: Option<String>,
) -> Option<Sse<impl Stream<Item = Result<Event, Infallible>> + Send + 'static + use<>>> {
// Atomically increment only if below the limit.
let counter = Arc::clone(&self.connection_count);
@@ -120,7 +169,15 @@ impl SseManager {
let rx = self.tx.subscribe();
let stream = BroadcastStream::new(rx)
.filter_map(|result| result.ok())
.filter_map(move |result| match result {
Ok(scoped) => match (&user_id, &scoped.user_id) {
(_, None) => Some(scoped.event),
(None, _) => Some(scoped.event),
(Some(sub), Some(ev)) if sub == ev => Some(scoped.event),
_ => None,
},
Err(_) => None,
})
.map(|event| {
let data = serde_json::to_string(&event).unwrap_or_default();
let event_type = match &event {
@@ -215,16 +272,14 @@ mod tests {
#[tokio::test]
async fn test_broadcast_to_receiver() {
let manager = SseManager::new();
let mut rx = BroadcastStream::new(manager.tx.subscribe());
let mut stream = Box::pin(manager.subscribe_raw(None).expect("should subscribe"));
manager.broadcast(SseEvent::Status {
message: "test".to_string(),
thread_id: None,
});
let event = rx.next().await;
assert!(event.is_some());
let event = event.unwrap().unwrap();
let event = stream.next().await.unwrap();
match event {
SseEvent::Status { message, .. } => assert_eq!(message, "test"),
_ => panic!("unexpected event type"),
@@ -234,7 +289,7 @@ mod tests {
#[tokio::test]
async fn test_subscribe_raw_receives_events() {
let manager = SseManager::new();
let mut stream = Box::pin(manager.subscribe_raw().expect("should subscribe"));
let mut stream = Box::pin(manager.subscribe_raw(None).expect("should subscribe"));
assert_eq!(manager.connection_count(), 1);
@@ -254,7 +309,7 @@ mod tests {
async fn test_subscribe_raw_decrements_on_drop() {
let manager = SseManager::new();
{
let _stream = Box::pin(manager.subscribe_raw().expect("should subscribe"));
let _stream = Box::pin(manager.subscribe_raw(None).expect("should subscribe"));
assert_eq!(manager.connection_count(), 1);
}
// Stream dropped, counter should decrement
@@ -264,8 +319,8 @@ mod tests {
#[tokio::test]
async fn test_subscribe_raw_multiple_subscribers() {
let manager = SseManager::new();
let mut s1 = Box::pin(manager.subscribe_raw().expect("should subscribe"));
let mut s2 = Box::pin(manager.subscribe_raw().expect("should subscribe"));
let mut s1 = Box::pin(manager.subscribe_raw(None).expect("should subscribe"));
let mut s2 = Box::pin(manager.subscribe_raw(None).expect("should subscribe"));
assert_eq!(manager.connection_count(), 2);
manager.broadcast(SseEvent::Heartbeat);
@@ -286,12 +341,51 @@ mod tests {
let mut manager = SseManager::new();
manager.max_connections = 2; // Low limit for testing
let _s1 = Box::pin(manager.subscribe_raw().expect("first should succeed"));
let _s2 = Box::pin(manager.subscribe_raw().expect("second should succeed"));
let _s1 = Box::pin(manager.subscribe_raw(None).expect("first should succeed"));
let _s2 = Box::pin(manager.subscribe_raw(None).expect("second should succeed"));
assert_eq!(manager.connection_count(), 2);
// Third should be rejected
assert!(manager.subscribe_raw().is_none());
assert!(manager.subscribe().is_none());
assert!(manager.subscribe_raw(None).is_none());
assert!(manager.subscribe(None).is_none());
}
#[tokio::test]
async fn test_scoped_events_filtered_by_user() {
let manager = SseManager::new();
let mut alice = Box::pin(
manager
.subscribe_raw(Some("alice".to_string()))
.expect("subscribe"),
);
let mut bob = Box::pin(
manager
.subscribe_raw(Some("bob".to_string()))
.expect("subscribe"),
);
// Send event scoped to alice
manager.broadcast_for_user(
"alice",
SseEvent::Status {
message: "alice only".to_string(),
thread_id: None,
},
);
// Send global event
manager.broadcast(SseEvent::Heartbeat);
// Alice gets her scoped event
let e = alice.next().await.unwrap();
assert!(matches!(e, SseEvent::Status { .. }));
// Alice also gets the global heartbeat
let e = alice.next().await.unwrap();
assert!(matches!(e, SseEvent::Heartbeat));
// Bob only gets the global heartbeat (alice's event was filtered)
let e = bob.next().await.unwrap();
assert!(matches!(e, SseEvent::Heartbeat));
}
}
src/channels/web/test_helpers.rs
+22 -5
View File
@@ -10,7 +10,8 @@ use std::sync::Arc;
use tokio::sync::mpsc;
use crate::channels::IncomingMessage;
use crate::channels::web::server::{GatewayState, RateLimiter, start_server};
use crate::channels::web::auth::MultiAuthState;
use crate::channels::web::server::{GatewayState, PerUserRateLimiter, RateLimiter, start_server};
use crate::channels::web::sse::SseManager;
use crate::channels::web::ws::WsConnectionTracker;
@@ -64,8 +65,9 @@ impl TestGatewayBuilder {
pub fn build(self) -> Arc<GatewayState> {
Arc::new(GatewayState {
msg_tx: tokio::sync::RwLock::new(self.msg_tx),
sse: SseManager::new(),
sse: Arc::new(SseManager::new()),
workspace: None,
workspace_pool: None,
session_manager: None,
log_broadcaster: None,
log_level_handle: None,
@@ -74,14 +76,14 @@ impl TestGatewayBuilder {
store: None,
job_manager: None,
prompt_queue: None,
user_id: self.user_id,
default_user_id: self.user_id,
shutdown_tx: tokio::sync::RwLock::new(None),
ws_tracker: Some(Arc::new(WsConnectionTracker::new())),
llm_provider: self.llm_provider,
skill_registry: None,
skill_catalog: None,
scheduler: None,
chat_rate_limiter: RateLimiter::new(30, 60),
chat_rate_limiter: PerUserRateLimiter::new(30, 60),
oauth_rate_limiter: RateLimiter::new(10, 60),
webhook_rate_limiter: RateLimiter::new(10, 60),
registry_entries: Vec::new(),
@@ -98,11 +100,26 @@ impl TestGatewayBuilder {
self,
auth_token: &str,
) -> Result<(SocketAddr, Arc<GatewayState>), crate::error::ChannelError> {
let auth = MultiAuthState::single(auth_token.to_string(), "test-user".to_string());
let state = self.build();
let addr: SocketAddr = "127.0.0.1:0"
.parse()
.expect("hard-coded address must parse");
let bound = start_server(addr, state.clone(), auth_token.to_string()).await?;
let bound = start_server(addr, state.clone(), auth).await?;
Ok((bound, state))
}
/// Build the state and start a gateway server with multi-user auth.
/// Returns the bound address and the shared state.
pub async fn start_multi(
self,
auth: MultiAuthState,
) -> Result<(SocketAddr, Arc<GatewayState>), crate::error::ChannelError> {
let state = self.build();
let addr: SocketAddr = "127.0.0.1:0"
.parse()
.expect("hard-coded address must parse");
let bound = start_server(addr, state.clone(), auth).await?;
Ok((bound, state))
}
}
src/channels/web/ws.rs
+21 -13
View File
@@ -62,7 +62,11 @@ impl Default for WsConnectionTracker {
///
/// When either task ends (client disconnect or broadcast closed), both are
/// cleaned up.
pub async fn handle_ws_connection(socket: WebSocket, state: Arc<GatewayState>) {
pub async fn handle_ws_connection(
socket: WebSocket,
state: Arc<GatewayState>,
user: crate::channels::web::auth::UserIdentity,
) {
let (mut ws_sink, mut ws_stream) = socket.split();
// Track connection
@@ -71,9 +75,9 @@ pub async fn handle_ws_connection(socket: WebSocket, state: Arc<GatewayState>) {
}
let tracker_for_drop = state.ws_tracker.clone();
// Subscribe to broadcast events (same source as SSE).
// Subscribe to broadcast events (same source as SSE), scoped to this user.
// Reject if we've hit the connection limit.
let Some(raw_stream) = state.sse.subscribe_raw() else {
let Some(raw_stream) = state.sse.subscribe_raw(Some(user.user_id.clone())) else {
tracing::warn!("WebSocket rejected: too many connections");
// Decrement the WS tracker we already incremented above.
if let Some(ref tracker) = tracker_for_drop {
@@ -117,7 +121,7 @@ pub async fn handle_ws_connection(socket: WebSocket, state: Arc<GatewayState>) {
});
// Receiver task: read client frames and route to agent
let user_id = state.user_id.clone();
let user_id = user.user_id;
while let Some(Ok(frame)) = ws_stream.next().await {
match frame {
Message::Text(text) => {
@@ -263,10 +267,11 @@ async fn handle_client_message(
token,
} => {
if let Some(ref ext_mgr) = state.extension_manager {
match ext_mgr.configure_token(&extension_name, &token).await {
match ext_mgr.configure_token(&extension_name, &token, user_id).await {
Ok(result) => {
if result.verification.is_some() {
state.sse.broadcast(
state.sse.broadcast_for_user(
user_id,
crate::channels::web::types::SseEvent::AuthRequired {
extension_name: extension_name.clone(),
instructions: Some(result.message),
@@ -275,8 +280,9 @@ async fn handle_client_message(
},
);
} else {
crate::channels::web::server::clear_auth_mode(state).await;
state.sse.broadcast(
crate::channels::web::server::clear_auth_mode(state, user_id).await;
state.sse.broadcast_for_user(
user_id,
crate::channels::web::types::SseEvent::AuthCompleted {
extension_name,
success: true,
@@ -288,7 +294,8 @@ async fn handle_client_message(
Err(e) => {
let msg = format!("Auth failed: {}", e);
if matches!(e, crate::extensions::ExtensionError::ValidationFailed(_)) {
state.sse.broadcast(
state.sse.broadcast_for_user(
user_id,
crate::channels::web::types::SseEvent::AuthRequired {
extension_name: extension_name.clone(),
instructions: Some(msg.clone()),
@@ -311,7 +318,7 @@ async fn handle_client_message(
}
}
WsClientMessage::AuthCancel { .. } => {
crate::channels::web::server::clear_auth_mode(state).await;
crate::channels::web::server::clear_auth_mode(state, user_id).await;
}
WsClientMessage::Ping => {
let _ = direct_tx.send(WsServerMessage::Pong).await;
@@ -498,8 +505,9 @@ mod tests {
GatewayState {
msg_tx: tokio::sync::RwLock::new(msg_tx),
sse: SseManager::new(),
sse: Arc::new(SseManager::new()),
workspace: None,
workspace_pool: None,
session_manager: None,
log_broadcaster: None,
log_level_handle: None,
@@ -509,13 +517,13 @@ mod tests {
job_manager: None,
prompt_queue: None,
scheduler: None,
user_id: "test".to_string(),
default_user_id: "test".to_string(),
shutdown_tx: tokio::sync::RwLock::new(None),
ws_tracker: Some(Arc::new(WsConnectionTracker::new())),
llm_provider: None,
skill_registry: None,
skill_catalog: None,
chat_rate_limiter: crate::channels::web::server::RateLimiter::new(30, 60),
chat_rate_limiter: crate::channels::web::server::PerUserRateLimiter::new(30, 60),
oauth_rate_limiter: crate::channels::web::server::RateLimiter::new(10, 60),
webhook_rate_limiter: crate::channels::web::server::RateLimiter::new(10, 60),
registry_entries: Vec::new(),
src/cli/oauth_defaults.rs
+2 -2
View File
@@ -447,8 +447,8 @@ pub struct PendingOAuthFlow {
pub user_id: String,
/// Secrets store reference for token persistence.
pub secrets: Arc<dyn SecretsStore + Send + Sync>,
/// SSE broadcast sender for notifying the web UI.
pub sse_sender: Option<tokio::sync::broadcast::Sender<crate::channels::web::types::SseEvent>>,
/// SSE broadcast manager for notifying the web UI.
pub sse_manager: Option<Arc<crate::channels::web::sse::SseManager>>,
/// Gateway auth token for authenticating with the platform token exchange proxy.
pub gateway_token: Option<String>,
/// Additional form params for the token exchange request.
src/config/channels.rs
+149
View File
@@ -2,6 +2,7 @@ use std::collections::HashMap;
use std::path::PathBuf;
use secrecy::SecretString;
use serde::Deserialize;
use crate::bootstrap::ironclaw_base_dir;
use crate::config::helpers::{optional_env, parse_bool_env, parse_optional_env};
@@ -45,6 +46,26 @@ pub struct GatewayConfig {
/// Bearer token for authentication. Random hex generated at startup if unset.
pub auth_token: Option<String>,
pub user_id: String,
/// Additional user scopes for workspace reads.
///
/// When set, the workspace will be able to read (search, read, list) from
/// these additional user scopes while writes remain isolated to `user_id`.
/// Parsed from `WORKSPACE_READ_SCOPES` (comma-separated).
pub workspace_read_scopes: Vec<String>,
/// Memory layer definitions (JSON in env var, or from external config).
pub memory_layers: Vec<crate::workspace::layer::MemoryLayer>,
/// Multi-user token map. When set, each token maps to a user identity.
/// Parsed from `GATEWAY_USER_TOKENS` (JSON string). When absent, falls back
/// to single-user mode via `auth_token` + `user_id`.
pub user_tokens: Option<HashMap<String, UserTokenConfig>>,
}
/// Per-user token configuration for multi-user mode.
#[derive(Debug, Clone, Deserialize)]
pub struct UserTokenConfig {
pub user_id: String,
#[serde(default)]
pub workspace_read_scopes: Vec<String>,
}
/// Signal channel configuration (signal-cli daemon HTTP/JSON-RPC).
@@ -115,6 +136,125 @@ impl ChannelsConfig {
.or_else(|| cs.gateway_user_id.clone())
.unwrap_or_else(|| owner_id.to_string());
let memory_layers: Vec<crate::workspace::layer::MemoryLayer> =
match optional_env("MEMORY_LAYERS")? {
Some(json_str) => {
serde_json::from_str(&json_str).map_err(|e| ConfigError::InvalidValue {
key: "MEMORY_LAYERS".to_string(),
message: format!("must be valid JSON array of layer objects: {e}"),
})?
}
None => crate::workspace::layer::MemoryLayer::default_for_user(&user_id),
};
// Validate layer names and scopes
for layer in &memory_layers {
if layer.name.trim().is_empty() {
return Err(ConfigError::InvalidValue {
key: "MEMORY_LAYERS".to_string(),
message: "layer name must not be empty".to_string(),
});
}
if layer.name.len() > 64 {
return Err(ConfigError::InvalidValue {
key: "MEMORY_LAYERS".to_string(),
message: format!(
"layer name '{}' exceeds 64 characters",
layer.name
),
});
}
if !layer
.name
.chars()
.all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '-')
{
return Err(ConfigError::InvalidValue {
key: "MEMORY_LAYERS".to_string(),
message: format!(
"layer name '{}' contains invalid characters \
(allowed: a-z, A-Z, 0-9, _, -)",
layer.name
),
});
}
if layer.scope.trim().is_empty() {
return Err(ConfigError::InvalidValue {
key: "MEMORY_LAYERS".to_string(),
message: format!(
"layer '{}' has an empty scope",
layer.name
),
});
}
}
// Check for duplicate layer names
{
let mut seen = std::collections::HashSet::new();
for layer in &memory_layers {
if !seen.insert(&layer.name) {
return Err(ConfigError::InvalidValue {
key: "MEMORY_LAYERS".to_string(),
message: format!("duplicate layer name '{}'", layer.name),
});
}
}
}
let user_tokens: Option<HashMap<String, UserTokenConfig>> =
match optional_env("GATEWAY_USER_TOKENS")? {
Some(json_str) => {
let tokens: HashMap<String, UserTokenConfig> =
serde_json::from_str(&json_str).map_err(|e| {
ConfigError::InvalidValue {
key: "GATEWAY_USER_TOKENS".to_string(),
message: format!(
"must be valid JSON object mapping tokens to user configs: {e}"
),
}
})?;
if tokens.is_empty() {
return Err(ConfigError::InvalidValue {
key: "GATEWAY_USER_TOKENS".to_string(),
message: "token map is empty — remove the variable to use single-user mode".to_string(),
});
}
for (tok, cfg) in &tokens {
if cfg.user_id.trim().is_empty() {
return Err(ConfigError::InvalidValue {
key: "GATEWAY_USER_TOKENS".to_string(),
message: format!(
"token '{}...' has an empty user_id",
&tok[..tok.len().min(8)]
),
});
}
}
Some(tokens)
}
None => None,
};
let workspace_read_scopes: Vec<String> = optional_env("WORKSPACE_READ_SCOPES")?
.map(|s| {
s.split(',')
.map(|s| s.trim().to_string())
.filter(|s| !s.is_empty())
.collect()
})
.unwrap_or_default();
for scope in &workspace_read_scopes {
if scope.len() > 128 {
return Err(ConfigError::InvalidValue {
key: "WORKSPACE_READ_SCOPES".to_string(),
message: format!(
"scope '{}...' exceeds 128 characters",
&scope[..32]
),
});
}
}
Some(GatewayConfig {
host: optional_env("GATEWAY_HOST")?
.or_else(|| cs.gateway_host.clone())
@@ -126,6 +266,9 @@ impl ChannelsConfig {
auth_token: optional_env("GATEWAY_AUTH_TOKEN")?
.or_else(|| cs.gateway_auth_token.clone()),
user_id,
workspace_read_scopes,
memory_layers,
user_tokens,
})
} else {
None
@@ -281,6 +424,9 @@ mod tests {
port: 3000,
auth_token: Some("tok-abc".to_string()),
user_id: "default".to_string(),
workspace_read_scopes: vec![],
memory_layers: vec![],
user_tokens: None,
};
assert_eq!(cfg.host, "127.0.0.1");
assert_eq!(cfg.port, 3000);
@@ -295,6 +441,9 @@ mod tests {
port: 3001,
auth_token: None,
user_id: "anon".to_string(),
workspace_read_scopes: vec![],
memory_layers: vec![],
user_tokens: None,
};
assert!(cfg.auth_token.is_none());
}
src/extensions/manager.rs
+234 -225
View File
File diff suppressed because it is too large Load Diff
src/main.rs
+59 -15
View File
@@ -589,15 +589,42 @@ async fn async_main() -> anyhow::Result<()> {
// ── Gateway channel ────────────────────────────────────────────────
let mut gateway_url: Option<String> = None;
let mut sse_sender: Option<
tokio::sync::broadcast::Sender<ironclaw::channels::web::types::SseEvent>,
> = None;
let mut sse_manager: Option<std::sync::Arc<ironclaw::channels::web::sse::SseManager>> = None;
let mut _gateway_state: Option<std::sync::Arc<ironclaw::channels::web::server::GatewayState>> =
None;
if let Some(ref gw_config) = config.channels.gateway {
let mut gw =
GatewayChannel::new(gw_config.clone()).with_llm_provider(Arc::clone(&components.llm));
// Build multi-user auth state if user_tokens is configured, else single-user.
let mut gw = if let Some(ref user_tokens) = gw_config.user_tokens {
use ironclaw::channels::web::auth::{MultiAuthState, UserIdentity};
let tokens = user_tokens
.iter()
.map(|(token, cfg)| {
(
token.clone(),
UserIdentity {
user_id: cfg.user_id.clone(),
workspace_read_scopes: cfg.workspace_read_scopes.clone(),
},
)
})
.collect();
let auth = MultiAuthState::multi(tokens);
GatewayChannel::new_multi_auth(gw_config.clone(), auth)
} else {
GatewayChannel::new(gw_config.clone())
};
gw = gw.with_llm_provider(Arc::clone(&components.llm));
if let Some(ref ws) = components.workspace {
gw = gw.with_workspace(Arc::clone(ws));
}
// Create per-user workspace pool for multi-user mode.
if let Some(ref db) = components.db {
let pool = Arc::new(ironclaw::channels::web::server::WorkspacePool::new(
Arc::clone(db),
components.embeddings.clone(),
));
gw = gw.with_workspace_pool(pool);
}
gw = gw.with_session_manager(Arc::clone(&session_manager));
gw = gw.with_log_broadcaster(Arc::clone(&log_broadcaster));
gw = gw.with_log_level_handle(Arc::clone(&log_level_handle));
@@ -648,8 +675,12 @@ async fn async_main() -> anyhow::Result<()> {
let mut rx = tx.subscribe();
let gw_state = Arc::clone(gw.state());
tokio::spawn(async move {
while let Ok((_job_id, event)) = rx.recv().await {
gw_state.sse.broadcast(event);
while let Ok((_job_id, user_id, event)) = rx.recv().await {
if user_id.is_empty() {
gw_state.sse.broadcast(event);
} else {
gw_state.sse.broadcast_for_user(&user_id, event);
}
}
});
}
@@ -691,7 +722,8 @@ async fn async_main() -> anyhow::Result<()> {
// Capture SSE sender and routine engine slot before moving gw into channels.
// IMPORTANT: This must come after all `with_*` calls since `rebuild_state`
// creates a new SseManager, which would orphan this sender.
sse_sender = Some(gw.state().sse.sender());
sse_manager = Some(Arc::clone(&gw.state().sse));
_gateway_state = Some(Arc::clone(gw.state()));
channel_names.push("gateway".to_string());
channels.add(Box::new(gw)).await;
}
@@ -774,12 +806,18 @@ async fn async_main() -> anyhow::Result<()> {
// Auto-activate WASM channels that were active in a previous session.
// Relay channels are handled separately below via restore_relay_channels().
let persisted = ext_mgr.load_persisted_active_channels().await;
let ext_user_id = config
.channels
.gateway
.as_ref()
.map(|g| g.user_id.clone())
.unwrap_or_else(|| "default".to_string());
let persisted = ext_mgr.load_persisted_active_channels(&ext_user_id).await;
for name in &persisted {
if active_at_startup.contains(name) || ext_mgr.is_relay_channel(name).await {
if active_at_startup.contains(name) || ext_mgr.is_relay_channel(name, &ext_user_id).await {
continue;
}
match ext_mgr.activate(name).await {
match ext_mgr.activate(name, &ext_user_id).await {
Ok(result) => {
tracing::debug!(
channel = %name,
@@ -804,14 +842,20 @@ async fn async_main() -> anyhow::Result<()> {
ext_mgr
.set_relay_channel_manager(Arc::clone(&channels))
.await;
ext_mgr.restore_relay_channels().await;
let ext_user_id = config
.channels
.gateway
.as_ref()
.map(|g| g.user_id.clone())
.unwrap_or_else(|| "default".to_string());
ext_mgr.restore_relay_channels(&ext_user_id).await;
}
// Wire SSE sender into extension manager for broadcasting status events.
if let Some(ref ext_mgr) = components.extension_manager
&& let Some(ref sender) = sse_sender
&& let Some(sse) = sse_manager
{
ext_mgr.set_sse_sender(sender.clone()).await;
ext_mgr.set_sse_sender(sse).await;
}
// Snapshot memory for trace recording before the agent starts
@@ -849,7 +893,7 @@ async fn async_main() -> anyhow::Result<()> {
skills_config: config.skills.clone(),
hooks: components.hooks,
cost_guard: components.cost_guard,
sse_tx: sse_sender,
sse_tx: None, // TODO: wire SseManager into scheduler (needs Sender<SseEvent> → Arc<SseManager> refactor)
http_interceptor,
transcription: config.transcription.create_provider().map(|p| {
Arc::new(ironclaw::llm::transcription::TranscriptionMiddleware::new(
src/orchestrator/api.rs
+24 -6
View File
@@ -40,7 +40,8 @@ pub struct OrchestratorState {
pub job_manager: Arc<ContainerJobManager>,
pub token_store: TokenStore,
/// Broadcast channel for job events (consumed by the web gateway SSE).
pub job_event_tx: Option<broadcast::Sender<(Uuid, SseEvent)>>,
/// Tuple: (job_id, user_id, event).
pub job_event_tx: Option<broadcast::Sender<(Uuid, String, SseEvent)>>,
/// Buffered follow-up prompts for sandbox jobs, keyed by job_id.
pub prompt_queue: Arc<Mutex<HashMap<Uuid, VecDeque<PendingPrompt>>>>,
/// Database handle for persisting job events.
@@ -351,9 +352,24 @@ async fn job_event_handler(
},
};
// Broadcast via the channel (if configured)
// Broadcast via the channel (if configured).
// Look up the job owner so the gateway can scope delivery per-user.
if let Some(ref tx) = state.job_event_tx {
let _ = tx.send((job_id, sse_event));
let user_id = match state.store.as_ref() {
Some(store) => store
.get_sandbox_job(job_id)
.await
.ok()
.flatten()
.map(|j| j.user_id),
None => None,
};
if let Some(uid) = user_id {
let _ = tx.send((job_id, uid, sse_event));
} else {
// Fallback: broadcast globally (single-user mode or job not found).
let _ = tx.send((job_id, String::new(), sse_event));
}
}
Ok(StatusCode::OK)
@@ -769,8 +785,10 @@ mod tests {
let resp = router.oneshot(req).await.unwrap();
assert_eq!(resp.status(), StatusCode::OK);
let (recv_id, event) = rx.recv().await.unwrap();
let (recv_id, recv_uid, event) = rx.recv().await.unwrap();
assert_eq!(recv_id, job_id);
// No store configured, so user_id falls back to empty string.
assert_eq!(recv_uid, "");
match event {
SseEvent::JobMessage {
job_id: jid,
@@ -824,7 +842,7 @@ mod tests {
let resp = router.oneshot(req).await.unwrap();
assert_eq!(resp.status(), StatusCode::OK);
let (_recv_id, event) = rx.recv().await.unwrap();
let (_recv_id, _recv_uid, event) = rx.recv().await.unwrap();
match event {
SseEvent::JobToolUse { tool_name, .. } => {
assert_eq!(tool_name, "shell");
@@ -869,7 +887,7 @@ mod tests {
let resp = router.oneshot(req).await.unwrap();
assert_eq!(resp.status(), StatusCode::OK);
let (_recv_id, event) = rx.recv().await.unwrap();
let (_recv_id, _recv_uid, event) = rx.recv().await.unwrap();
// Unknown event types fall through to JobStatus
assert!(matches!(event, SseEvent::JobStatus { .. }));
}
src/orchestrator/mod.rs
+1 -1
View File
@@ -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<Arc<ContainerJobManager>>,
pub job_event_tx: Option<broadcast::Sender<(Uuid, SseEvent)>>,
pub job_event_tx: Option<broadcast::Sender<(Uuid, String, SseEvent)>>,
pub prompt_queue: Arc<Mutex<HashMap<Uuid, VecDeque<api::PendingPrompt>>>>,
pub docker_status: crate::sandbox::DockerStatus,
}
src/tools/builtin/extension_tools.rs
+17 -17
View File
@@ -130,7 +130,7 @@ impl Tool for ToolInstallTool {
async fn execute(
&self,
params: serde_json::Value,
_ctx: &JobContext,
ctx: &JobContext,
) -> Result<ToolOutput, ToolError> {
let start = std::time::Instant::now();
@@ -150,7 +150,7 @@ impl Tool for ToolInstallTool {
let result = self
.manager
.install(name, url, kind_hint)
.install(name, url, kind_hint, &ctx.user_id)
.await
.map_err(|e| ToolError::ExecutionFailed(e.to_string()))?;
@@ -205,7 +205,7 @@ impl Tool for ToolAuthTool {
async fn execute(
&self,
params: serde_json::Value,
_ctx: &JobContext,
ctx: &JobContext,
) -> Result<ToolOutput, ToolError> {
let start = std::time::Instant::now();
@@ -213,13 +213,13 @@ impl Tool for ToolAuthTool {
let result = self
.manager
.auth(name)
.auth(name, &ctx.user_id)
.await
.map_err(|e| ToolError::ExecutionFailed(e.to_string()))?;
// Auto-activate after successful auth so tools are available immediately
if result.is_authenticated() {
match self.manager.activate(name).await {
match self.manager.activate(name, &ctx.user_id).await {
Ok(activate_result) => {
let output = serde_json::json!({
"status": "authenticated_and_activated",
@@ -304,13 +304,13 @@ impl Tool for ToolActivateTool {
async fn execute(
&self,
params: serde_json::Value,
_ctx: &JobContext,
ctx: &JobContext,
) -> Result<ToolOutput, ToolError> {
let start = std::time::Instant::now();
let name = require_str(&params, "name")?;
match self.manager.activate(name).await {
match self.manager.activate(name, &ctx.user_id).await {
Ok(result) => {
let output = serde_json::to_value(&result)
.unwrap_or_else(|_| serde_json::json!({"error": "serialization failed"}));
@@ -329,12 +329,12 @@ impl Tool for ToolActivateTool {
// Activation failed due to missing auth; initiate auth flow
// so the agent loop can show the auth card.
match self.manager.auth(name).await {
match self.manager.auth(name, &ctx.user_id).await {
Ok(auth_result) if auth_result.is_authenticated() => {
// Auth succeeded (e.g. env var was set); retry activation.
let result = self
.manager
.activate(name)
.activate(name, &ctx.user_id)
.await
.map_err(|e| ToolError::ExecutionFailed(e.to_string()))?;
let output = serde_json::to_value(&result).unwrap_or_else(
@@ -404,7 +404,7 @@ impl Tool for ToolListTool {
async fn execute(
&self,
params: serde_json::Value,
_ctx: &JobContext,
ctx: &JobContext,
) -> Result<ToolOutput, ToolError> {
let start = std::time::Instant::now();
@@ -425,7 +425,7 @@ impl Tool for ToolListTool {
let extensions = self
.manager
.list(kind_filter, include_available)
.list(kind_filter, include_available, &ctx.user_id)
.await
.map_err(|e| ToolError::ExecutionFailed(e.to_string()))?;
@@ -477,7 +477,7 @@ impl Tool for ToolRemoveTool {
async fn execute(
&self,
params: serde_json::Value,
_ctx: &JobContext,
ctx: &JobContext,
) -> Result<ToolOutput, ToolError> {
let start = std::time::Instant::now();
@@ -485,7 +485,7 @@ impl Tool for ToolRemoveTool {
let message = self
.manager
.remove(name)
.remove(name, &ctx.user_id)
.await
.map_err(|e| ToolError::ExecutionFailed(e.to_string()))?;
@@ -541,7 +541,7 @@ impl Tool for ToolUpgradeTool {
async fn execute(
&self,
params: serde_json::Value,
_ctx: &JobContext,
ctx: &JobContext,
) -> Result<ToolOutput, ToolError> {
let start = std::time::Instant::now();
@@ -549,7 +549,7 @@ impl Tool for ToolUpgradeTool {
let result = self
.manager
.upgrade(name)
.upgrade(name, &ctx.user_id)
.await
.map_err(|e| ToolError::ExecutionFailed(e.to_string()))?;
@@ -603,7 +603,7 @@ impl Tool for ExtensionInfoTool {
async fn execute(
&self,
params: serde_json::Value,
_ctx: &JobContext,
ctx: &JobContext,
) -> Result<ToolOutput, ToolError> {
let start = std::time::Instant::now();
@@ -611,7 +611,7 @@ impl Tool for ExtensionInfoTool {
let info = self
.manager
.extension_info(name)
.extension_info(name, &ctx.user_id)
.await
.map_err(|e| ToolError::ExecutionFailed(e.to_string()))?;
src/tools/builtin/job.rs
+2 -2
View File
@@ -85,7 +85,7 @@ pub struct CreateJobTool {
job_manager: Option<Arc<ContainerJobManager>>,
store: Option<Arc<dyn Database>>,
/// Broadcast sender for job events (used to subscribe a monitor).
event_tx: Option<tokio::sync::broadcast::Sender<(Uuid, SseEvent)>>,
event_tx: Option<tokio::sync::broadcast::Sender<(Uuid, String, SseEvent)>>,
/// Injection channel for pushing messages into the agent loop.
inject_tx: Option<tokio::sync::mpsc::Sender<IncomingMessage>>,
/// 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, SseEvent)>,
event_tx: tokio::sync::broadcast::Sender<(Uuid, String, SseEvent)>,
inject_tx: tokio::sync::mpsc::Sender<IncomingMessage>,
) -> Self {
self.event_tx = Some(event_tx);
src/tools/registry.rs
+1 -1
View File
@@ -361,7 +361,7 @@ impl ToolRegistry {
job_manager: Option<Arc<ContainerJobManager>>,
store: Option<Arc<dyn Database>>,
job_event_tx: Option<
tokio::sync::broadcast::Sender<(uuid::Uuid, crate::channels::web::types::SseEvent)>,
tokio::sync::broadcast::Sender<(uuid::Uuid, String, crate::channels::web::types::SseEvent)>,
>,
inject_tx: Option<tokio::sync::mpsc::Sender<crate::channels::IncomingMessage>>,
prompt_queue: Option<PromptQueue>,
tests/e2e_advanced_traces.rs
+1 -1
View File
@@ -661,7 +661,7 @@ mod advanced {
.await
.expect("failed to inject test token");
let activate_result = ext_mgr.activate("mock-notion").await;
let activate_result = ext_mgr.activate("mock-notion", "default").await;
assert!(
activate_result.is_ok(),
"activation failed: {:?}",
tests/module_init_integration.rs
+1 -1
View File
@@ -216,7 +216,7 @@ async fn extension_manager_with_process_manager_constructs() {
);
// Verify the manager is functional — list returns Ok.
let result = manager.list(None, false).await;
let result = manager.list(None, false, "test").await;
assert!(result.is_ok(), "list should succeed on empty manager");
assert!(result.unwrap().is_empty());
}
tests/multi_tenant_integration.rs
+1057
View File
File diff suppressed because it is too large Load Diff
tests/multi_tenant_system_prompt.rs
+254
View File
@@ -0,0 +1,254 @@
//! Tests proving that multi-tenant system prompts are broken.
//!
//! Bug: In multi-tenant mode, the agent loop uses `self.workspace()` which
//! returns a single shared workspace (user_id="default"). Identity files
//! (IDENTITY.md, SOUL.md, USER.md) seeded under per-user IDs ("alice",
//! "bob") are invisible to this workspace, so the system prompt is
//! empty/wrong.
//!
//! These tests:
//! 1. Seed identity files for two users (alice, bob) in the database
//! 2. Send messages as each user
//! 3. Verify the system prompt in captured LLM requests contains the
//! correct user's identity
//! 4. Verify user A's identity doesn't leak into user B's prompt
//!
//! All tests are expected to FAIL until the bug is fixed.
#[cfg(feature = "libsql")]
mod support;
#[cfg(feature = "libsql")]
mod tests {
use std::sync::Arc;
use std::time::Duration;
use ironclaw::channels::IncomingMessage;
use ironclaw::llm::Role;
use ironclaw::workspace::Workspace;
use crate::support::test_rig::TestRigBuilder;
use crate::support::trace_llm::{LlmTrace, TraceResponse, TraceStep};
const TIMEOUT: Duration = Duration::from_secs(15);
const ALICE_USER_ID: &str = "alice";
const BOB_USER_ID: &str = "bob";
const ALICE_IDENTITY: &str = "You are Alice's personal assistant. \
Alice is a software engineer who lives in Seattle.";
const BOB_IDENTITY: &str = "You are Bob's personal assistant. \
Bob is a marine biologist who lives in Miami.";
/// Create a simple trace that returns a canned text response.
/// We need one step per message we plan to send.
fn simple_trace(num_steps: usize) -> LlmTrace {
let steps: Vec<TraceStep> = (0..num_steps)
.map(|i| TraceStep {
request_hint: None,
response: TraceResponse::Text {
content: format!("Response {}", i),
input_tokens: 100,
output_tokens: 10,
},
expected_tool_results: Vec::new(),
})
.collect();
// Create separate turns for each step so the trace replays correctly.
let turns: Vec<crate::support::trace_llm::TraceTurn> = steps
.into_iter()
.enumerate()
.map(|(i, step)| crate::support::trace_llm::TraceTurn {
user_input: format!("message {}", i),
steps: vec![step],
expects: Default::default(),
})
.collect();
LlmTrace::new("test-model", turns)
}
/// Seed identity files for a user by creating a workspace scoped to that
/// user and writing IDENTITY.md.
async fn seed_identity(db: &Arc<dyn ironclaw::db::Database>, user_id: &str, content: &str) {
let ws = Workspace::new_with_db(user_id, db.clone());
ws.write("IDENTITY.md", content)
.await
.unwrap_or_else(|e| panic!("Failed to seed IDENTITY.md for {user_id}: {e}"));
}
/// Extract the system prompt from captured LLM requests.
///
/// The system prompt is the first message with role=System in the first
/// LLM request for a given turn.
fn extract_system_prompt(requests: &[Vec<ironclaw::llm::ChatMessage>]) -> Option<String> {
requests.last().and_then(|msgs| {
msgs.iter()
.find(|m| matches!(m.role, Role::System))
.map(|m| m.content.clone())
})
}
// -----------------------------------------------------------------------
// Test 1: Alice's identity should appear in system prompt when messaging
// as Alice.
// -----------------------------------------------------------------------
#[tokio::test]
async fn alice_system_prompt_contains_alice_identity() {
let trace = simple_trace(1);
let rig = TestRigBuilder::new()
.with_trace(trace)
.build()
.await;
// Seed alice's identity into the database
let db = rig.database();
seed_identity(db, ALICE_USER_ID, ALICE_IDENTITY).await;
// Send a message AS alice (using her user_id)
let msg = IncomingMessage::new("test", ALICE_USER_ID, "Hello, who am I?");
rig.send_incoming(msg).await;
let _responses = rig.wait_for_responses(1, TIMEOUT).await;
// The system prompt sent to the LLM should contain Alice's identity
let requests = rig.captured_llm_requests();
let system_prompt = extract_system_prompt(&requests)
.expect("Expected a system prompt in the LLM request");
assert!(
system_prompt.contains("Alice is a software engineer"),
"System prompt should contain Alice's identity when messaging as Alice.\n\
Actual system prompt:\n{system_prompt}"
);
rig.shutdown();
}
// -----------------------------------------------------------------------
// Test 2: Bob's identity should appear in system prompt when messaging
// as Bob.
// -----------------------------------------------------------------------
#[tokio::test]
async fn bob_system_prompt_contains_bob_identity() {
let trace = simple_trace(1);
let rig = TestRigBuilder::new()
.with_trace(trace)
.build()
.await;
// Seed bob's identity into the database
let db = rig.database();
seed_identity(db, BOB_USER_ID, BOB_IDENTITY).await;
// Send a message AS bob
let msg = IncomingMessage::new("test", BOB_USER_ID, "Hello, who am I?");
rig.send_incoming(msg).await;
let _responses = rig.wait_for_responses(1, TIMEOUT).await;
// The system prompt should contain Bob's identity
let requests = rig.captured_llm_requests();
let system_prompt = extract_system_prompt(&requests)
.expect("Expected a system prompt in the LLM request");
assert!(
system_prompt.contains("Bob is a marine biologist"),
"System prompt should contain Bob's identity when messaging as Bob.\n\
Actual system prompt:\n{system_prompt}"
);
rig.shutdown();
}
// -----------------------------------------------------------------------
// Test 3: Alice's identity must NOT appear in Bob's system prompt.
// -----------------------------------------------------------------------
#[tokio::test]
async fn alice_identity_does_not_leak_into_bob_prompt() {
let trace = simple_trace(1);
let rig = TestRigBuilder::new()
.with_trace(trace)
.build()
.await;
// Seed BOTH users' identities
let db = rig.database();
seed_identity(db, ALICE_USER_ID, ALICE_IDENTITY).await;
seed_identity(db, BOB_USER_ID, BOB_IDENTITY).await;
// Send a message AS bob
let msg = IncomingMessage::new("test", BOB_USER_ID, "Tell me about myself");
rig.send_incoming(msg).await;
let _responses = rig.wait_for_responses(1, TIMEOUT).await;
// Bob's prompt must NOT contain Alice's identity
let requests = rig.captured_llm_requests();
let system_prompt = extract_system_prompt(&requests);
if let Some(ref prompt) = system_prompt {
assert!(
!prompt.contains("Alice is a software engineer"),
"Alice's identity LEAKED into Bob's system prompt!\n\
System prompt:\n{prompt}"
);
}
// Also verify Bob's identity IS present (compound check)
let prompt = system_prompt
.expect("Expected a system prompt in the LLM request");
assert!(
prompt.contains("Bob is a marine biologist"),
"Bob's own identity should be in his system prompt.\n\
Actual system prompt:\n{prompt}"
);
rig.shutdown();
}
// -----------------------------------------------------------------------
// Test 4: Bob's identity must NOT appear in Alice's system prompt.
// -----------------------------------------------------------------------
#[tokio::test]
async fn bob_identity_does_not_leak_into_alice_prompt() {
let trace = simple_trace(1);
let rig = TestRigBuilder::new()
.with_trace(trace)
.build()
.await;
// Seed BOTH users' identities
let db = rig.database();
seed_identity(db, ALICE_USER_ID, ALICE_IDENTITY).await;
seed_identity(db, BOB_USER_ID, BOB_IDENTITY).await;
// Send a message AS alice
let msg = IncomingMessage::new("test", ALICE_USER_ID, "Tell me about myself");
rig.send_incoming(msg).await;
let _responses = rig.wait_for_responses(1, TIMEOUT).await;
// Alice's prompt must NOT contain Bob's identity
let requests = rig.captured_llm_requests();
let system_prompt = extract_system_prompt(&requests);
if let Some(ref prompt) = system_prompt {
assert!(
!prompt.contains("Bob is a marine biologist"),
"Bob's identity LEAKED into Alice's system prompt!\n\
System prompt:\n{prompt}"
);
}
// Also verify Alice's identity IS present
let prompt = system_prompt
.expect("Expected a system prompt in the LLM request");
assert!(
prompt.contains("Alice is a software engineer"),
"Alice's own identity should be in her system prompt.\n\
Actual system prompt:\n{prompt}"
);
rig.shutdown();
}
}
tests/openai_compat_integration.rs
+22 -11
View File
@@ -191,8 +191,9 @@ async fn start_test_server_with_provider(
) -> (SocketAddr, Arc<GatewayState>) {
let state = Arc::new(GatewayState {
msg_tx: tokio::sync::RwLock::new(None),
sse: SseManager::new(),
sse: Arc::new(SseManager::new()),
workspace: None,
workspace_pool: None,
session_manager: None,
log_broadcaster: None,
log_level_handle: None,
@@ -202,13 +203,13 @@ async fn start_test_server_with_provider(
job_manager: None,
prompt_queue: None,
scheduler: None,
user_id: "test-user".to_string(),
default_user_id: "test-user".to_string(),
shutdown_tx: tokio::sync::RwLock::new(None),
ws_tracker: Some(Arc::new(WsConnectionTracker::new())),
llm_provider: Some(llm_provider),
skill_registry: None,
skill_catalog: None,
chat_rate_limiter: ironclaw::channels::web::server::RateLimiter::new(30, 60),
chat_rate_limiter: ironclaw::channels::web::server::PerUserRateLimiter::new(30, 60),
oauth_rate_limiter: ironclaw::channels::web::server::RateLimiter::new(10, 60),
webhook_rate_limiter: ironclaw::channels::web::server::RateLimiter::new(10, 60),
registry_entries: Vec::new(),
@@ -218,8 +219,12 @@ async fn start_test_server_with_provider(
active_config: ironclaw::channels::web::server::ActiveConfigSnapshot::default(),
});
let auth = ironclaw::channels::web::auth::MultiAuthState::single(
AUTH_TOKEN.to_string(),
"test-user".to_string(),
);
let addr: SocketAddr = "127.0.0.1:0".parse().unwrap();
let bound_addr = start_server(addr, state.clone(), AUTH_TOKEN.to_string())
let bound_addr = start_server(addr, state.clone(), auth)
.await
.expect("Failed to start test server");
@@ -684,8 +689,9 @@ async fn test_no_llm_provider_returns_503() {
// Create state WITHOUT llm_provider
let state = Arc::new(GatewayState {
msg_tx: tokio::sync::RwLock::new(None),
sse: SseManager::new(),
sse: Arc::new(SseManager::new()),
workspace: None,
workspace_pool: None,
session_manager: None,
log_broadcaster: None,
log_level_handle: None,
@@ -695,13 +701,13 @@ async fn test_no_llm_provider_returns_503() {
job_manager: None,
prompt_queue: None,
scheduler: None,
user_id: "test-user".to_string(),
default_user_id: "test-user".to_string(),
shutdown_tx: tokio::sync::RwLock::new(None),
ws_tracker: Some(Arc::new(WsConnectionTracker::new())),
llm_provider: None, // No LLM!
skill_registry: None,
skill_catalog: None,
chat_rate_limiter: ironclaw::channels::web::server::RateLimiter::new(30, 60),
chat_rate_limiter: ironclaw::channels::web::server::PerUserRateLimiter::new(30, 60),
oauth_rate_limiter: ironclaw::channels::web::server::RateLimiter::new(10, 60),
webhook_rate_limiter: ironclaw::channels::web::server::RateLimiter::new(10, 60),
registry_entries: Vec::new(),
@@ -711,8 +717,12 @@ async fn test_no_llm_provider_returns_503() {
active_config: ironclaw::channels::web::server::ActiveConfigSnapshot::default(),
});
let auth = ironclaw::channels::web::auth::MultiAuthState::single(
AUTH_TOKEN.to_string(),
"test-user".to_string(),
);
let addr: SocketAddr = "127.0.0.1:0".parse().unwrap();
let bound_addr = start_server(addr, state, AUTH_TOKEN.to_string())
let bound_addr = start_server(addr, state, auth)
.await
.unwrap();
@@ -741,9 +751,10 @@ async fn test_chat_completions_body_too_large() {
let state = ironclaw::channels::web::test_helpers::TestGatewayBuilder::new()
.llm_provider(llm_provider)
.build();
let auth_state = ironclaw::channels::web::auth::AuthState {
token: AUTH_TOKEN.to_string(),
};
let auth_state = ironclaw::channels::web::auth::MultiAuthState::single(
AUTH_TOKEN.to_string(),
"test-user".to_string(),
);
let app = Router::new()
.route(
tests/support/gateway_workflow_harness.rs
+9 -6
View File
@@ -14,7 +14,8 @@ use ironclaw::agent::{Agent, AgentDeps, SessionManager as AgentSessionManager};
use ironclaw::app::{AppBuilder, AppBuilderFlags};
use ironclaw::channels::IncomingMessage;
use ironclaw::channels::web::log_layer::LogBroadcaster;
use ironclaw::channels::web::server::{GatewayState, RateLimiter, start_server};
use ironclaw::channels::web::auth::MultiAuthState;
use ironclaw::channels::web::server::{GatewayState, PerUserRateLimiter, RateLimiter, start_server};
use ironclaw::channels::web::sse::SseManager;
use ironclaw::channels::web::ws::WsConnectionTracker;
use ironclaw::config::{Config, RegistryProviderConfig, RoutineConfig};
@@ -211,8 +212,9 @@ impl GatewayWorkflowHarness {
let gateway_state = Arc::new(GatewayState {
msg_tx: tokio::sync::RwLock::new(Some(gw_tx)),
sse: SseManager::new(),
sse: Arc::new(SseManager::new()),
workspace: components.workspace.clone(),
workspace_pool: None,
session_manager: Some(Arc::clone(&agent_session_manager)),
log_broadcaster: None,
log_level_handle: None,
@@ -222,13 +224,13 @@ impl GatewayWorkflowHarness {
job_manager: None,
prompt_queue: None,
scheduler: Some(scheduler_slot.clone()),
user_id: user_id.clone(),
default_user_id: user_id.clone(),
shutdown_tx: tokio::sync::RwLock::new(None),
ws_tracker: Some(Arc::new(WsConnectionTracker::new())),
llm_provider: Some(Arc::clone(&components.llm)),
skill_registry: components.skill_registry.clone(),
skill_catalog: components.skill_catalog.clone(),
chat_rate_limiter: RateLimiter::new(120, 60),
chat_rate_limiter: PerUserRateLimiter::new(120, 60),
oauth_rate_limiter: RateLimiter::new(10, 60),
webhook_rate_limiter: RateLimiter::new(10, 60),
registry_entries: Vec::new(),
@@ -254,7 +256,7 @@ impl GatewayWorkflowHarness {
skills_config: components.config.skills.clone(),
hooks: components.hooks,
cost_guard: components.cost_guard,
sse_tx: Some(gateway_state.sse.sender()),
sse_tx: None,
http_interceptor: None,
transcription: None,
document_extraction: None,
@@ -288,10 +290,11 @@ impl GatewayWorkflowHarness {
}
let auth_token = "gateway-test-token".to_string();
let auth = MultiAuthState::single(auth_token.clone(), user_id.clone());
let addr = start_server(
"127.0.0.1:0".parse().expect("valid localhost addr"),
Arc::clone(&gateway_state),
auth_token.clone(),
auth,
)
.await
.expect("failed to start gateway server");
tests/ws_gateway_integration.rs
+9 -4
View File
@@ -39,8 +39,9 @@ async fn start_test_server() -> (
let state = Arc::new(GatewayState {
msg_tx: tokio::sync::RwLock::new(Some(agent_tx)),
sse: SseManager::new(),
sse: Arc::new(SseManager::new()),
workspace: None,
workspace_pool: None,
session_manager: None,
log_broadcaster: None,
log_level_handle: None,
@@ -50,13 +51,13 @@ async fn start_test_server() -> (
job_manager: None,
prompt_queue: None,
scheduler: None,
user_id: "test-user".to_string(),
default_user_id: "test-user".to_string(),
shutdown_tx: tokio::sync::RwLock::new(None),
ws_tracker: Some(Arc::new(WsConnectionTracker::new())),
llm_provider: None,
skill_registry: None,
skill_catalog: None,
chat_rate_limiter: ironclaw::channels::web::server::RateLimiter::new(30, 60),
chat_rate_limiter: ironclaw::channels::web::server::PerUserRateLimiter::new(30, 60),
oauth_rate_limiter: ironclaw::channels::web::server::RateLimiter::new(10, 60),
webhook_rate_limiter: ironclaw::channels::web::server::RateLimiter::new(10, 60),
registry_entries: Vec::new(),
@@ -66,8 +67,12 @@ async fn start_test_server() -> (
active_config: ironclaw::channels::web::server::ActiveConfigSnapshot::default(),
});
let auth = ironclaw::channels::web::auth::MultiAuthState::single(
AUTH_TOKEN.to_string(),
"test-user".to_string(),
);
let addr: SocketAddr = "127.0.0.1:0".parse().unwrap();
let bound_addr = start_server(addr, state.clone(), AUTH_TOKEN.to_string())
let bound_addr = start_server(addr, state.clone(), auth)
.await
.expect("Failed to start test server");