public/galaxy
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Fix tool proposal handoff logging

Browse Source
This commit is contained in:
rkw6086
2026-08-12 09:27:30 -05:00
parent 1ad2ab4010
commit b806cd76f8
3 changed files with 197 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/src/ai/blocklist/action_model.rs
+50 -1
View File
@@ -341,7 +341,7 @@ fn action_result_type_name(result: &AIAgentActionResultType) -> &'static str {
#[cfg(not(target_family = "wasm"))]
fn action_result_status(result: &AIAgentActionResultType) -> &'static str {
if result.is_successful() {
if action_result_is_success_for_remote_log(result) {
"success"
} else if result.is_failed() || action_result_failure_summary(result).is_some() {
"error"
@@ -352,6 +352,55 @@ fn action_result_status(result: &AIAgentActionResultType) -> &'static str {
}
}
#[cfg(not(target_family = "wasm"))]
fn action_result_is_success_for_remote_log(result: &AIAgentActionResultType) -> bool {
if result.is_successful() {
return true;
}
match result {
AIAgentActionResultType::WriteToLongRunningShellCommand(
WriteToLongRunningShellCommandResult::Snapshot { .. },
) => true,
AIAgentActionResultType::WriteToLongRunningShellCommand(
WriteToLongRunningShellCommandResult::CommandFinished { exit_code, .. },
) => exit_code.was_successful(),
AIAgentActionResultType::WriteToLongRunningShellCommand(
WriteToLongRunningShellCommandResult::Cancelled
| WriteToLongRunningShellCommandResult::Error(_),
)
| AIAgentActionResultType::RequestCommandOutput(_)
| AIAgentActionResultType::RequestFileEdits(_)
| AIAgentActionResultType::ReadFiles(_)
| AIAgentActionResultType::UploadArtifact(_)
| AIAgentActionResultType::SearchCodebase(_)
| AIAgentActionResultType::Grep(_)
| AIAgentActionResultType::FileGlob(_)
| AIAgentActionResultType::FileGlobV2(_)
| AIAgentActionResultType::ReadMCPResource(_)
| AIAgentActionResultType::CallMCPTool(_)
| AIAgentActionResultType::ReadSkill(_)
| AIAgentActionResultType::SuggestNewConversation(_)
| AIAgentActionResultType::SuggestPrompt(_)
| AIAgentActionResultType::OpenCodeReview
| AIAgentActionResultType::InsertReviewComments(_)
| AIAgentActionResultType::InitProject
| AIAgentActionResultType::ReadDocuments(_)
| AIAgentActionResultType::EditDocuments(_)
| AIAgentActionResultType::CreateDocuments(_)
| AIAgentActionResultType::ReadShellCommandOutput(_)
| AIAgentActionResultType::UseComputer(_)
| AIAgentActionResultType::RequestComputerUse(_)
| AIAgentActionResultType::FetchConversation(_)
| AIAgentActionResultType::StartAgent(_)
| AIAgentActionResultType::SendMessageToAgent(_)
| AIAgentActionResultType::TransferShellCommandControlToUser(_)
| AIAgentActionResultType::AskUserQuestion(_)
| AIAgentActionResultType::RunAgents(_)
| AIAgentActionResultType::WaitForEvents(_) => false,
}
}
#[cfg(not(target_family = "wasm"))]
fn action_result_log_level(result: &AIAgentActionResultType) -> RemoteLogLevel {
if result.is_failed() || action_result_failure_summary(result).is_some() {
app/src/ai/blocklist/controller.rs
+125 -1
View File
@@ -54,6 +54,8 @@ use crate::ai::agent::{
PassiveSuggestionTriggerType, RenderableAIError, RequestCommandOutputResult, RequestCost,
RequestMetadata, RunningCommand, StaticQueryType, TransientNetworkErrorKind, UserQueryMode,
};
#[cfg(not(target_family = "wasm"))]
use crate::ai::agent::{AIAgentAction, AIAgentActionTypeDiscriminants};
use crate::ai::agent_events::AgentMessageEventMetadata;
#[cfg(not(target_family = "wasm"))]
use crate::ai::agent_sdk::ClaudeHarness;
@@ -63,6 +65,8 @@ use crate::ai::document::ai_document_model::{
};
use crate::ai::llms::{LLMId, LLMPreferences};
use crate::ai::provider::types::ContentPart;
#[cfg(not(target_family = "wasm"))]
use crate::ai::remote_logging::{self, RemoteLogLevel, RemoteLogRecord};
use crate::ai::AIRequestUsageModel;
use crate::cloud_object::model::persistence::CloudModel;
use crate::features::FeatureFlag;
@@ -162,6 +166,29 @@ impl SessionContext {
}
}
#[cfg(not(target_family = "wasm"))]
fn remote_action_tool_name(action: &AIAgentAction) -> String {
action
.tool_name
.clone()
.unwrap_or_else(|| format!("{:?}", AIAgentActionTypeDiscriminants::from(&action.action)))
}
#[cfg(not(target_family = "wasm"))]
fn remote_action_summaries(actions: &[AIAgentAction]) -> Vec<serde_json::Value> {
actions
.iter()
.map(|action| {
serde_json::json!({
"action_id": action.id.to_string(),
"task_id": action.task_id.to_string(),
"tool_name": remote_action_tool_name(action),
"requires_result": action.requires_result,
})
})
.collect()
}
pub enum BlocklistAIControllerEvent {
/// Emitted when a request is sent to the AI agent API.
SentRequest {
@@ -3469,6 +3496,13 @@ impl BlocklistAIController {
let history_model = BlocklistAIHistoryModel::handle(ctx);
match event {
Ok(api::StreamEvent::ToolProposed(action)) => {
#[cfg(not(target_family = "wasm"))]
let action_log_context = (
action.id.to_string(),
action.task_id.to_string(),
remote_action_tool_name(&action),
action.requires_result,
);
let apply_result = history_model.update(ctx, |history_model, ctx| {
history_model.apply_domain_tool_proposal(
&stream_id,
@@ -3482,6 +3516,28 @@ impl BlocklistAIController {
log::error!(
"Failed to apply Rig tool proposal to conversation: {error:?}"
);
#[cfg(not(target_family = "wasm"))]
{
let (action_id, task_id, tool_name, requires_result) =
action_log_context;
remote_logging::log_model_event(
ctx,
RemoteLogRecord {
level: RemoteLogLevel::Error,
message: "Tool proposal apply failed".to_string(),
context: serde_json::json!({
"event": "tool_proposal_apply_failed",
"stream_id": stream_id.as_str(),
"conversation_id": conversation_id.to_string(),
"action_id": action_id,
"task_id": task_id,
"tool_name": tool_name,
"requires_result": requires_result,
"error": remote_logging::sanitize_error(format!("{error:?}")),
}),
},
);
}
}
}
Ok(api::StreamEvent::Response(event)) => {
@@ -3823,7 +3879,10 @@ impl BlocklistAIController {
);
});
}
ResponseStreamEvent::AfterStreamFinished { cancellation } => {
ResponseStreamEvent::AfterStreamFinished {
cancellation,
proposed_actions,
} => {
// Cancellations provide conversation_id (survives truncation); otherwise use dynamic lookup.
let conversation_id = match &cancellation {
Some(stream_cancellation) => stream_cancellation.conversation_id,
@@ -3946,6 +4005,71 @@ impl BlocklistAIController {
}
}
let history_action_count = actions_to_queue.len();
let proposed_action_count = proposed_actions.len();
let mut queued_action_ids = actions_to_queue
.iter()
.map(|action| action.id.clone())
.collect::<HashSet<_>>();
let mut queued_from_stream_snapshot_count = 0;
for action in proposed_actions {
if queued_action_ids.insert(action.id.clone()) {
queued_from_stream_snapshot_count += 1;
actions_to_queue.push(action.clone());
}
}
#[cfg(not(target_family = "wasm"))]
{
let will_queue_actions = cancellation.is_none() && !is_any_exchange_unfinished;
let used_stream_snapshot_fallback =
will_queue_actions && queued_from_stream_snapshot_count > 0;
let decision = if cancellation.is_some() {
"cancelled"
} else if is_any_exchange_unfinished {
"unfinished_exchange"
} else if actions_to_queue.is_empty() {
"no_actions"
} else if used_stream_snapshot_fallback {
"queue_actions_with_stream_snapshot_fallback"
} else {
"queue_actions"
};
let level = if used_stream_snapshot_fallback {
RemoteLogLevel::Warn
} else {
RemoteLogLevel::Info
};
remote_logging::log_model_event(
ctx,
RemoteLogRecord {
level,
message: "Tool queue decision".to_string(),
context: serde_json::json!({
"event": "tool_queue_decision",
"stream_id": stream_id.as_str(),
"conversation_id": conversation_id.to_string(),
"decision": decision,
"history_action_count": history_action_count,
"proposed_action_count": proposed_action_count,
"candidate_action_count": actions_to_queue.len(),
"will_queue_action_count": if will_queue_actions {
actions_to_queue.len()
} else {
0
},
"queued_from_stream_snapshot_count": queued_from_stream_snapshot_count,
"was_passive_request": was_passive_request,
"is_any_exchange_unfinished": is_any_exchange_unfinished,
"cancellation_reason": cancellation
.as_ref()
.map(|stream_cancellation| format!("{:?}", stream_cancellation.reason)),
"queued_tools": remote_action_summaries(&actions_to_queue),
"proposed_tools": remote_action_summaries(&proposed_actions),
}),
},
);
}
if let Some(stream_cancellation) = &cancellation {
// If this is a shared session, send a synthetic StreamFinished event to notify viewers
// of any user-initiated cancellation. We skip internal cancellations that preserve
app/src/ai/blocklist/controller/response_stream.rs
+22 -2
View File
@@ -31,7 +31,7 @@ use crate::ai::agent::api::{self, ConvertToAPITypeError};
use crate::ai::agent::conversation::AIConversationId;
#[cfg(not(target_family = "wasm"))]
use crate::ai::agent::AIAgentInput;
use crate::ai::agent::{AIIdentifiers, CancellationReason};
use crate::ai::agent::{AIAgentAction, AIIdentifiers, CancellationReason};
use crate::ai::bedrock::client::BedrockClientConfig;
#[cfg(not(target_family = "wasm"))]
use crate::ai::blocklist::BlocklistAIPermissions;
@@ -96,6 +96,10 @@ fn recovery_action(
pub struct ResponseStreamId(String);
impl ResponseStreamId {
pub fn as_str(&self) -> &str {
&self.0
}
pub fn for_shared_session(init_event: &response_event::StreamInit) -> Self {
// Make the stream ID unique per viewing by appending a local UUID
// This prevents collisions when replaying the same conversation multiple times
@@ -142,6 +146,12 @@ pub struct ResponseStream {
/// Track whether we've received any client actions
/// If true, we cannot retry on subsequent errors since actions may have been executed
has_received_client_actions: bool,
/// Domain tool proposals observed directly from the response stream for the current request.
///
/// The controller normally queues actions by reading them back from history after the stream
/// finishes. Keeping this snapshot prevents a final tool proposal from being lost if stream
/// completion is handled before that proposal has been applied to history.
proposed_actions: Vec<AIAgentAction>,
/// AI identifiers for telemetry emission
ai_identifiers: AIIdentifiers,
#[cfg(not(target_family = "wasm"))]
@@ -213,6 +223,7 @@ impl ResponseStream {
cancellation_tx: Some(cancellation_tx),
original_error: None,
has_received_client_actions: false,
proposed_actions: Vec::new(),
ai_identifiers: AIIdentifiers::default(),
#[cfg(not(target_family = "wasm"))]
remote_log_backend: "provider".to_string(),
@@ -807,6 +818,7 @@ impl ResponseStream {
coding_model_fallback_attempted: false,
original_error: None,
has_received_client_actions: false,
proposed_actions: Vec::new(),
ai_identifiers,
#[cfg(not(target_family = "wasm"))]
remote_log_backend,
@@ -925,6 +937,7 @@ impl ResponseStream {
self.retry_count += 1;
// Reset per-attempt state for the new attempt.
self.has_received_client_actions = false;
self.proposed_actions.clear();
self.stream_finished_received = false;
self.error_event_emitted = false;
self.deferred_retry_pending = false;
@@ -1016,6 +1029,7 @@ impl ResponseStream {
reason,
conversation_id,
}),
proposed_actions: self.proposed_actions.clone(),
});
}
@@ -1083,6 +1097,7 @@ impl ResponseStream {
match &event {
Ok(api::StreamEvent::ToolProposed(action)) => {
self.has_received_client_actions = true;
self.proposed_actions.push(action.clone());
log::debug!(
"Rig proposed domain tool action {} for task {}",
action.id,
@@ -1414,7 +1429,10 @@ impl ResponseStream {
}
}
ctx.emit(ResponseStreamEvent::AfterStreamFinished { cancellation: None });
ctx.emit(ResponseStreamEvent::AfterStreamFinished {
cancellation: None,
proposed_actions: self.proposed_actions.clone(),
});
self.cancellation_tx = None;
}
@@ -1665,6 +1683,8 @@ pub enum ResponseStreamEvent {
AfterStreamFinished {
/// Some for cancellation (with context), None for natural completion (uses dynamic lookup).
cancellation: Option<StreamCancellation>,
/// Domain tool proposals observed directly from the stream before it finished.
proposed_actions: Vec<AIAgentAction>,
},
}