ooodc/PicoBot
1
0
Fork 0
Code Pull Requests Activity

Compare commits

base: ooodc:cb58d9f3f03ea8dedd45dc8db90950d7043cb849
Branches Tags
ooodc:main
..
compare: ooodc:ea6fabe41d3e7bd02712453e3331edbb66647a62
Branches Tags
ooodc:main

No commits in common. "cb58d9f3f03ea8dedd45dc8db90950d7043cb849" and "ea6fabe41d3e7bd02712453e3331edbb66647a62" have entirely different histories.
cb58d9f3f0 ... ea6fabe41d

3 changed files with 8 additions and 369 deletions
Show Stats Expand all files Collapse all files

333
src/agent/agent_loop.rs
View File

@ -838,123 +838,6 @@ impl AgentLoop {
&self.tools
}
/// Sanitize message history by removing trailing assistant messages with
/// tool_calls that don't have corresponding tool result messages.
///
/// This can happen if the process was interrupted mid-execution: the
/// assistant message with tool_calls was persisted but the tool results
/// were not. Sending such incomplete sequences to the API causes errors
/// like "insufficient tool messages following tool_calls message".
///
/// Returns the number of messages removed.
fn sanitize_incomplete_tool_call_sequences(messages: &mut Vec<ChatMessage>) -> usize {
let mut removed = 0;
// Phase 1: Remove trailing assistant messages with tool_calls that lack
// corresponding tool result messages. We loop because removing one may
// expose another incomplete sequence.
loop {
let last_assistant_idx = match messages.iter().rposition(|m| {
m.role == "assistant"
&& m.tool_calls
.as_ref()
.map_or(false, |calls| !calls.is_empty())
}) {
Some(idx) => idx,
None => break,
};
// Collect all tool_call_ids from this assistant message
let tool_call_ids: Vec<&str> = messages[last_assistant_idx]
.tool_calls
.as_ref()
.unwrap()
.iter()
.map(|tc| tc.id.as_str())
.collect();
// Check if ALL tool_call_ids have corresponding tool messages
// appearing AFTER this assistant message
let all_have_results = tool_call_ids.iter().all(|&tc_id| {
messages[last_assistant_idx + 1..]
.iter()
.any(|m| m.role == "tool" && m.tool_call_id.as_deref() == Some(tc_id))
});
if all_have_results {
// Complete sequence found, stop trimming
break;
}
let tool_call_count = tool_call_ids.len();
let missing_count = tool_call_ids
.iter()
.filter(|&&tc_id| {
!messages[last_assistant_idx + 1..]
.iter()
.any(|m| m.role == "tool" && m.tool_call_id.as_deref() == Some(tc_id))
})
.count();
tracing::warn!(
tool_call_count,
missing_tool_results = missing_count,
message_id = %messages[last_assistant_idx].id,
"Removing assistant message with incomplete tool call sequence — \
tool results were never persisted (likely due to process interruption)"
);
messages.remove(last_assistant_idx);
removed += 1;
}
// Phase 2: Remove orphaned trailing tool messages that no longer have a
// corresponding assistant tool_calls message before them. These are left
// over from Phase 1 removals.
//
// We work backwards: a tool message at the end of the sequence is
// orphaned if none of the preceding messages is an assistant with a
// matching tool_call in its tool_calls.
while let Some(last_idx) = messages.last().map(|_| messages.len() - 1) {
let last = &messages[last_idx];
if last.role != "tool" {
break;
}
let tool_id = match &last.tool_call_id {
Some(id) => id.as_str(),
None => break, // tool message without tool_call_id — shouldn't happen, but safe
};
// Check if any preceding assistant message has this tool_call_id in
// its tool_calls
let has_parent = messages[..last_idx].iter().any(|m| {
m.role == "assistant"
&& m.tool_calls
.as_ref()
.map_or(false, |calls| {
calls.iter().any(|tc| tc.id == tool_id)
})
});
if has_parent {
break; // This tool message has a valid parent, stop
}
tracing::warn!(
tool_call_id = %tool_id,
message_id = %last.id,
"Removing orphaned tool result message — its parent assistant \
tool_calls message was removed or never persisted"
);
messages.remove(last_idx);
removed += 1;
}
removed
}
/// Process a message using the provided conversation history.
/// History management is handled externally by SessionManager.
///
@ -978,10 +861,6 @@ impl AgentLoop {
"Starting agent process"
);
// Sanitize: remove any trailing incomplete tool call sequences
// that may have been persisted before a process interruption.
Self::sanitize_incomplete_tool_call_sequences(&mut messages);
// Track tool calls for loop detection
let mut loop_detector = LoopDetector::new(LoopDetectorConfig::default());
let mut emitted_messages = Vec::new();
@ -1884,218 +1763,6 @@ mod tests {
assert_eq!(filtered[0].media_refs.len(), 0, "age=19 的消息图片应被过滤");
assert!(filtered[0].content.contains("超出 10 条消息范围"));
}
// ====================
// sanitize_incomplete_tool_call_sequences tests
// ====================
#[test]
fn test_sanitize_removes_trailing_incomplete_tool_call_sequence() {
let mut messages = vec![
ChatMessage::user("hello"),
ChatMessage::assistant_with_tool_calls(
"calling tool",
vec![ToolCall {
id: "call_1".to_string(),
name: "calculator".to_string(),
arguments: serde_json::json!({"expression": "1+1"}),
}],
),
// Tool result for call_1 is MISSING — incomplete sequence
];
let removed = AgentLoop::sanitize_incomplete_tool_call_sequences(&mut messages);
assert_eq!(removed, 1);
assert_eq!(messages.len(), 1);
assert_eq!(messages[0].role, "user");
}
#[test]
fn test_sanitize_preserves_complete_tool_call_sequence() {
let mut messages = vec![
ChatMessage::user("hello"),
ChatMessage::assistant_with_tool_calls(
"calling tool",
vec![ToolCall {
id: "call_1".to_string(),
name: "calculator".to_string(),
arguments: serde_json::json!({"expression": "1+1"}),
}],
),
ChatMessage::tool("call_1", "calculator", "2"),
];
let removed = AgentLoop::sanitize_incomplete_tool_call_sequences(&mut messages);
assert_eq!(removed, 0);
assert_eq!(messages.len(), 3);
}
#[test]
fn test_sanitize_removes_multiple_incomplete_sequences() {
let mut messages = vec![
ChatMessage::user("hello"),
ChatMessage::assistant_with_tool_calls(
"first tool call",
vec![ToolCall {
id: "call_1".to_string(),
name: "calculator".to_string(),
arguments: serde_json::json!({"expression": "1+1"}),
}],
),
// Missing tool result for call_1
ChatMessage::user("second question"),
ChatMessage::assistant_with_tool_calls(
"second tool call",
vec![ToolCall {
id: "call_2".to_string(),
name: "read".to_string(),
arguments: serde_json::json!({"path": "README.md"}),
}],
),
// Also missing tool result for call_2
];
let removed = AgentLoop::sanitize_incomplete_tool_call_sequences(&mut messages);
// Should remove both trailing assistant messages with incomplete tool calls
assert_eq!(removed, 2);
assert_eq!(messages.len(), 2);
assert_eq!(messages[0].role, "user");
assert_eq!(messages[0].content, "hello");
assert_eq!(messages[1].role, "user");
assert_eq!(messages[1].content, "second question");
}
#[test]
fn test_sanitize_removes_assistant_when_partial_tool_results() {
// Assistant makes 2 tool calls, but only 1 tool result exists
let mut messages = vec![
ChatMessage::user("hello"),
ChatMessage::assistant_with_tool_calls(
"calling two tools",
vec![
ToolCall {
id: "call_1".to_string(),
name: "calculator".to_string(),
arguments: serde_json::json!({"expression": "1+1"}),
},
ToolCall {
id: "call_2".to_string(),
name: "read".to_string(),
arguments: serde_json::json!({"path": "README.md"}),
},
],
),
ChatMessage::tool("call_1", "calculator", "2"),
// Missing tool result for call_2
];
let removed_count = AgentLoop::sanitize_incomplete_tool_call_sequences(&mut messages);
// Phase 1 removes the assistant message (call_2 has no result).
// Phase 2 removes the orphaned tool result for call_1 (its parent
// assistant was removed).
assert_eq!(removed_count, 2);
assert_eq!(messages.len(), 1);
assert_eq!(messages[0].role, "user");
}
#[test]
fn test_sanitize_preserves_messages_without_tool_calls() {
let mut messages = vec![
ChatMessage::user("hello"),
ChatMessage::assistant("hi there"),
ChatMessage::user("how are you"),
];
let removed = AgentLoop::sanitize_incomplete_tool_call_sequences(&mut messages);
assert_eq!(removed, 0);
assert_eq!(messages.len(), 3);
}
#[test]
fn test_sanitize_handles_empty_messages() {
let mut messages: Vec<ChatMessage> = vec![];
let removed = AgentLoop::sanitize_incomplete_tool_call_sequences(&mut messages);
assert_eq!(removed, 0);
}
#[test]
fn test_sanitize_removes_orphaned_tool_messages() {
// A lone tool message without a preceding assistant tool_calls
// is orphaned and should be removed.
let mut messages = vec![
ChatMessage::tool("call_1", "calculator", "2"),
];
let removed = AgentLoop::sanitize_incomplete_tool_call_sequences(&mut messages);
assert_eq!(removed, 1);
assert!(messages.is_empty());
}
#[test]
fn test_sanitize_preserves_complete_sequence_with_multiple_tool_calls() {
let mut messages = vec![
ChatMessage::user("do two things"),
ChatMessage::assistant_with_tool_calls(
"calling two tools",
vec![
ToolCall {
id: "call_1".to_string(),
name: "calculator".to_string(),
arguments: serde_json::json!({"expression": "1+1"}),
},
ToolCall {
id: "call_2".to_string(),
name: "read".to_string(),
arguments: serde_json::json!({"path": "README.md"}),
},
],
),
ChatMessage::tool("call_1", "calculator", "2"),
ChatMessage::tool("call_2", "read", "contents of README"),
];
let removed = AgentLoop::sanitize_incomplete_tool_call_sequences(&mut messages);
assert_eq!(removed, 0);
assert_eq!(messages.len(), 4);
}
#[test]
fn test_sanitize_only_trims_trailing_incomplete_sequence() {
// Complete sequence followed by an incomplete one — only the
// trailing incomplete one should be removed
let mut messages = vec![
ChatMessage::user("first question"),
ChatMessage::assistant_with_tool_calls(
"first tool call",
vec![ToolCall {
id: "call_1".to_string(),
name: "calculator".to_string(),
arguments: serde_json::json!({"expression": "1+1"}),
}],
),
ChatMessage::tool("call_1", "calculator", "2"),
ChatMessage::assistant("the answer is 2"),
ChatMessage::user("second question"),
ChatMessage::assistant_with_tool_calls(
"second tool call",
vec![ToolCall {
id: "call_2".to_string(),
name: "read".to_string(),
arguments: serde_json::json!({"path": "README.md"}),
}],
),
// Missing tool result for call_2 — only THIS sequence should be trimmed
];
let removed = AgentLoop::sanitize_incomplete_tool_call_sequences(&mut messages);
assert_eq!(removed, 1);
// First complete sequence preserved (5 messages), user message for second
// question preserved
assert_eq!(messages.len(), 5);
assert_eq!(messages[0].content, "first question");
assert_eq!(messages[3].content, "the answer is 2");
assert_eq!(messages[4].content, "second question");
}
}
#[derive(Debug)]

36
src/gateway/ws.rs
View File

@ -459,13 +459,12 @@ async fn handle_inbound(
}
// 加载子智能体任务消息
if let Some(task_session_id) = response.metadata.get("task_session_id") {
// 提前提取 task_id，用于给历史消息打标记
let task_id = response.metadata.get("task_id").cloned().unwrap_or_default();
if let Err(e) = send_task_messages(&store, task_session_id, sender, Some(task_id.clone())).await {
if let Err(e) = send_task_messages(&store, task_session_id, sender).await {
tracing::warn!(error = %e, task_session_id = %task_session_id, "Failed to send task messages");
}
// 发送 TaskMessagesLoaded 元数据
let task_id = response.metadata.get("task_id").cloned().unwrap_or_default();
let description = response.metadata.get("task_description").cloned().unwrap_or_default();
let subagent_type = response.metadata.get("task_subagent_type").cloned().unwrap_or_default();
let status = response.metadata.get("task_status").cloned().unwrap_or_default();
@ -497,7 +496,7 @@ async fn handle_inbound(
&load_chat_channel,
load_chat_id,
);
if let Err(e) = send_task_messages(&store, &session_id, sender, None).await {
if let Err(e) = send_task_messages(&store, &session_id, sender).await {
tracing::warn!(
error = %e,
channel = %load_chat_channel,
@ -592,19 +591,13 @@ async fn send_task_messages(
store: &Arc<crate::storage::SessionStore>,
session_id: &str,
sender: &mpsc::Sender<WsOutbound>,
subagent_task_id: Option<String>,
) -> Result<(), Box<dyn std::error::Error>> {
let messages = store.load_messages(session_id)?;
tracing::info!(session_id = %session_id, message_count = messages.len(), "Sending task messages");
for msg in messages {
let mut outbound = chat_message_to_ws_outbound(&msg);
if let Some(ref task_id) = subagent_task_id {
if let Some(ref mut ob) = outbound {
set_subagent_task_id(ob, task_id);
}
}
let outbound = chat_message_to_ws_outbound(&msg);
if let Some(outbound) = outbound {
let _ = sender.send(outbound).await;
}
@ -613,27 +606,6 @@ async fn send_task_messages(
Ok(())
}
/// 给 WsOutbound 消息注入 subagent_task_id（仅对有该字段的变体生效）
fn set_subagent_task_id(outbound: &mut WsOutbound, task_id: &str) {
match outbound {
WsOutbound::AssistantResponse {
subagent_task_id, ..
}
| WsOutbound::ToolCall {
subagent_task_id, ..
}
| WsOutbound::ToolResult {
subagent_task_id, ..
}
| WsOutbound::ToolPending {
subagent_task_id, ..
} => {
*subagent_task_id = Some(task_id.to_string());
}
_ => {} // 其他变体没有 subagent_task_id 字段
}
}
/// 将 ChatMessage 转换为 WsOutbound
fn chat_message_to_ws_outbound(msg: &crate::bus::ChatMessage) -> Option<WsOutbound> {
use crate::bus::message::ToolMessageState;

8
web/src/hooks/useChat.ts
View File

@ -251,11 +251,11 @@ export function useChat(): UseChatReturn {
return
}
// Only accept messages explicitly tagged with matching subagent_task_id.
// History messages are now tagged by the backend (send_task_messages),
// and live sub-agent messages are tagged by SubAgentEmitter.
// Route messages to sub-agent view:
// - Messages without subagent_task_id = loaded history, always accept
// - Messages with subagent_task_id = live emitter, only accept if matching
const msgSubagentTaskId = getSubagentTaskId(message)
if (msgSubagentTaskId && msgSubagentTaskId === currentSubAgentView.taskId) {
if (!msgSubagentTaskId || msgSubagentTaskId === currentSubAgentView.taskId) {
appendToSubAgentViewMessage(message)
}
return