mirror of
https://github.com/outbackdingo/optimclaw.git
synced 2026-08-27 08:00:17 +00:00
Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
74cbe5c2ea |
+94
-174
@@ -954,7 +954,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
|
||||
@@ -972,41 +975,31 @@ impl Agent {
|
||||
return Ok(SubmissionResult::ok_with_message(""));
|
||||
}
|
||||
|
||||
thread.take_pending_approval()
|
||||
};
|
||||
|
||||
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 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;
|
||||
match sess.threads.get_mut(&thread_id) {
|
||||
Some(thread) => {
|
||||
thread.await_approval(pending);
|
||||
}
|
||||
let taken = match thread.take_pending_approval() {
|
||||
Some(p) => p,
|
||||
None => {
|
||||
tracing::warn!(
|
||||
tracing::debug!(
|
||||
%thread_id,
|
||||
"Thread disappeared while restoring pending approval after request ID mismatch"
|
||||
"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
|
||||
&& 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.",
|
||||
));
|
||||
}
|
||||
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
|
||||
@@ -1023,19 +1016,8 @@ impl Agent {
|
||||
// Reset thread state to processing
|
||||
{
|
||||
let mut sess = session.lock().await;
|
||||
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",
|
||||
));
|
||||
}
|
||||
if let Some(thread) = sess.threads.get_mut(&thread_id) {
|
||||
thread.state = ThreadState::Processing;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1119,21 +1101,13 @@ impl Agent {
|
||||
// Record sanitized result in thread
|
||||
{
|
||||
let mut sess = session.lock().await;
|
||||
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(result_content.clone());
|
||||
} else {
|
||||
turn.record_tool_result(serde_json::json!(result_content));
|
||||
}
|
||||
}
|
||||
}
|
||||
None => {
|
||||
tracing::error!(
|
||||
%thread_id,
|
||||
"Thread disappeared while recording tool result during approval"
|
||||
);
|
||||
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(result_content.clone());
|
||||
} else {
|
||||
turn.record_tool_result(serde_json::json!(result_content));
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1381,22 +1355,13 @@ impl Agent {
|
||||
// Record sanitized result in thread
|
||||
{
|
||||
let mut sess = session.lock().await;
|
||||
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(deferred_content.clone());
|
||||
} else {
|
||||
turn.record_tool_result(serde_json::json!(deferred_content));
|
||||
}
|
||||
}
|
||||
}
|
||||
None => {
|
||||
tracing::error!(
|
||||
%thread_id,
|
||||
tool_name = %tc.name,
|
||||
"Thread disappeared while recording deferred tool result during approval"
|
||||
);
|
||||
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(deferred_content.clone());
|
||||
} else {
|
||||
turn.record_tool_result(serde_json::json!(deferred_content));
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1449,19 +1414,8 @@ impl Agent {
|
||||
|
||||
{
|
||||
let mut sess = session.lock().await;
|
||||
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",
|
||||
));
|
||||
}
|
||||
if let Some(thread) = sess.threads.get_mut(&thread_id) {
|
||||
thread.await_approval(new_pending);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1593,28 +1547,17 @@ impl Agent {
|
||||
);
|
||||
{
|
||||
let mut sess = session.lock().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",
|
||||
));
|
||||
}
|
||||
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;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2156,70 +2099,6 @@ 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};
|
||||
@@ -2326,6 +2205,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,
|
||||
|
||||
Reference in New Issue
Block a user