Compare commits

...
Author SHA1 Message Date
Claude 702af21aeb style: fix cargo fmt line-length violation in process_approval
[skip-regression-check]

https://claude.ai/code/session_01Ey7iG1mvBHeAWq6iEccmjB
2026-03-25 17:14:58 +00:00
Zaki d9d48fdc72 Merge branch 'fix/approval-thread-safety' of https://github.com/nearai/ironclaw into fix/approval-thread-safety 2026-03-25 09:59:22 -07:00
ZakiandClaude Opus 4.6 71e200af74 merge: resolve conflicts with staging (identity-based tool recording)
Combines this PR's explicit match/error handling for missing threads
with staging's identity-based record_tool_result_for/record_tool_error_for
methods from the reasoning PR.

Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>
2026-03-25 09:59:18 -07:00
Claude ba4645b192 Merge remote-tracking branch 'origin/staging' into fix/approval-thread-safety 2026-03-24 18:23:27 +00:00
Zaki ManianandGitHub d59e2a1d4b Merge branch 'staging' into fix/approval-thread-safety 2026-03-24 10:11:38 -07:00
ZakiandClaude Opus 4.6 bafe945c90 fix: convert remaining let-chains to nested ifs for MSRV compat
Refactor four let-chain patterns (`if let ... && ...`) in
process_approval() to use nested `if`/`if let` blocks. Let-chains
require `#![feature(let_chains)]` which is not available on our MSRV.
Add `#[allow(clippy::collapsible_if)]` with a comment explaining why.

Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>
2026-03-24 09:14:29 -07:00
Claude 170566ab93 style: run cargo fmt
https://claude.ai/code/session_01Mdiz3XwyZcjqMkqicaynGs
2026-03-23 14:12:34 +00:00
ZakiandClaude Opus 4.6 9170b28f7b test(agent): strengthen regression test for missing thread error handling
Replace shallow assertion-only test with one that exercises the actual
match-based error detection pattern used in process_approval()'s
rejection and state-setting paths.

Addresses Gemini review feedback on #1579.

Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>
2026-03-23 06:12:29 -07:00
ZakiandClaude Opus 4.6 cda0df3d7f fix(agent): return errors when approval thread disappears (#1487)
Replace silent if-let-Some patterns with explicit match arms that log
errors and return error responses when threads are not found during
approval processing. Critical state mutations (complete turn, clear
approval, set Processing, await approval) return errors. Auxiliary
operations (record tool result) log errors but continue since the tool
already executed.

Closes #1487

Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>
2026-03-23 06:12:29 -07:00
ZakiandClaude Opus 4.6 75f7bea0fd fix(security): eliminate TOCTOU race in approval request_id check (#1486)
Hold session lock for the entire take-verify sequence so pending approval
cannot be lost if a concurrent operation modifies the thread between
take and restore.

Closes #1486

Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>
2026-03-23 06:10:49 -07:00
+263 -99
View File
@@ -968,6 +968,10 @@ impl Agent {
}
/// Process an approval or rejection of a pending tool execution.
// Nested `if` blocks are intentional: collapsing them would produce
// `if let … && …` (let-chains), which require `#![feature(let_chains)]`
// and are not available on our MSRV.
#[allow(clippy::collapsible_if)]
pub(super) async fn process_approval(
&self,
message: &IncomingMessage,
@@ -977,7 +981,10 @@ impl Agent {
approved: bool,
always: bool,
) -> Result<SubmissionResult, Error> {
// Get pending approval for this thread
// Get pending approval for this thread.
// The take-verify sequence is atomic under a single lock acquisition
// to prevent a TOCTOU race where a concurrent operation could modify
// or delete the thread between take and restore (#1486).
let pending = {
let mut sess = session.lock().await;
let thread = sess
@@ -995,33 +1002,31 @@ impl Agent {
return Ok(SubmissionResult::ok_with_message(""));
}
thread.take_pending_approval()
};
let taken = match thread.take_pending_approval() {
Some(p) => p,
None => {
tracing::debug!(
%thread_id,
"Ignoring stale approval: no pending approval found"
);
return Ok(SubmissionResult::ok_with_message(""));
}
};
let pending = match pending {
Some(p) => p,
None => {
tracing::debug!(
%thread_id,
"Ignoring stale approval: no pending approval found"
);
return Ok(SubmissionResult::ok_with_message(""));
// Verify request ID while still holding the lock — atomic with take
if let Some(req_id) = request_id {
if req_id != taken.request_id {
// Restore atomically under same lock
thread.await_approval(taken);
return Ok(SubmissionResult::error(
"Request ID mismatch. Use the correct request ID.",
));
}
}
};
// Verify request ID if provided
if let Some(req_id) = request_id
&& req_id != pending.request_id
{
// Put it back and return error
let mut sess = session.lock().await;
if let Some(thread) = sess.threads.get_mut(&thread_id) {
thread.await_approval(pending);
}
return Ok(SubmissionResult::error(
"Request ID mismatch. Use the correct request ID.",
));
}
taken
// Lock dropped here — pending approval validated
};
if approved {
// If always, add to auto-approved set
@@ -1038,8 +1043,19 @@ impl Agent {
// Reset thread state to processing
{
let mut sess = session.lock().await;
if let Some(thread) = sess.threads.get_mut(&thread_id) {
thread.state = ThreadState::Processing;
match sess.threads.get_mut(&thread_id) {
Some(thread) => {
thread.state = ThreadState::Processing;
}
None => {
tracing::error!(
%thread_id,
"Thread disappeared while setting state to Processing during approval"
);
return Ok(SubmissionResult::error(
"Internal error: thread no longer exists",
));
}
}
}
@@ -1090,20 +1106,20 @@ impl Agent {
)
.await;
if let Ok(ref output) = tool_result
&& !output.is_empty()
{
let _ = self
.channels
.send_status(
&message.channel,
StatusUpdate::ToolResult {
name: pending.tool_name.clone(),
preview: output.clone(),
},
&message.metadata,
)
.await;
if let Ok(ref output) = tool_result {
if !output.is_empty() {
let _ = self
.channels
.send_status(
&message.channel,
StatusUpdate::ToolResult {
name: pending.tool_name.clone(),
preview: output.clone(),
},
&message.metadata,
)
.await;
}
}
// Build context including the tool result
@@ -1123,15 +1139,26 @@ impl Agent {
// Record sanitized result in thread
{
let mut sess = session.lock().await;
if let Some(thread) = sess.threads.get_mut(&thread_id)
&& let Some(turn) = thread.last_turn_mut()
{
if is_tool_error {
turn.record_tool_error_for(&pending.tool_call_id, result_content.clone());
} else {
turn.record_tool_result_for(
&pending.tool_call_id,
serde_json::json!(result_content),
match sess.threads.get_mut(&thread_id) {
Some(thread) => {
if let Some(turn) = thread.last_turn_mut() {
if is_tool_error {
turn.record_tool_error_for(
&pending.tool_call_id,
result_content.clone(),
);
} else {
turn.record_tool_result_for(
&pending.tool_call_id,
serde_json::json!(result_content),
);
}
}
}
None => {
tracing::error!(
%thread_id,
"Thread disappeared while recording tool result during approval"
);
}
}
@@ -1351,20 +1378,20 @@ impl Agent {
let mut deferred_auth: Option<String> = None;
for (tc, deferred_result) in exec_results {
if let Ok(ref output) = deferred_result
&& !output.is_empty()
{
let _ = self
.channels
.send_status(
&message.channel,
StatusUpdate::ToolResult {
name: tc.name.clone(),
preview: output.clone(),
},
&message.metadata,
)
.await;
if let Ok(ref output) = deferred_result {
if !output.is_empty() {
let _ = self
.channels
.send_status(
&message.channel,
StatusUpdate::ToolResult {
name: tc.name.clone(),
preview: output.clone(),
},
&message.metadata,
)
.await;
}
}
// Sanitize first, then record the cleaned version in thread.
@@ -1380,35 +1407,45 @@ impl Agent {
// Record sanitized result in thread
{
let mut sess = session.lock().await;
if let Some(thread) = sess.threads.get_mut(&thread_id)
&& let Some(turn) = thread.last_turn_mut()
{
if is_deferred_error {
turn.record_tool_error_for(&tc.id, deferred_content.clone());
} else {
turn.record_tool_result_for(
&tc.id,
serde_json::json!(deferred_content),
match sess.threads.get_mut(&thread_id) {
Some(thread) => {
if let Some(turn) = thread.last_turn_mut() {
if is_deferred_error {
turn.record_tool_error_for(&tc.id, deferred_content.clone());
} else {
turn.record_tool_result_for(
&tc.id,
serde_json::json!(deferred_content),
);
}
}
}
None => {
tracing::error!(
%thread_id,
tool_name = %tc.name,
"Thread disappeared while recording deferred tool result during approval"
);
}
}
}
// Auth detection — defer return until all results are recorded
if deferred_auth.is_none()
&& let Some((ext_name, instructions)) =
if deferred_auth.is_none() {
if let Some((ext_name, instructions)) =
check_auth_required(&tc.name, &deferred_result)
{
self.handle_auth_intercept(
&session,
thread_id,
message,
&deferred_result,
ext_name,
instructions.clone(),
)
.await;
deferred_auth = Some(instructions);
{
self.handle_auth_intercept(
&session,
thread_id,
message,
&deferred_result,
ext_name,
instructions.clone(),
)
.await;
deferred_auth = Some(instructions);
}
}
context_messages.push(ChatMessage::tool_result(&tc.id, &tc.name, deferred_content));
@@ -1442,8 +1479,19 @@ impl Agent {
{
let mut sess = session.lock().await;
if let Some(thread) = sess.threads.get_mut(&thread_id) {
thread.await_approval(new_pending);
match sess.threads.get_mut(&thread_id) {
Some(thread) => {
thread.await_approval(new_pending);
}
None => {
tracing::error!(
%thread_id,
"Thread disappeared while setting up deferred tool approval"
);
return Ok(SubmissionResult::error(
"Internal error: thread no longer exists",
));
}
}
}
@@ -1576,17 +1624,28 @@ impl Agent {
);
{
let mut sess = session.lock().await;
if let Some(thread) = sess.threads.get_mut(&thread_id) {
thread.clear_pending_approval();
thread.complete_turn(&rejection);
// User message already persisted at turn start; save rejection response
self.persist_assistant_response(
thread_id,
&message.channel,
&message.user_id,
&rejection,
)
.await;
match sess.threads.get_mut(&thread_id) {
Some(thread) => {
thread.clear_pending_approval();
thread.complete_turn(&rejection);
// User message already persisted at turn start; save rejection response
self.persist_assistant_response(
thread_id,
&message.channel,
&message.user_id,
&rejection,
)
.await;
}
None => {
tracing::error!(
%thread_id,
"Thread disappeared during approval rejection"
);
return Ok(SubmissionResult::error(
"Internal error: thread no longer exists",
));
}
}
}
@@ -2145,6 +2204,70 @@ mod tests {
}
}
#[tokio::test]
async fn test_approval_on_missing_thread_should_error() {
// Regression for #1487: when a thread disappears from the session
// during approval processing, the code must return a visible error
// rather than silently succeeding.
//
// We can't call process_approval() directly (requires full Agent),
// so we simulate the exact code pattern used in the rejection and
// state-setting paths: lock session, match on get_mut, verify the
// None arm produces an error.
use crate::agent::session::{Session, Thread, ThreadState};
use std::sync::Arc;
use tokio::sync::Mutex;
use uuid::Uuid;
let thread_id = Uuid::new_v4();
let session_id = Uuid::new_v4();
let session = Arc::new(Mutex::new(Session::new("test-user")));
// Scenario 1: Thread never existed
{
let sess = session.lock().await;
let result = match sess.threads.get(&thread_id) {
Some(_) => Ok("processed"),
None => Err("Internal error: thread no longer exists"),
};
assert!(result.is_err());
assert_eq!(
result.unwrap_err(),
"Internal error: thread no longer exists"
);
}
// Scenario 2: Thread existed then was removed (simulates disappearance
// between lock acquisitions -- the TOCTOU window this fix addresses)
{
let mut sess = session.lock().await;
let mut thread = Thread::with_id(thread_id, session_id);
thread.start_turn("pending approval");
thread.state = ThreadState::AwaitingApproval;
sess.threads.insert(thread_id, thread);
}
{
let mut sess = session.lock().await;
// Simulate thread disappearing (e.g., pruned by another task)
sess.threads.remove(&thread_id);
// The rejection path must detect this and return an error
let result = match sess.threads.get_mut(&thread_id) {
Some(thread) => {
thread.clear_pending_approval();
thread.complete_turn("rejected");
Ok("rejection persisted")
}
None => Err("Internal error: thread no longer exists"),
};
assert!(result.is_err());
assert_eq!(
result.unwrap_err(),
"Internal error: thread no longer exists"
);
}
}
#[test]
fn test_queue_cap_rejects_at_capacity() {
use crate::agent::session::{MAX_PENDING_MESSAGES, Thread, ThreadState};
@@ -2251,6 +2374,47 @@ mod tests {
assert!(t.pending_messages.is_empty());
}
#[test]
fn test_approval_request_id_mismatch_restores_pending() {
// Regression test for #1486: after a request_id mismatch, the pending
// approval must still be intact (take + verify + restore is atomic).
use crate::agent::session::{PendingApproval, Thread, ThreadState};
use uuid::Uuid;
let session_id = Uuid::new_v4();
let thread_id = Uuid::new_v4();
let mut thread = Thread::with_id(thread_id, session_id);
let correct_request_id = Uuid::new_v4();
let pending = PendingApproval {
request_id: correct_request_id,
tool_name: "shell".to_string(),
parameters: serde_json::json!({}),
display_parameters: serde_json::json!({}),
description: "test".to_string(),
tool_call_id: "call_0".to_string(),
context_messages: vec![],
deferred_tool_calls: vec![],
user_timezone: None,
allow_always: true,
};
thread.await_approval(pending);
assert_eq!(thread.state, ThreadState::AwaitingApproval);
// Simulate: take, verify mismatch, restore -- all must be atomic
let taken = thread.take_pending_approval().unwrap();
assert_eq!(taken.request_id, correct_request_id);
// On mismatch, restore
thread.await_approval(taken);
// Must still be in AwaitingApproval with pending intact
assert_eq!(thread.state, ThreadState::AwaitingApproval);
assert!(thread.pending_approval.is_some());
assert_eq!(
thread.pending_approval.as_ref().unwrap().request_id,
correct_request_id
);
}
// Helper function to extract the approval message without needing a full Agent instance
fn extract_approval_message(
session: &crate::agent::session::Session,