ooodc/PicoBot
1
0
Fork 0
Code Pull Requests Activity

Compare commits

base: ooodc:1d4ebb27a7501e682b72e8d733d44eb3c08b6816
Branches Tags
ooodc:main
..
compare: ooodc:15dfc488370e52e0eab7e91e15935b90d6f100ac
Branches Tags
ooodc:main

No commits in common. "1d4ebb27a7501e682b72e8d733d44eb3c08b6816" and "15dfc488370e52e0eab7e91e15935b90d6f100ac" have entirely different histories.
1d4ebb27a7 ... 15dfc48837

21 changed files with 124 additions and 618 deletions
Show Stats Expand all files Collapse all files

384
src/agent/agent_loop.rs
View File

@ -646,8 +646,6 @@ pub struct AgentLoop {
observer: Option<Arc<dyn Observer>>,
emitted_message_handler: Option<Arc<dyn EmittedMessageHandler>>,
max_iterations: usize,
/// 取消信号接收端：Agent 在每次迭代开始时检查是否被取消
cancel_token: Option<tokio::sync::watch::Receiver<()>>,
}
#[derive(Debug, Clone)]
@ -744,7 +742,6 @@ impl AgentLoop {
tool_context: ToolContext::default(),
observer: None,
emitted_message_handler: None,
cancel_token: None,
max_iterations,
})
}
@ -767,7 +764,6 @@ impl AgentLoop {
tool_context: ToolContext::default(),
observer: None,
emitted_message_handler: None,
cancel_token: None,
max_iterations,
})
}
@ -791,7 +787,6 @@ impl AgentLoop {
tool_context: ToolContext::default(),
observer: None,
emitted_message_handler: None,
cancel_token: None,
max_iterations,
})
}
@ -819,7 +814,6 @@ impl AgentLoop {
tool_context: ToolContext::default(),
observer: None,
emitted_message_handler: None,
cancel_token: None,
max_iterations,
})
}
@ -840,129 +834,122 @@ impl AgentLoop {
self
}
/// 设置取消信号接收端。
///
/// Agent 在每次迭代开始时检查 `cancel_token.has_changed()`，
/// 如果已收到取消信号则提前返回。
pub fn with_cancel_token(mut self, token: tokio::sync::watch::Receiver<()>) -> Self {
self.cancel_token = Some(token);
self
}
pub fn tools(&self) -> &Arc<ToolRegistry> {
&self.tools
}
/// Sanitize message history by removing assistant messages with tool_calls
/// that don't have corresponding tool result messages, at ANY position in
/// the history (not just trailing).
/// Sanitize message history by removing trailing assistant messages with
/// tool_calls that don't have corresponding tool result messages.
///
/// Incomplete sequences can appear in the middle of history when:
/// 1. The process was interrupted mid-execution (before commit cb58d9f),
/// then a new user message was appended, burying the orphan.
/// 2. History compaction preserves orphaned tool_calls from a pre-fix era
/// or from a race condition between persistence and snapshot.
///
/// Sending such incomplete sequences to the API causes errors like
/// "insufficient tool messages following tool_calls message".
/// 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 {
use std::collections::HashSet;
let mut removed = 0;
// Phase 1: Single reverse pass to find ALL assistant messages with
// incomplete tool_calls, regardless of position.
//
// Scanning right-to-left means we encounter tool results before their
// parent assistants, so we naturally know which tool_call_ids have
// corresponding results.
let mut resolved_ids: HashSet<String> = HashSet::new();
let mut with_parent: HashSet<String> = HashSet::new();
let mut remove_indices: Vec<usize> = Vec::new();
// 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,
};
// Phase 1: Reverse pass — collect tool result IDs first, then
// validate each assistant's tool_calls against already-seen results.
//
// Because we scan right-to-left, any tool result we've already seen
// appears AFTER the current message in forward order. This correctly
// identifies which assistant tool_calls have corresponding results.
for i in (0..messages.len()).rev() {
let msg = &messages[i];
if msg.role == "tool" {
if let Some(ref tc_id) = msg.tool_call_id {
resolved_ids.insert(tc_id.clone());
}
}
if msg.role == "assistant"
&& msg.tool_calls.as_ref().map_or(false, |calls| !calls.is_empty())
{
let tool_calls = msg.tool_calls.as_ref().unwrap();
let all_have_results = tool_calls
// 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()
.all(|tc| resolved_ids.contains(&tc.id));
.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 {
for tc in tool_calls.iter() {
with_parent.insert(tc.id.clone());
// Complete sequence found, stop trimming
break;
}
} else {
let missing_count = tool_calls
let tool_call_count = tool_call_ids.len();
let missing_count = tool_call_ids
.iter()
.filter(|tc| !resolved_ids.contains(&tc.id))
.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 = tool_calls.len(),
tool_call_count,
missing_tool_results = missing_count,
message_id = %msg.id,
message_index = i,
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 \
or history compaction preserving an orphan)"
tool results were never persisted (likely due to process interruption)"
);
remove_indices.push(i);
}
}
}
// Remove in descending index order to avoid shifting
for &idx in &remove_indices {
messages.remove(idx);
messages.remove(last_assistant_idx);
removed += 1;
}
// Phase 2: Forward pass to remove ALL orphaned tool messages (not just
// trailing ones). A tool message is orphaned if its tool_call_id has no
// matching parent assistant remaining in the history.
if !with_parent.is_empty() || !resolved_ids.is_empty() {
let mut i = 0;
while i < messages.len() {
let msg = &messages[i];
if msg.role == "tool" {
let is_orphaned = match &msg.tool_call_id {
Some(tc_id) => !with_parent.contains(tc_id),
None => true,
// 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
};
if is_orphaned {
// 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 = ?msg.tool_call_id,
message_id = %msg.id,
message_index = i,
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(i);
messages.remove(last_idx);
removed += 1;
continue;
}
}
i += 1;
}
}
removed
@ -1003,25 +990,6 @@ impl AgentLoop {
#[cfg(debug_assertions)]
tracing::debug!(iteration, "Agent iteration started");
// 检查取消信号
if let Some(ref token) = self.cancel_token {
if token.has_changed().unwrap_or(false) {
tracing::info!(iteration, "Agent execution cancelled by user");
let cancel_message = format!(
"\n\n[用户已取消执行。已迭代 {} 次，取消前共生成了 {} 条消息。]",
iteration,
emitted_messages.len()
);
let assistant_message = ChatMessage::assistant(cancel_message);
emitted_messages.push(assistant_message.clone());
self.emit_live_tool_call_message(assistant_message.clone()).await;
return Ok(AgentProcessResult {
final_response: assistant_message,
emitted_messages,
});
}
}
// Build request
let tool_defs = self.tools.get_definitions();
let tools = if tool_defs.is_empty() {
@ -2128,190 +2096,6 @@ mod tests {
assert_eq!(messages[3].content, "the answer is 2");
assert_eq!(messages[4].content, "second question");
}
#[test]
fn test_sanitize_removes_mid_history_orphaned_tool_calls() {
// Bug scenario: orphaned tool_calls in the MIDDLE of history,
// followed by a complete sequence. The old trailing-only sanitizer
// would stop at the complete sequence and never remove the orphan.
let mut messages = vec![
ChatMessage::user("first question"),
ChatMessage::assistant_with_tool_calls(
"orphaned 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 — ORPHAN in the middle
ChatMessage::user("second question"),
ChatMessage::assistant_with_tool_calls(
"valid tool call",
vec![ToolCall {
id: "call_2".to_string(),
name: "read".to_string(),
arguments: serde_json::json!({"path": "README.md"}),
}],
),
ChatMessage::tool("call_2", "read", "file contents"),
];
let removed = AgentLoop::sanitize_incomplete_tool_call_sequences(&mut messages);
// call_1 assistant removed (1), rest preserved (4)
assert_eq!(removed, 1);
assert_eq!(messages.len(), 4);
assert_eq!(messages[0].role, "user");
assert_eq!(messages[0].content, "first question");
assert_eq!(messages[1].role, "user");
assert_eq!(messages[1].content, "second question");
// The complete call_2 sequence is preserved
assert_eq!(messages[2].role, "assistant");
assert_eq!(messages[3].role, "tool");
}
#[test]
fn test_sanitize_removes_multiple_mid_history_orphans() {
// Multiple orphaned tool_calls scattered throughout history
let mut messages = vec![
ChatMessage::user("first"),
ChatMessage::assistant_with_tool_calls(
"orphan 1",
vec![ToolCall {
id: "orphan_1".to_string(),
name: "tool_a".to_string(),
arguments: serde_json::json!({}),
}],
),
// Missing result for orphan_1
ChatMessage::user("second"),
ChatMessage::assistant_with_tool_calls(
"orphan 2",
vec![ToolCall {
id: "orphan_2".to_string(),
name: "tool_b".to_string(),
arguments: serde_json::json!({}),
}],
),
// Missing result for orphan_2
ChatMessage::user("third"),
];
let removed = AgentLoop::sanitize_incomplete_tool_call_sequences(&mut messages);
assert_eq!(removed, 2);
assert_eq!(messages.len(), 3);
assert_eq!(messages[0].content, "first");
assert_eq!(messages[1].content, "second");
assert_eq!(messages[2].content, "third");
}
#[test]
fn test_sanitize_removes_orphaned_tool_results_for_removed_mid_assistant() {
// When removing a mid-history assistant with partial tool results,
// both the assistant AND its orphaned tool results must be removed.
// Assistant has 2 tool calls, only 1 has a result → assistant is
// incomplete → both assistant and its lone tool result removed.
let mut messages = vec![
ChatMessage::user("first question"),
ChatMessage::assistant_with_tool_calls(
"two tool calls, only one has result",
vec![
ToolCall {
id: "call_has_result".to_string(),
name: "tool_a".to_string(),
arguments: serde_json::json!({}),
},
ToolCall {
id: "call_no_result".to_string(),
name: "tool_b".to_string(),
arguments: serde_json::json!({}),
},
],
),
ChatMessage::tool("call_has_result", "tool_a", "some result"),
// Missing tool result for call_no_result → incomplete sequence
ChatMessage::user("second question"),
ChatMessage::assistant_with_tool_calls(
"valid tool call",
vec![ToolCall {
id: "call_valid".to_string(),
name: "good_tool".to_string(),
arguments: serde_json::json!({}),
}],
),
ChatMessage::tool("call_valid", "good_tool", "valid result"),
];
let removed = AgentLoop::sanitize_incomplete_tool_call_sequences(&mut messages);
// call_has_result/call_no_result assistant (1) + its orphaned tool result (1) = 2 removed
assert_eq!(removed, 2);
assert_eq!(messages.len(), 4);
assert_eq!(messages[0].content, "first question");
assert_eq!(messages[1].content, "second question");
assert_eq!(messages[2].role, "assistant");
assert_eq!(messages[3].role, "tool");
// Verify the remaining tool belongs to call_valid
assert_eq!(messages[3].tool_call_id.as_deref(), Some("call_valid"));
}
#[test]
fn test_sanitize_handles_complex_interleaved_history() {
// Complete → Orphaned → Complete: a realistic scenario after
// history compaction
let mut messages = vec![
ChatMessage::user("task 1"),
ChatMessage::assistant_with_tool_calls(
"doing task 1",
vec![ToolCall {
id: "t1_call".to_string(),
name: "read".to_string(),
arguments: serde_json::json!({"path": "a.txt"}),
}],
),
ChatMessage::tool("t1_call", "read", "content A"),
ChatMessage::assistant("task 1 is done"),
// End of task 1 — complete sequence
ChatMessage::user("task 2"),
ChatMessage::assistant_with_tool_calls(
"doing task 2 — this got interrupted",
vec![
ToolCall {
id: "t2_call_1".to_string(),
name: "write".to_string(),
arguments: serde_json::json!({"path": "b.txt"}),
},
ToolCall {
id: "t2_call_2".to_string(),
name: "calculator".to_string(),
arguments: serde_json::json!({"expression": "2+2"}),
},
],
),
// Missing BOTH tool results — process was killed here
// End of task 2 — orphaned sequence in the middle
ChatMessage::user("task 3"),
ChatMessage::assistant_with_tool_calls(
"doing task 3",
vec![ToolCall {
id: "t3_call".to_string(),
name: "search".to_string(),
arguments: serde_json::json!({"query": "hello"}),
}],
),
ChatMessage::tool("t3_call", "search", "found results"),
ChatMessage::assistant("task 3 is done"),
];
let removed = AgentLoop::sanitize_incomplete_tool_call_sequences(&mut messages);
// Removed: assistant with t2_call_1/t2_call_2 (1 message)
assert_eq!(removed, 1);
// Original 10 messages - 1 = 9
assert_eq!(messages.len(), 9);
assert_eq!(messages[4].content, "task 2");
assert_eq!(messages[5].content, "task 3");
}
}
#[derive(Debug)]

20
src/channels/feishu.rs
View File

@ -1931,24 +1931,15 @@ impl FeishuChannel {
);
}
/// Restore all code block placeholders in the given text.
fn restore_code_blocks(text: &str, code_blocks: &[String]) -> String {
let mut result = text.to_string();
for (i, cb) in code_blocks.iter().enumerate() {
result = result.replace(&format!("\x00CODE{}\x00", i), cb);
}
result
}
let mut elements: Vec<serde_json::Value> = Vec::new();
let mut last_end = 0;
for m in patterns.heading_re.find_iter(&protected) {
let before = protected[last_end..m.start()].trim();
let before = &protected[last_end..m.start()].trim();
if !before.is_empty() {
elements.push(serde_json::json!({
"tag": "markdown",
"content": restore_code_blocks(before, &code_blocks)
"content": before
}));
}
@ -1972,9 +1963,14 @@ impl FeishuChannel {
let remaining = protected[last_end..].trim();
if !remaining.is_empty() {
// Restore code blocks
let mut final_content = remaining.to_string();
for (i, cb) in code_blocks.iter().enumerate() {
final_content = final_content.replace(&format!("\x00CODE{}\x00", i), cb);
}
elements.push(serde_json::json!({
"tag": "markdown",
"content": restore_code_blocks(remaining, &code_blocks)
"content": final_content
}));
}

5
src/command/adapters/channel.rs
View File

@ -133,11 +133,6 @@ impl InputAdapter for ChannelInputAdapter {
return Ok(Some(Command::GetCurrentSession));
}
// 解析 /stop 命令 - 停止当前执行的 Agent
if trimmed == "/stop" {
return Ok(Some(Command::StopExecution));
}
// 解析 /help 命令 - 显示所有支持的命令
if trimmed == "/help" {
return Ok(Some(Command::Help));

5
src/command/adapters/cli.rs
View File

@ -134,11 +134,6 @@ impl InputAdapter for CliInputAdapter {
return Ok(Some(Command::GetCurrentSession));
}
// 解析 /stop 命令 - 停止当前执行的 Agent
if trimmed == "/stop" {
return Ok(Some(Command::StopExecution));
}
// 解析 /help 命令 - 显示所有支持的命令
if trimmed == "/help" {
return Ok(Some(Command::Help));

1
src/command/handlers/mod.rs
View File

@ -11,7 +11,6 @@ pub mod load_topic;
pub mod save_session;
pub mod save_topic;
pub mod session;
pub mod stop_execution;
pub mod switch_topic;
// 导出公共函数供其他模块复用

57
src/command/handlers/stop_execution.rs
View File

@ -1,57 +0,0 @@
use async_trait::async_trait;
use crate::command::context::CommandContext;
use crate::command::handler::{CommandHandler, CommandMetadata};
use crate::command::response::{CommandError, CommandResponse, MessageKind};
use crate::command::Command;
use crate::gateway::cancel_manager::CancelManager;
/// 处理 StopExecution 命令：按话题取消当前正在执行的 Agent。
pub struct StopExecutionCommandHandler {
cancel_manager: CancelManager,
}
impl StopExecutionCommandHandler {
pub fn new(cancel_manager: CancelManager) -> Self {
Self { cancel_manager }
}
}
#[async_trait]
impl CommandHandler for StopExecutionCommandHandler {
fn can_handle(&self, cmd: &Command) -> bool {
matches!(cmd, Command::StopExecution)
}
fn metadata(&self) -> Option<CommandMetadata> {
Some(CommandMetadata {
name: "stop",
description: "停止当前话题正在执行的 Agent",
usage: "/stop",
})
}
async fn handle(
&self,
_cmd: Command,
ctx: CommandContext,
) -> Result<CommandResponse, CommandError> {
let topic_id = match ctx.topic_id.as_deref() {
Some(id) => id,
None => {
return Ok(CommandResponse::success(ctx.request_id)
.with_message(MessageKind::Notification, "当前没有活跃的话题，无法停止"));
}
};
let cancelled = self.cancel_manager.cancel_by_topic(topic_id).await;
if cancelled {
Ok(CommandResponse::success(ctx.request_id)
.with_message(MessageKind::Notification, "正在停止当前任务..."))
} else {
Ok(CommandResponse::success(ctx.request_id)
.with_message(MessageKind::Notification, "当前没有正在执行的任务"))
}
}
}

3
src/command/mod.rs
View File

@ -52,8 +52,6 @@ pub enum Command {
channel: String,
chat_id: String,
},
/// 停止当前正在执行的 Agent
StopExecution,
}
impl Command {
@ -74,7 +72,6 @@ impl Command {
Command::LoadTaskMessages { .. } => "load_task_messages",
Command::ListSchedulerJobs => "list_scheduler_jobs",
Command::LoadChatMessages { .. } => "load_chat_messages",
Command::StopExecution => "stop_execution",
}
}
}

11
src/gateway/agent_factory.rs
View File

@ -23,8 +23,6 @@ pub(crate) struct AgentBuildRequest<'a> {
pub(crate) sender_id: Option<&'a str>,
pub(crate) message_id: Option<&'a str>,
pub(crate) provider_config: LLMProviderConfig,
/// 取消信号接收端（可选）：Agent 在每次迭代时检查是否被取消
pub(crate) cancel_token: Option<tokio::sync::watch::Receiver<()>>,
}
impl AgentFactory {
@ -66,7 +64,7 @@ impl AgentFactory {
let tool_chat_id = request
.notification_chat_id
.unwrap_or(request.session_chat_id);
let mut agent = agent.with_tool_context(ToolContext {
agent.with_tool_context(ToolContext {
channel_name: Some(request.channel_name.to_string()),
sender_id: request.sender_id.map(str::to_string),
chat_id: Some(tool_chat_id.to_string()),
@ -75,12 +73,7 @@ impl AgentFactory {
message_id: request.message_id.map(str::to_string),
message_seq: None,
subagent_description: None,
});
// 如果有取消信号接收端，注入 Agent
if let Some(token) = request.cancel_token {
agent = agent.with_cancel_token(token);
}
agent
})
})
}
}

61
src/gateway/cancel_manager.rs
View File

@ -1,61 +0,0 @@
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::{Mutex, watch};
/// 共享的 Agent 取消注册表。
///
/// 每个正在执行的 Agent 在启动前按 topic_id 注册一个 watch::Sender，
/// 外部（如 /stop 命令）通过 cancel_by_topic() 发送取消信号。
/// Agent 循环内部通过 watch::Receiver::has_changed() 检测取消。
///
/// key 使用 topic_id（UUID），全局唯一，精确到话题级别。
#[derive(Clone)]
pub struct CancelManager {
tokens: Arc<Mutex<HashMap<String, watch::Sender<()>>>>,
}
impl CancelManager {
pub fn new() -> Self {
Self {
tokens: Arc::new(Mutex::new(HashMap::new())),
}
}
/// 按 topic_id 注册一个取消通道，返回 receiver 供 Agent 持有。
///
/// 如果同 topic_id 已有注册，旧 sender 被覆盖并 drop，
/// 旧 receiver 将收到通道关闭信号。
pub async fn register(&self, topic_id: &str) -> watch::Receiver<()> {
let (tx, rx) = watch::channel(());
self.tokens.lock().await.insert(topic_id.to_string(), tx);
rx
}
/// 按 topic_id 发送取消信号并移除注册条目。
///
/// 返回 `true` 表示找到了对应的任务并发送了取消信号，
/// 返回 `false` 表示没有找到对应的任务（可能已经完成或从未注册）。
pub async fn cancel_by_topic(&self, topic_id: &str) -> bool {
if let Some(tx) = self.tokens.lock().await.remove(topic_id) {
// send 可能失败（receiver 已被 drop），这不影响语义
let _ = tx.send(());
true
} else {
false
}
}
/// 按 topic_id 正常完成后清理注册条目（幂等）。
///
/// 与 cancel_by_topic() 不同，此方法不发送取消信号，仅移除条目。
/// 如果条目已被 cancel_by_topic() 移除，此调用为 no-op。
pub async fn remove_by_topic(&self, topic_id: &str) {
self.tokens.lock().await.remove(topic_id);
}
}
impl Default for CancelManager {
fn default() -> Self {
Self::new()
}
}

8
src/gateway/mod.rs
View File

@ -1,7 +1,6 @@
pub mod agent_factory;
pub mod agent_prompt_provider;
pub mod agent_task_executor;
pub mod cancel_manager;
pub mod cli_session;
pub mod command;
pub mod compaction;
@ -43,7 +42,6 @@ use crate::scheduler::Scheduler;
use crate::skills::SkillRuntime;
use crate::tools::task::repository::TaskRepository;
use agent_task_executor::{AgentTaskExecutor, SchedulerMaintenanceService};
use cancel_manager::CancelManager;
use outbound_dispatcher::OutboundDispatcher;
use processor::InboundProcessor;
use runtime::build_session_manager_with_sender;
@ -57,7 +55,6 @@ pub struct GatewayState {
pub channel_manager: ChannelManager,
pub bus: Arc<MessageBus>,
pub task_repository: Arc<dyn TaskRepository>,
pub cancel_manager: CancelManager,
}
impl GatewayState {
@ -99,15 +96,12 @@ impl GatewayState {
Some(bus.clone()),
)?;
let cancel_manager = CancelManager::new();
Ok(Self {
config,
session_manager,
channel_manager,
bus,
task_repository,
cancel_manager,
})
}
@ -128,7 +122,7 @@ impl GatewayState {
}
};
let inbound_processor =
InboundProcessor::new(self.bus.clone(), self.session_manager.clone(), semaphore, provider_config, self.cancel_manager.clone());
InboundProcessor::new(self.bus.clone(), self.session_manager.clone(), semaphore, provider_config);
tokio::spawn(inbound_processor.run());
// Spawn outbound dispatcher

27
src/gateway/processor.rs
View File

@ -14,11 +14,9 @@ use crate::command::handlers::load_topic::LoadTopicCommandHandler;
use crate::command::handlers::save_session::SaveSessionCommandHandler;
use crate::command::handlers::save_topic::SaveTopicCommandHandler;
use crate::command::handlers::session::SessionCommandHandler;
use crate::command::handlers::stop_execution::StopExecutionCommandHandler;
use crate::command::handlers::switch_topic::SwitchTopicCommandHandler;
use crate::config::LLMProviderConfig;
use crate::gateway::agent_prompt_provider::AgentPromptProvider;
use crate::gateway::cancel_manager::CancelManager;
use crate::providers::{create_provider, ProviderRuntimeConfig};
use crate::skills::SkillPromptProvider;
use crate::storage::persistent_session_id;
@ -33,7 +31,6 @@ pub struct InboundProcessor {
semaphore: Arc<Semaphore>,
provider_config: LLMProviderConfig,
command_router: Arc<CommandRouter>,
cancel_manager: CancelManager,
}
impl InboundProcessor {
@ -42,7 +39,6 @@ impl InboundProcessor {
session_manager: SessionManager,
semaphore: Arc<Semaphore>,
provider_config: LLMProviderConfig,
cancel_manager: CancelManager,
) -> Self {
// 创建命令路由器并注册处理器
let mut command_router = CommandRouter::new();
@ -101,18 +97,12 @@ impl InboundProcessor {
let metadata = command_router.metadata_arc();
command_router.register(Box::new(HelpCommandHandler::new(metadata)));
// 注册 stop_execution 处理器
command_router.register(Box::new(StopExecutionCommandHandler::new(
cancel_manager.clone(),
)));
Self {
bus,
session_manager,
semaphore,
provider_config,
command_router: Arc::new(command_router),
cancel_manager,
}
}
@ -246,18 +236,6 @@ impl InboundProcessor {
current_topic.clone(),
));
// 保存 channel 和 chat_id 用于后续清理（因 match 中可能 move inbound）
let channel = inbound.channel.clone();
let chat_id = inbound.chat_id.clone();
// 按 topic_id 注册取消信号：Agent 构建时通过 Session 消费该 receiver
if let Some(ref topic_id) = current_topic {
let cancel_rx = self.cancel_manager.register(topic_id).await;
self.session_manager
.set_agent_cancel_token(&channel, &chat_id, cancel_rx)
.await;
}
match self
.session_manager
.handle_message(
@ -330,11 +308,6 @@ impl InboundProcessor {
}
}
// 清理取消信号注册（幂等：如果已被 cancel_by_topic() 移除则为 no-op）
if let Some(ref topic_id) = current_topic {
self.cancel_manager.remove_by_topic(topic_id).await;
}
Ok(())
}
}

38
src/gateway/session.rs
View File

@ -45,9 +45,6 @@ pub struct Session {
compressor: ContextCompressor,
history: SessionHistory,
store: Arc<SessionStore>,
/// 等待中的取消信号接收端（按 chat_id 索引）。
/// 在 Agent 执行前由外部注入，Agent 构建时消费。
pending_cancel_tokens: HashMap<String, tokio::sync::watch::Receiver<()>>,
}
pub struct BusToolCallEmitter {
@ -166,7 +163,6 @@ impl Session {
skill_events,
),
store,
pending_cancel_tokens: HashMap::new(),
})
}
@ -183,19 +179,6 @@ impl Session {
}
}
/// 存入待使用的取消信号接收端。
///
/// 在 Agent 执行前由处理器调用，Agent 构建时（create_agent）自动消费。
/// 每个 chat_id 同时只允许一个 pending token；新 token 会替换旧 token。
pub fn set_cancel_receiver(
&mut self,
chat_id: &str,
receiver: tokio::sync::watch::Receiver<()>,
) {
self.pending_cancel_tokens
.insert(chat_id.to_string(), receiver);
}
/// 获取当前话题 ID（指定 chat）
pub fn current_topic(&self, chat_id: &str) -> Option<&str> {
self.history.chat_topic(chat_id)
@ -437,7 +420,7 @@ impl Session {
/// 创建一个临时的 AgentLoop 实例来处理消息
pub fn create_agent(
&mut self,
&self,
chat_id: &str,
sender_id: Option<&str>,
message_id: Option<&str>,
@ -452,15 +435,13 @@ impl Session {
}
pub fn create_agent_with_provider_config(
&mut self,
&self,
session_chat_id: &str,
notification_chat_id: Option<&str>,
sender_id: Option<&str>,
message_id: Option<&str>,
provider_config: LLMProviderConfig,
) -> Result<AgentLoop, AgentError> {
// 消费 pending 的取消信号接收端（如果存在）
let cancel_token = self.pending_cancel_tokens.remove(session_chat_id);
self.agent_factory.create(AgentBuildRequest {
channel_name: &self.channel_name,
session_chat_id,
@ -468,7 +449,6 @@ impl Session {
sender_id,
message_id,
provider_config,
cancel_token,
})
}
}
@ -627,20 +607,6 @@ impl SessionManager {
}
}
/// 存入 Agent 取消信号接收端，供 Agent 构建时消费。
///
/// 在 Agent 执行前由处理器调用。Agent 在 create_agent() 时自动取出。
pub async fn set_agent_cancel_token(
&self,
channel_name: &str,
chat_id: &str,
token: tokio::sync::watch::Receiver<()>,
) {
if let Some(session) = self.get(channel_name).await {
session.lock().await.set_cancel_receiver(chat_id, token);
}
}
/// 更新最后活跃时间
pub async fn touch(&self, channel_name: &str) {
self.lifecycle.touch(channel_name).await;

5
src/gateway/ws.rs
View File

@ -17,7 +17,6 @@ use crate::command::handlers::load_task_messages::LoadTaskMessagesCommandHandler
use crate::command::handlers::load_topic::LoadTopicCommandHandler;
use crate::command::handlers::save_session::SaveSessionCommandHandler;
use crate::command::handlers::session::SessionCommandHandler;
use crate::command::handlers::stop_execution::StopExecutionCommandHandler;
use crate::command::handlers::switch_topic::SwitchTopicCommandHandler;
use crate::gateway::agent_prompt_provider::AgentPromptProvider;
use crate::protocol::{WsInbound, WsOutbound, MediaSummary, parse_inbound, serialize_outbound};
@ -406,10 +405,6 @@ async fn handle_inbound(
router.register(Box::new(ListSchedulerJobsCommandHandler::new(store.clone())));
// 注册 load_chat_messages 处理器
router.register(Box::new(LoadChatMessagesCommandHandler::new()));
// 注册 stop_execution 处理器
router.register(Box::new(StopExecutionCommandHandler::new(
state.cancel_manager.clone(),
)));
// 构建命令上下文
tracing::debug!(

2
src/protocol/mod.rs
View File

@ -187,8 +187,6 @@ pub enum WsOutbound {
SchedulerJobList {
jobs: Vec<SchedulerJobSummary>,
},
#[serde(rename = "execution_cancelled")]
ExecutionCancelled { message: String },
#[serde(rename = "pong")]
Pong,
}

1
tests/test_request_format.rs
View File

@ -155,7 +155,6 @@ fn test_tool_result_outbound_serialization() {
content: "工具结果: calculator\n\n2".to_string(),
role: "tool".to_string(),
subagent_task_id: None,
duration_ms: None,
};
let json = serde_json::to_string(&msg).unwrap();

11
web/src/App.tsx
View File

@ -51,7 +51,6 @@ function App() {
requestTopicList,
enterSubAgentView,
exitSubAgentView,
handleStop,
} = useChat()
const { status, sendMessage } = useWebSocket({
@ -128,9 +127,6 @@ function App() {
case 'save':
cmd = { type: 'save_topic', filepath: args[0] || undefined, include_subagents: false }
break
case 'stop':
cmd = { type: 'stop_execution' }
break
default:
alert(`Unknown command: /${command}`)
return
@ -151,12 +147,6 @@ function App() {
[sendMessage, handleMessage, handleCommand, sessionId, chatId, isReadOnly]
)
const handleStopExecution = useCallback(() => {
const cmd = handleStop()
handleCommand(cmd)
sendMessage({ type: 'command', payload: JSON.stringify(cmd) })
}, [sendMessage, handleCommand, handleStop])
const handleCreateTopic = useCallback(() => {
if (isReadOnly || !sessionId) {
return
@ -425,7 +415,6 @@ function App() {
}
onSendMessage={subAgentView || schedulerView ? () => {} : handleSendMessage}
onNavigateToSubAgent={handleNavigateToSubAgent}
onStop={handleStopExecution}
/>
</div>
</div>

3
web/src/components/Chat/ChatContainer.tsx
View File

@ -9,7 +9,6 @@ interface ChatContainerProps {
channelName?: string
onSendMessage: (content: string, attachments: Attachment[]) => void
onNavigateToSubAgent?: (taskId: string, description: string) => void
onStop?: () => void
}
export function ChatContainer({
@ -19,7 +18,6 @@ export function ChatContainer({
channelName,
onSendMessage,
onNavigateToSubAgent,
onStop,
}: ChatContainerProps) {
return (
<div className="flex h-full flex-col">
@ -28,7 +26,6 @@ export function ChatContainer({
</div>
<MessageInput
onSend={onSendMessage}
onStop={onStop}
disabled={isLoading}
isLoading={isLoading}
isReadOnly={isReadOnly}

2
web/src/components/Chat/MessageBubble.tsx
View File

@ -207,7 +207,7 @@ export function MessageBubble({ message, onNavigateToSubAgent }: MessageBubblePr
const displayContent = hasResult ? stripToolResultPrefix(message.resultContent!) : ''
const isTaskTool = message.toolName === 'task'
const taskResult = isTaskTool && hasResult ? parseTaskResult(displayContent) : null
const taskResult = isTaskTool && hasResult ? parseTaskResult(message.resultContent!) : null
const isSubAgent = !!message.subagentTaskId
const subagentType = (message.arguments as Record<string, unknown> | null)?.subagent_type as string || 'general'
const taskDescription = (message.arguments as Record<string, unknown> | null)?.description as string || ''

16
web/src/components/Chat/MessageInput.tsx
View File

@ -1,4 +1,4 @@
import { Send, Loader2, Square, Sparkles, Eye, Paperclip, X, FileIcon, ImageIcon, MusicIcon, VideoIcon } from 'lucide-react'
import { Send, Loader2, Sparkles, Eye, Paperclip, X, FileIcon, ImageIcon, MusicIcon, VideoIcon } from 'lucide-react'
import { useState, useRef, useEffect } from 'react'
import type { Attachment } from '../../types/protocol'
@ -6,7 +6,6 @@ const MAX_FILE_SIZE = 50 * 1024 * 1024 // 50MB
interface MessageInputProps {
onSend: (content: string, attachments: Attachment[]) => void
onStop?: () => void
disabled?: boolean
isLoading?: boolean
placeholder?: string
@ -30,7 +29,6 @@ function getMediaType(mimeType: string): string {
export function MessageInput({
onSend,
onStop,
disabled = false,
isLoading = false,
placeholder = '输入消息...按 / 查看命令',
@ -368,16 +366,7 @@ export function MessageInput({
<Sparkles className="absolute right-3 top-1/2 -translate-y-1/2 h-4 w-4 text-zinc-600 pointer-events-none" />
</div>
{/* 发送/停止按钮 */}
{isLoading && onStop ? (
<button
onClick={onStop}
title="停止执行"
className="flex h-10 w-10 shrink-0 items-center justify-center rounded-xl bg-red-500/10 border border-red-500/20 text-red-400 hover:bg-red-500/20 hover:text-red-300 transition-all"
>
<Square className="h-4 w-4" />
</button>
) : (
{/* 发送按钮 */}
<button
onClick={handleSend}
disabled={disabled || (!content.trim() && attachments.length === 0)}
@ -389,7 +378,6 @@ export function MessageInput({
<Send className="h-4 w-4" />
)}
</button>
)}
</div>
{/* 提示 */}

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

@ -72,9 +72,6 @@ interface UseChatReturn {
schedulerView: SchedulerJobView | null
enterSchedulerJobView: (lookup: SchedulerJobSessionLookup, jobId: string, description: string) => Command
exitSchedulerJobView: () => void
// 停止当前 Agent 执行
handleStop: () => Command
}
interface SubAgentView {
@ -260,8 +257,8 @@ export function useChat(): UseChatReturn {
const msgSubagentTaskId = getSubagentTaskId(message)
if (msgSubagentTaskId && msgSubagentTaskId === currentSubAgentView.taskId) {
appendToSubAgentViewMessage(message)
return
}
return
}
// In main view, skip sub-agent messages (they belong to sub-agent view).
@ -413,21 +410,6 @@ export function useChat(): UseChatReturn {
break
}
case 'execution_cancelled': {
setMessages((prev) => [
...prev,
{
id: generateMessageId(),
role: 'assistant',
content: (message as { type: 'execution_cancelled'; message: string }).message,
timestamp: Date.now(),
type: 'message',
},
])
setIsLoading(false)
break
}
case 'error': {
setMessages((prev) => [
...prev,
@ -586,10 +568,6 @@ export function useChat(): UseChatReturn {
setSchedulerView(null)
}, [])
const handleStop = useCallback((): Command => {
return { type: 'stop_execution' }
}, [])
// Memoize messages: sub-agent view > scheduler view > main
const resolvedMessages = useMemo(() => {
if (subAgentView) {
@ -634,6 +612,5 @@ export function useChat(): UseChatReturn {
schedulerView,
enterSchedulerJobView,
exitSchedulerJobView,
handleStop,
}
}

11
web/src/types/protocol.ts
View File

@ -192,11 +192,6 @@ export interface TaskMessagesLoaded {
summary?: string
}
export interface ExecutionCancelled {
type: 'execution_cancelled'
message: string
}
export type WsOutbound =
| AssistantResponse
| ToolCall
@ -212,7 +207,6 @@ export type WsOutbound =
| ChannelList
| TaskMessagesLoaded
| SchedulerJobList
| ExecutionCancelled
| Pong
// ============================================================================
@ -290,10 +284,6 @@ export interface LoadChatMessagesCommand {
chat_id: string
}
export interface StopExecutionCommand {
type: 'stop_execution'
}
export type Command =
| CreateSessionCommand
| ListSessionsCommand
@ -309,7 +299,6 @@ export type Command =
| LoadTaskMessagesCommand
| ListSchedulerJobsCommand
| LoadChatMessagesCommand
| StopExecutionCommand
// ============================================================================
// UI Types