mirror of
https://github.com/outbackdingo/optimclaw.git
synced 2026-08-26 15:40:18 +00:00
Compare commits
10
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
702af21aeb | ||
|
|
d9d48fdc72 | ||
|
|
71e200af74 | ||
|
|
ba4645b192 | ||
|
|
d59e2a1d4b | ||
|
|
bafe945c90 | ||
|
|
170566ab93 | ||
|
|
9170b28f7b | ||
|
|
cda0df3d7f | ||
|
|
75f7bea0fd |
+263
-99
@@ -968,6 +968,10 @@ impl Agent {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/// Process an approval or rejection of a pending tool execution.
|
/// 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(
|
pub(super) async fn process_approval(
|
||||||
&self,
|
&self,
|
||||||
message: &IncomingMessage,
|
message: &IncomingMessage,
|
||||||
@@ -977,7 +981,10 @@ impl Agent {
|
|||||||
approved: bool,
|
approved: bool,
|
||||||
always: bool,
|
always: bool,
|
||||||
) -> Result<SubmissionResult, Error> {
|
) -> 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 pending = {
|
||||||
let mut sess = session.lock().await;
|
let mut sess = session.lock().await;
|
||||||
let thread = sess
|
let thread = sess
|
||||||
@@ -995,33 +1002,31 @@ impl Agent {
|
|||||||
return Ok(SubmissionResult::ok_with_message(""));
|
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 {
|
// Verify request ID while still holding the lock — atomic with take
|
||||||
Some(p) => p,
|
if let Some(req_id) = request_id {
|
||||||
None => {
|
if req_id != taken.request_id {
|
||||||
tracing::debug!(
|
// Restore atomically under same lock
|
||||||
%thread_id,
|
thread.await_approval(taken);
|
||||||
"Ignoring stale approval: no pending approval found"
|
return Ok(SubmissionResult::error(
|
||||||
);
|
"Request ID mismatch. Use the correct request ID.",
|
||||||
return Ok(SubmissionResult::ok_with_message(""));
|
));
|
||||||
|
}
|
||||||
}
|
}
|
||||||
};
|
|
||||||
|
|
||||||
// Verify request ID if provided
|
taken
|
||||||
if let Some(req_id) = request_id
|
// Lock dropped here — pending approval validated
|
||||||
&& 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.",
|
|
||||||
));
|
|
||||||
}
|
|
||||||
|
|
||||||
if approved {
|
if approved {
|
||||||
// If always, add to auto-approved set
|
// If always, add to auto-approved set
|
||||||
@@ -1038,8 +1043,19 @@ impl Agent {
|
|||||||
// Reset thread state to processing
|
// Reset thread state to processing
|
||||||
{
|
{
|
||||||
let mut sess = session.lock().await;
|
let mut sess = session.lock().await;
|
||||||
if let Some(thread) = sess.threads.get_mut(&thread_id) {
|
match sess.threads.get_mut(&thread_id) {
|
||||||
thread.state = ThreadState::Processing;
|
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;
|
.await;
|
||||||
|
|
||||||
if let Ok(ref output) = tool_result
|
if let Ok(ref output) = tool_result {
|
||||||
&& !output.is_empty()
|
if !output.is_empty() {
|
||||||
{
|
let _ = self
|
||||||
let _ = self
|
.channels
|
||||||
.channels
|
.send_status(
|
||||||
.send_status(
|
&message.channel,
|
||||||
&message.channel,
|
StatusUpdate::ToolResult {
|
||||||
StatusUpdate::ToolResult {
|
name: pending.tool_name.clone(),
|
||||||
name: pending.tool_name.clone(),
|
preview: output.clone(),
|
||||||
preview: output.clone(),
|
},
|
||||||
},
|
&message.metadata,
|
||||||
&message.metadata,
|
)
|
||||||
)
|
.await;
|
||||||
.await;
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Build context including the tool result
|
// Build context including the tool result
|
||||||
@@ -1123,15 +1139,26 @@ impl Agent {
|
|||||||
// Record sanitized result in thread
|
// Record sanitized result in thread
|
||||||
{
|
{
|
||||||
let mut sess = session.lock().await;
|
let mut sess = session.lock().await;
|
||||||
if let Some(thread) = sess.threads.get_mut(&thread_id)
|
match sess.threads.get_mut(&thread_id) {
|
||||||
&& let Some(turn) = thread.last_turn_mut()
|
Some(thread) => {
|
||||||
{
|
if let Some(turn) = thread.last_turn_mut() {
|
||||||
if is_tool_error {
|
if is_tool_error {
|
||||||
turn.record_tool_error_for(&pending.tool_call_id, result_content.clone());
|
turn.record_tool_error_for(
|
||||||
} else {
|
&pending.tool_call_id,
|
||||||
turn.record_tool_result_for(
|
result_content.clone(),
|
||||||
&pending.tool_call_id,
|
);
|
||||||
serde_json::json!(result_content),
|
} 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;
|
let mut deferred_auth: Option<String> = None;
|
||||||
|
|
||||||
for (tc, deferred_result) in exec_results {
|
for (tc, deferred_result) in exec_results {
|
||||||
if let Ok(ref output) = deferred_result
|
if let Ok(ref output) = deferred_result {
|
||||||
&& !output.is_empty()
|
if !output.is_empty() {
|
||||||
{
|
let _ = self
|
||||||
let _ = self
|
.channels
|
||||||
.channels
|
.send_status(
|
||||||
.send_status(
|
&message.channel,
|
||||||
&message.channel,
|
StatusUpdate::ToolResult {
|
||||||
StatusUpdate::ToolResult {
|
name: tc.name.clone(),
|
||||||
name: tc.name.clone(),
|
preview: output.clone(),
|
||||||
preview: output.clone(),
|
},
|
||||||
},
|
&message.metadata,
|
||||||
&message.metadata,
|
)
|
||||||
)
|
.await;
|
||||||
.await;
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Sanitize first, then record the cleaned version in thread.
|
// Sanitize first, then record the cleaned version in thread.
|
||||||
@@ -1380,35 +1407,45 @@ impl Agent {
|
|||||||
// Record sanitized result in thread
|
// Record sanitized result in thread
|
||||||
{
|
{
|
||||||
let mut sess = session.lock().await;
|
let mut sess = session.lock().await;
|
||||||
if let Some(thread) = sess.threads.get_mut(&thread_id)
|
match sess.threads.get_mut(&thread_id) {
|
||||||
&& let Some(turn) = thread.last_turn_mut()
|
Some(thread) => {
|
||||||
{
|
if let Some(turn) = thread.last_turn_mut() {
|
||||||
if is_deferred_error {
|
if is_deferred_error {
|
||||||
turn.record_tool_error_for(&tc.id, deferred_content.clone());
|
turn.record_tool_error_for(&tc.id, deferred_content.clone());
|
||||||
} else {
|
} else {
|
||||||
turn.record_tool_result_for(
|
turn.record_tool_result_for(
|
||||||
&tc.id,
|
&tc.id,
|
||||||
serde_json::json!(deferred_content),
|
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
|
// Auth detection — defer return until all results are recorded
|
||||||
if deferred_auth.is_none()
|
if deferred_auth.is_none() {
|
||||||
&& let Some((ext_name, instructions)) =
|
if let Some((ext_name, instructions)) =
|
||||||
check_auth_required(&tc.name, &deferred_result)
|
check_auth_required(&tc.name, &deferred_result)
|
||||||
{
|
{
|
||||||
self.handle_auth_intercept(
|
self.handle_auth_intercept(
|
||||||
&session,
|
&session,
|
||||||
thread_id,
|
thread_id,
|
||||||
message,
|
message,
|
||||||
&deferred_result,
|
&deferred_result,
|
||||||
ext_name,
|
ext_name,
|
||||||
instructions.clone(),
|
instructions.clone(),
|
||||||
)
|
)
|
||||||
.await;
|
.await;
|
||||||
deferred_auth = Some(instructions);
|
deferred_auth = Some(instructions);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
context_messages.push(ChatMessage::tool_result(&tc.id, &tc.name, deferred_content));
|
context_messages.push(ChatMessage::tool_result(&tc.id, &tc.name, deferred_content));
|
||||||
@@ -1442,8 +1479,19 @@ impl Agent {
|
|||||||
|
|
||||||
{
|
{
|
||||||
let mut sess = session.lock().await;
|
let mut sess = session.lock().await;
|
||||||
if let Some(thread) = sess.threads.get_mut(&thread_id) {
|
match sess.threads.get_mut(&thread_id) {
|
||||||
thread.await_approval(new_pending);
|
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;
|
let mut sess = session.lock().await;
|
||||||
if let Some(thread) = sess.threads.get_mut(&thread_id) {
|
match sess.threads.get_mut(&thread_id) {
|
||||||
thread.clear_pending_approval();
|
Some(thread) => {
|
||||||
thread.complete_turn(&rejection);
|
thread.clear_pending_approval();
|
||||||
// User message already persisted at turn start; save rejection response
|
thread.complete_turn(&rejection);
|
||||||
self.persist_assistant_response(
|
// User message already persisted at turn start; save rejection response
|
||||||
thread_id,
|
self.persist_assistant_response(
|
||||||
&message.channel,
|
thread_id,
|
||||||
&message.user_id,
|
&message.channel,
|
||||||
&rejection,
|
&message.user_id,
|
||||||
)
|
&rejection,
|
||||||
.await;
|
)
|
||||||
|
.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]
|
#[test]
|
||||||
fn test_queue_cap_rejects_at_capacity() {
|
fn test_queue_cap_rejects_at_capacity() {
|
||||||
use crate::agent::session::{MAX_PENDING_MESSAGES, Thread, ThreadState};
|
use crate::agent::session::{MAX_PENDING_MESSAGES, Thread, ThreadState};
|
||||||
@@ -2251,6 +2374,47 @@ mod tests {
|
|||||||
assert!(t.pending_messages.is_empty());
|
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
|
// Helper function to extract the approval message without needing a full Agent instance
|
||||||
fn extract_approval_message(
|
fn extract_approval_message(
|
||||||
session: &crate::agent::session::Session,
|
session: &crate::agent::session::Session,
|
||||||
|
|||||||
Reference in New Issue
Block a user