diff --git a/src/agent/agent_loop.rs b/src/agent/agent_loop.rs index 15853f14..b945a812 100644 --- a/src/agent/agent_loop.rs +++ b/src/agent/agent_loop.rs @@ -738,6 +738,18 @@ impl Agent { } async fn handle_message(&self, message: &IncomingMessage) -> Result, Error> { + // Log at info level only for tracking without exposing PII (user_id can be a phone number) + tracing::info!(message_id = %message.id, "Processing message"); + + // Log sensitive details at debug level for troubleshooting + tracing::debug!( + message_id = %message.id, + user_id = %message.user_id, + channel = %message.channel, + thread_id = ?message.thread_id, + "Message details" + ); + // Set message tool context for this turn (current channel and target) // For Signal, use signal_target from metadata (group:ID or phone number), // otherwise fall back to user_id @@ -786,10 +798,19 @@ impl Agent { // Hydrate thread from DB if it's a historical thread not in memory if let Some(ref external_thread_id) = message.thread_id { + tracing::debug!( + message_id = %message.id, + thread_id = %external_thread_id, + "Hydrating thread from DB" + ); self.maybe_hydrate_thread(message, external_thread_id).await; } // Resolve session and thread + tracing::debug!( + message_id = %message.id, + "Resolving session and thread" + ); let (session, thread_id) = self .session_manager .resolve_thread( @@ -798,6 +819,11 @@ impl Agent { message.thread_id.as_deref(), ) .await; + tracing::info!( + message_id = %message.id, + thread_id = %thread_id, + "Resolved session and thread" + ); // Auth mode interception: if the thread is awaiting a token, route // the message directly to the credential store. Nothing touches diff --git a/src/agent/thread_ops.rs b/src/agent/thread_ops.rs index 758e98ed..c987b826 100644 --- a/src/agent/thread_ops.rs +++ b/src/agent/thread_ops.rs @@ -113,6 +113,13 @@ impl Agent { thread_id: Uuid, content: &str, ) -> Result { + tracing::debug!( + message_id = %message.id, + thread_id = %thread_id, + content_len = content.len(), + "Processing user input" + ); + // First check thread state without holding lock during I/O let thread_state = { let sess = session.lock().await; @@ -123,19 +130,41 @@ impl Agent { thread.state }; + tracing::debug!( + message_id = %message.id, + thread_id = %thread_id, + thread_state = ?thread_state, + "Checked thread state" + ); + // Check thread state match thread_state { ThreadState::Processing => { + tracing::warn!( + message_id = %message.id, + thread_id = %thread_id, + "Thread is processing, rejecting new input" + ); return Ok(SubmissionResult::error( "Turn in progress. Use /interrupt to cancel.", )); } ThreadState::AwaitingApproval => { + tracing::warn!( + message_id = %message.id, + thread_id = %thread_id, + "Thread awaiting approval, rejecting new input" + ); return Ok(SubmissionResult::error( "Waiting for approval. Use /interrupt to cancel.", )); } ThreadState::Completed => { + tracing::warn!( + message_id = %message.id, + thread_id = %thread_id, + "Thread completed, rejecting new input" + ); return Ok(SubmissionResult::error( "Thread completed. Use /thread new.", )); @@ -269,9 +298,20 @@ impl Agent { }; // Persist user message to DB immediately so it survives crashes + tracing::debug!( + message_id = %message.id, + thread_id = %thread_id, + "Persisting user message to DB" + ); self.persist_user_message(thread_id, &message.user_id, effective_content) .await; + tracing::debug!( + message_id = %message.id, + thread_id = %thread_id, + "User message persisted, starting agentic loop" + ); + // Send thinking status let _ = self .channels diff --git a/src/channels/web/handlers/chat.rs b/src/channels/web/handlers/chat.rs index e82c2583..b7f4425c 100644 --- a/src/channels/web/handlers/chat.rs +++ b/src/channels/web/handlers/chat.rs @@ -35,6 +35,7 @@ pub async fn chat_send_handler( } let msg_id = msg.id; + let thread_id = msg.thread_id.clone(); let tx_guard = state.msg_tx.read().await; let tx = tx_guard.as_ref().ok_or(( @@ -49,6 +50,13 @@ pub async fn chat_send_handler( ) })?; + tracing::debug!( + message_id = %msg_id, + thread_id = ?thread_id, + content_len = req.content.len(), + "Message queued to agent loop" + ); + Ok(( StatusCode::ACCEPTED, Json(SendMessageResponse { @@ -263,7 +271,6 @@ pub async fn chat_history_handler( ))?; let session = session_manager.get_or_create_session(&state.user_id).await; - let sess = session.lock().await; let limit = query.limit.unwrap_or(50); let before_cursor = query @@ -281,11 +288,12 @@ pub async fn chat_history_handler( }) .transpose()?; - // Find the thread + // Find the thread (lock only briefly to get active_thread if needed) let thread_id = if let Some(ref tid) = query.thread_id { Uuid::parse_str(tid) .map_err(|_| (StatusCode::BAD_REQUEST, "Invalid thread_id".to_string()))? } else { + let sess = session.lock().await; sess.active_thread .ok_or((StatusCode::NOT_FOUND, "No active thread".to_string()))? }; @@ -298,8 +306,11 @@ pub async fn chat_history_handler( .conversation_belongs_to_user(thread_id, &state.user_id) .await .unwrap_or(false); - if !owned && !sess.threads.contains_key(&thread_id) { - return Err((StatusCode::NOT_FOUND, "Thread not found".to_string())); + if !owned { + let sess = session.lock().await; + if !sess.threads.contains_key(&thread_id) { + return Err((StatusCode::NOT_FOUND, "Thread not found".to_string())); + } } } @@ -324,56 +335,60 @@ pub async fn chat_history_handler( } // Try in-memory first (freshest data for active threads) - if let Some(thread) = sess.threads.get(&thread_id) - && (!thread.turns.is_empty() || thread.pending_approval.is_some()) + // Lock only when checking in-memory state { - let turns: Vec = thread - .turns - .iter() - .map(|t| TurnInfo { - turn_number: t.turn_number, - user_input: t.user_input.clone(), - response: t.response.clone(), - state: format!("{:?}", t.state), - started_at: t.started_at.to_rfc3339(), - completed_at: t.completed_at.map(|dt| dt.to_rfc3339()), - tool_calls: t - .tool_calls - .iter() - .map(|tc| ToolCallInfo { - name: tc.name.clone(), - has_result: tc.result.is_some(), - has_error: tc.error.is_some(), - result_preview: tc.result.as_ref().map(|r| { - let s = match r { - serde_json::Value::String(s) => s.clone(), - other => other.to_string(), - }; - truncate_preview(&s, 500) - }), - error: tc.error.clone(), - }) - .collect(), - }) - .collect(); + let sess = session.lock().await; + if let Some(thread) = sess.threads.get(&thread_id) + && (!thread.turns.is_empty() || thread.pending_approval.is_some()) + { + let turns: Vec = thread + .turns + .iter() + .map(|t| TurnInfo { + turn_number: t.turn_number, + user_input: t.user_input.clone(), + response: t.response.clone(), + state: format!("{:?}", t.state), + started_at: t.started_at.to_rfc3339(), + completed_at: t.completed_at.map(|dt| dt.to_rfc3339()), + tool_calls: t + .tool_calls + .iter() + .map(|tc| ToolCallInfo { + name: tc.name.clone(), + has_result: tc.result.is_some(), + has_error: tc.error.is_some(), + result_preview: tc.result.as_ref().map(|r| { + let s = match r { + serde_json::Value::String(s) => s.clone(), + other => other.to_string(), + }; + truncate_preview(&s, 500) + }), + error: tc.error.clone(), + }) + .collect(), + }) + .collect(); - let pending_approval = thread - .pending_approval - .as_ref() - .map(|pa| PendingApprovalInfo { - request_id: pa.request_id.to_string(), - tool_name: pa.tool_name.clone(), - description: pa.description.clone(), - parameters: serde_json::to_string_pretty(&pa.parameters).unwrap_or_default(), - }); + let pending_approval = thread + .pending_approval + .as_ref() + .map(|pa| PendingApprovalInfo { + request_id: pa.request_id.to_string(), + tool_name: pa.tool_name.clone(), + description: pa.description.clone(), + parameters: serde_json::to_string_pretty(&pa.parameters).unwrap_or_default(), + }); - return Ok(Json(HistoryResponse { - thread_id, - turns, - has_more: false, - oldest_timestamp: None, - pending_approval, - })); + return Ok(Json(HistoryResponse { + thread_id, + turns, + has_more: false, + oldest_timestamp: None, + pending_approval, + })); + } } // Fall back to DB for historical threads not in memory (paginated) @@ -415,7 +430,6 @@ pub async fn chat_threads_handler( ))?; let session = session_manager.get_or_create_session(&state.user_id).await; - let sess = session.lock().await; // Try DB first for persistent thread list if let Some(ref store) = state.store { @@ -465,15 +479,22 @@ pub async fn chat_threads_handler( }); } + // Read active thread while holding minimal lock (just before return) + let active_thread = { + let sess = session.lock().await; + sess.active_thread + }; + return Ok(Json(ThreadListResponse { assistant_thread, threads, - active_thread: sess.active_thread, + active_thread, })); } } // Fallback: in-memory only (no assistant thread without DB) + let sess = session.lock().await; let mut sorted_threads: Vec<_> = sess.threads.values().collect(); sorted_threads.sort_by(|a, b| b.updated_at.cmp(&a.updated_at)); let threads: Vec = sorted_threads @@ -490,10 +511,13 @@ pub async fn chat_threads_handler( }) .collect(); + let active_thread = sess.active_thread; + drop(sess); // Explicit drop to release lock + Ok(Json(ThreadListResponse { assistant_thread: None, threads, - active_thread: sess.active_thread, + active_thread, })) } diff --git a/tests/ws_gateway_integration.rs b/tests/ws_gateway_integration.rs index da44f766..6f66e19e 100644 --- a/tests/ws_gateway_integration.rs +++ b/tests/ws_gateway_integration.rs @@ -340,3 +340,72 @@ async fn test_ws_multiple_events_in_sequence() { ws.close(None).await.unwrap(); } + +/// Regression test: verify session lock is not held during API handler operations. +/// +/// This test ensures that concurrent API requests (e.g., listing threads) don't +/// block the agent loop from processing messages. Previously, chat_threads_handler +/// and chat_history_handler held session locks during slow DB operations, which +/// would deadlock the agent loop waiting to resolve sessions for incoming messages. +/// +/// The test verifies that concurrent access to session state completes quickly +/// without deadlock. If locks are heavily contended, the test will timeout. +#[tokio::test] +async fn test_session_lock_not_held_during_api_operations() { + use ironclaw::agent::SessionManager; + + let (_addr, _state, _agent_rx) = start_test_server().await; + + // Create a session manager and attach it to state + let session_manager = Arc::new(SessionManager::new()); + + // Note: We can't directly modify state.session_manager in the test due to its type. + // Instead, we test the session manager directly in isolation to verify lock behavior. + + // Spawn concurrent operations simulating API handler + agent loop interaction + let mut handles = vec![]; + + // Simulate API handler threads accessing sessions + for user_id in 0..5 { + let sm = session_manager.clone(); + handles.push(tokio::spawn(async move { + for _ in 0..20 { + let session = sm.get_or_create_session(&format!("user-{}", user_id)).await; + // Lock and release quickly (simulating API reading session state) + { + let _sess = session.lock().await; + tokio::time::sleep(Duration::from_micros(100)).await; + } + } + })); + } + + // Simulate agent loop thread resolving threads + let sm = session_manager.clone(); + let agent_handle = tokio::spawn(async move { + for i in 0..20 { + let (_session, _thread_id) = sm + .resolve_thread(&format!("user-{}", i % 5), "gateway", None) + .await; + // Should not block waiting for API handler locks + tokio::time::sleep(Duration::from_micros(100)).await; + } + }); + handles.push(agent_handle); + + // Wait for all tasks to complete within reasonable time + // If session locks are held during slow operations, this will timeout + let timeout_duration = Duration::from_secs(5); + let wait_result = timeout(timeout_duration, async { + for handle in handles { + let _ = handle.await; + } + }) + .await; + + assert!( + wait_result.is_ok(), + "Concurrent session access deadlocked or timed out. \ + This suggests session locks are held too long during I/O operations." + ); +}