e0313ab8f3
- agent_loop/processor: catch_unwind 隔离 panic，防止消息静默丢失 - calculator: 拒绝 NaN/Infinity 输入，修复阶乘溢出（上限 34），拒绝非有限表达式结果 - message: 修复 sanitize 两阶段删除索引未排序导致的越界 panic - utils: 新增 panic_payload_message 提取可读 panic 消息 - .gitignore: 忽略 artifacts/ 测试产物目录
1076 lines
38 KiB
Rust
1076 lines
38 KiB
Rust
use serde::{Deserialize, Serialize};
|
||
use std::collections::{HashMap, HashSet};
|
||
|
||
use crate::domain::messages::ToolCall;
|
||
use crate::utils::current_timestamp;
|
||
|
||
pub const SYSTEM_CONTEXT_AGENT_PROMPT: &str = "agent_prompt";
|
||
pub const SYSTEM_CONTEXT_SCHEDULED_PROMPT: &str = "scheduled_system_prompt";
|
||
pub const SYSTEM_CONTEXT_HISTORY_COMPACTION: &str = "history_compaction";
|
||
|
||
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
|
||
#[serde(rename_all = "snake_case")]
|
||
pub enum ToolMessageState {
|
||
Completed,
|
||
PendingUserAction,
|
||
}
|
||
|
||
// ============================================================================
|
||
// MediaItem - Media metadata for messages
|
||
// ============================================================================
|
||
|
||
#[derive(Debug, Clone)]
|
||
pub struct MediaItem {
|
||
pub path: String, // Local file path
|
||
pub media_type: String, // "image", "audio", "file", "video"
|
||
pub mime_type: Option<String>,
|
||
pub original_key: Option<String>, // Feishu file_key for download
|
||
pub content_base64: Option<String>, // Base64-encoded file content for web download
|
||
pub file_name: Option<String>, // Display file name
|
||
}
|
||
|
||
impl MediaItem {
|
||
pub fn new(path: impl Into<String>, media_type: impl Into<String>) -> Self {
|
||
Self {
|
||
path: path.into(),
|
||
media_type: media_type.into(),
|
||
mime_type: None,
|
||
original_key: None,
|
||
content_base64: None,
|
||
file_name: None,
|
||
}
|
||
}
|
||
}
|
||
|
||
// ============================================================================
|
||
// ChatMessage - Used by AgentLoop for LLM conversation history
|
||
// ============================================================================
|
||
|
||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||
pub struct ChatMessage {
|
||
pub id: String,
|
||
pub role: String,
|
||
pub content: String,
|
||
pub media_refs: Vec<String>, // Paths to media files for context
|
||
pub timestamp: i64,
|
||
#[serde(skip_serializing_if = "Option::is_none")]
|
||
pub system_context: Option<String>,
|
||
#[serde(skip_serializing_if = "Option::is_none")]
|
||
pub reasoning_content: Option<String>,
|
||
#[serde(skip_serializing_if = "Option::is_none")]
|
||
pub tool_call_id: Option<String>,
|
||
#[serde(skip_serializing_if = "Option::is_none")]
|
||
pub tool_name: Option<String>,
|
||
#[serde(skip_serializing_if = "Option::is_none")]
|
||
pub tool_state: Option<ToolMessageState>,
|
||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||
pub tool_duration_ms: Option<u64>,
|
||
#[serde(skip_serializing_if = "Option::is_none")]
|
||
pub tool_calls: Option<Vec<ToolCall>>,
|
||
/// LLM 调用 usage（仅 assistant 消息有值，来自 provider 响应）
|
||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||
pub usage: Option<MessageUsage>,
|
||
}
|
||
|
||
/// 单次 LLM 调用的 token 用量
|
||
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
|
||
pub struct MessageUsage {
|
||
pub prompt_tokens: u32,
|
||
pub completion_tokens: u32,
|
||
pub total_tokens: u32,
|
||
/// 本次调用所用模型的上下文窗口大小（来自 AgentRuntimeConfig）。
|
||
/// 与 prompt_tokens 一起持久化，用于计算上下文占用率。
|
||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||
pub context_window_tokens: Option<u32>,
|
||
}
|
||
|
||
impl MessageUsage {
|
||
pub fn from_provider_usage(u: crate::providers::Usage) -> Self {
|
||
Self {
|
||
prompt_tokens: u.prompt_tokens,
|
||
completion_tokens: u.completion_tokens,
|
||
total_tokens: u.total_tokens,
|
||
context_window_tokens: None,
|
||
}
|
||
}
|
||
|
||
/// 链式设置 context_window_tokens（来自 AgentRuntimeConfig）
|
||
pub fn with_context_window(mut self, ctx: usize) -> Self {
|
||
self.context_window_tokens = Some(ctx as u32);
|
||
self
|
||
}
|
||
}
|
||
|
||
impl ChatMessage {
|
||
pub fn user(content: impl Into<String>) -> Self {
|
||
Self {
|
||
id: uuid::Uuid::new_v4().to_string(),
|
||
role: "user".to_string(),
|
||
content: content.into(),
|
||
media_refs: Vec::new(),
|
||
timestamp: current_timestamp(),
|
||
system_context: None,
|
||
reasoning_content: None,
|
||
tool_call_id: None,
|
||
tool_name: None,
|
||
tool_duration_ms: None,
|
||
tool_state: None,
|
||
tool_calls: None,
|
||
usage: None,
|
||
}
|
||
}
|
||
|
||
pub fn user_with_media(content: impl Into<String>, media_refs: Vec<String>) -> Self {
|
||
Self {
|
||
id: uuid::Uuid::new_v4().to_string(),
|
||
role: "user".to_string(),
|
||
content: content.into(),
|
||
media_refs,
|
||
timestamp: current_timestamp(),
|
||
system_context: None,
|
||
reasoning_content: None,
|
||
tool_call_id: None,
|
||
tool_name: None,
|
||
tool_duration_ms: None,
|
||
tool_state: None,
|
||
tool_calls: None,
|
||
usage: None,
|
||
}
|
||
}
|
||
|
||
pub fn assistant(content: impl Into<String>) -> Self {
|
||
Self {
|
||
id: uuid::Uuid::new_v4().to_string(),
|
||
role: "assistant".to_string(),
|
||
content: content.into(),
|
||
media_refs: Vec::new(),
|
||
timestamp: current_timestamp(),
|
||
system_context: None,
|
||
reasoning_content: None,
|
||
tool_call_id: None,
|
||
tool_name: None,
|
||
tool_duration_ms: None,
|
||
tool_state: None,
|
||
tool_calls: None,
|
||
usage: None,
|
||
}
|
||
}
|
||
|
||
pub fn assistant_with_reasoning(
|
||
content: impl Into<String>,
|
||
reasoning_content: impl Into<String>,
|
||
) -> Self {
|
||
let mut message = Self::assistant(content);
|
||
message.reasoning_content = Some(reasoning_content.into());
|
||
message
|
||
}
|
||
|
||
pub fn assistant_with_tool_calls(
|
||
content: impl Into<String>,
|
||
tool_calls: Vec<ToolCall>,
|
||
) -> Self {
|
||
Self {
|
||
id: uuid::Uuid::new_v4().to_string(),
|
||
role: "assistant".to_string(),
|
||
content: content.into(),
|
||
media_refs: Vec::new(),
|
||
timestamp: current_timestamp(),
|
||
system_context: None,
|
||
reasoning_content: None,
|
||
tool_call_id: None,
|
||
tool_name: None,
|
||
tool_duration_ms: None,
|
||
tool_state: None,
|
||
tool_calls: Some(tool_calls),
|
||
usage: None,
|
||
}
|
||
}
|
||
|
||
pub fn assistant_with_tool_calls_and_reasoning(
|
||
content: impl Into<String>,
|
||
tool_calls: Vec<ToolCall>,
|
||
reasoning_content: impl Into<String>,
|
||
) -> Self {
|
||
let mut message = Self::assistant_with_tool_calls(content, tool_calls);
|
||
message.reasoning_content = Some(reasoning_content.into());
|
||
message
|
||
}
|
||
|
||
pub fn system(content: impl Into<String>) -> Self {
|
||
Self::system_with_context(content, None::<String>)
|
||
}
|
||
|
||
pub fn system_with_context(
|
||
content: impl Into<String>,
|
||
system_context: impl Into<Option<String>>,
|
||
) -> Self {
|
||
Self {
|
||
id: uuid::Uuid::new_v4().to_string(),
|
||
role: "system".to_string(),
|
||
content: content.into(),
|
||
media_refs: Vec::new(),
|
||
timestamp: current_timestamp(),
|
||
system_context: system_context.into(),
|
||
reasoning_content: None,
|
||
tool_call_id: None,
|
||
tool_name: None,
|
||
tool_duration_ms: None,
|
||
tool_state: None,
|
||
tool_calls: None,
|
||
usage: None,
|
||
}
|
||
}
|
||
|
||
pub fn tool(
|
||
tool_call_id: impl Into<String>,
|
||
tool_name: impl Into<String>,
|
||
content: impl Into<String>,
|
||
) -> Self {
|
||
Self::tool_with_state(
|
||
tool_call_id,
|
||
tool_name,
|
||
content,
|
||
ToolMessageState::Completed,
|
||
)
|
||
}
|
||
|
||
pub fn tool_with_state(
|
||
tool_call_id: impl Into<String>,
|
||
tool_name: impl Into<String>,
|
||
content: impl Into<String>,
|
||
tool_state: ToolMessageState,
|
||
) -> Self {
|
||
Self {
|
||
id: uuid::Uuid::new_v4().to_string(),
|
||
role: "tool".to_string(),
|
||
content: content.into(),
|
||
media_refs: Vec::new(),
|
||
timestamp: current_timestamp(),
|
||
system_context: None,
|
||
reasoning_content: None,
|
||
tool_call_id: Some(tool_call_id.into()),
|
||
tool_name: Some(tool_name.into()),
|
||
tool_duration_ms: None,
|
||
tool_state: Some(tool_state),
|
||
tool_calls: None,
|
||
usage: None,
|
||
}
|
||
}
|
||
|
||
pub fn with_tool_duration(mut self, ms: u64) -> Self {
|
||
self.tool_duration_ms = Some(ms);
|
||
self
|
||
}
|
||
|
||
pub fn has_system_context(&self, expected: &str) -> bool {
|
||
self.system_context.as_deref() == Some(expected)
|
||
}
|
||
|
||
pub fn is_assistant_tool_call_message(&self) -> bool {
|
||
self.role == "assistant"
|
||
&& self
|
||
.tool_calls
|
||
.as_ref()
|
||
.map(|calls| !calls.is_empty())
|
||
.unwrap_or(false)
|
||
}
|
||
}
|
||
|
||
// ============================================================================
|
||
// Message sanitization
|
||
// ============================================================================
|
||
|
||
/// 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).
|
||
///
|
||
/// Incomplete sequences can appear in the middle of history when:
|
||
/// 1. The process was interrupted mid-execution, 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".
|
||
///
|
||
/// Returns the number of messages removed.
|
||
pub(crate) fn sanitize_incomplete_tool_call_sequences(messages: &mut Vec<ChatMessage>) -> usize {
|
||
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();
|
||
|
||
// 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.iter().all(|tc| resolved_ids.contains(&tc.id));
|
||
|
||
if all_have_results {
|
||
for tc in tool_calls.iter() {
|
||
with_parent.insert(tc.id.clone());
|
||
}
|
||
} else {
|
||
let missing_count = tool_calls
|
||
.iter()
|
||
.filter(|tc| !resolved_ids.contains(&tc.id))
|
||
.count();
|
||
|
||
tracing::warn!(
|
||
tool_call_count = tool_calls.len(),
|
||
missing_tool_results = missing_count,
|
||
message_id = %msg.id,
|
||
message_index = i,
|
||
"Removing assistant message with incomplete tool call sequence — \
|
||
tool results were never persisted (likely due to process interruption \
|
||
or history compaction preserving an orphan)"
|
||
);
|
||
|
||
remove_indices.push(i);
|
||
}
|
||
}
|
||
}
|
||
|
||
// Phase 1.5: Forward-order check — verify tool messages IMMEDIATELY follow
|
||
// the assistant(tool_calls). If any non-tool message appears between the
|
||
// assistant and its tool results, the API rejects with
|
||
// "insufficient tool messages following tool_calls message".
|
||
//
|
||
// The reverse scan in Phase 1 only checks existence (tool result appears
|
||
// somewhere after assistant), NOT immediacy. This pass catches cases like:
|
||
// [assistant(tool_calls=[A]), user, tool(A)]
|
||
// ^ Phase 1 sees tool(A) after assistant → "resolved"
|
||
// but API requires tool(A) to be IMMEDIATELY after assistant
|
||
{
|
||
let mut pending_tool_ids: HashSet<String> = HashSet::new();
|
||
let mut pending_assistant_idx: Option<usize> = None;
|
||
|
||
for (i, m) in messages.iter().enumerate() {
|
||
// If we have pending tool_ids and encounter a non-tool message,
|
||
// the assistant's tool results were NOT immediately following.
|
||
if !pending_tool_ids.is_empty() && m.role != "tool" {
|
||
if let Some(idx) = pending_assistant_idx {
|
||
if !remove_indices.contains(&idx) {
|
||
tracing::warn!(
|
||
message_index = idx,
|
||
interrupted_by_index = i,
|
||
interrupted_by_role = %m.role,
|
||
pending_tool_call_count = pending_tool_ids.len(),
|
||
"Removing assistant with tool_calls — tool results \
|
||
not immediately following (interrupted by non-tool message)"
|
||
);
|
||
// Remove this assistant's tool_call_ids from with_parent
|
||
// so Phase 2 cleans up the now-orphaned tool messages
|
||
if let Some(calls) = messages[idx].tool_calls.as_ref() {
|
||
for tc in calls.iter() {
|
||
with_parent.remove(&tc.id);
|
||
}
|
||
}
|
||
remove_indices.push(idx);
|
||
}
|
||
}
|
||
pending_tool_ids.clear();
|
||
pending_assistant_idx = None;
|
||
}
|
||
|
||
if m.role == "assistant"
|
||
&& m.tool_calls
|
||
.as_ref()
|
||
.map_or(false, |calls| !calls.is_empty())
|
||
{
|
||
let already_marked = remove_indices.contains(&i);
|
||
if !already_marked {
|
||
pending_tool_ids = m
|
||
.tool_calls
|
||
.as_ref()
|
||
.unwrap()
|
||
.iter()
|
||
.map(|tc| tc.id.clone())
|
||
.collect();
|
||
pending_assistant_idx = Some(i);
|
||
}
|
||
} else if m.role == "tool" {
|
||
if let Some(ref tc_id) = m.tool_call_id {
|
||
pending_tool_ids.remove(tc_id);
|
||
if pending_tool_ids.is_empty() {
|
||
pending_assistant_idx = None;
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
// Handle trailing assistant with unresolved immediate tool results
|
||
if !pending_tool_ids.is_empty() {
|
||
if let Some(idx) = pending_assistant_idx {
|
||
if !remove_indices.contains(&idx) {
|
||
tracing::warn!(
|
||
message_index = idx,
|
||
"Removing trailing assistant with incomplete immediate tool results"
|
||
);
|
||
if let Some(calls) = messages[idx].tool_calls.as_ref() {
|
||
for tc in calls.iter() {
|
||
with_parent.remove(&tc.id);
|
||
}
|
||
}
|
||
remove_indices.push(idx);
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
// Remove in descending index order to avoid shifting.
|
||
// 两阶段产出的索引并非全局降序：Phase 1（反向扫描）按降序追加，
|
||
// Phase 1.5（正向扫描）按升序追加。逐个 Vec::remove 前必须全局排序，
|
||
// 否则已删除元素会使后续索引漂移（删错消息）甚至越界 panic。
|
||
remove_indices.sort_unstable_by(|a, b| b.cmp(a));
|
||
remove_indices.dedup();
|
||
for &idx in &remove_indices {
|
||
messages.remove(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.
|
||
//
|
||
// Always execute this pass unconditionally — even when Phase 1 found no
|
||
// assistant messages with tool_calls (e.g., after heavy compaction that
|
||
// summarized all tool_call sequences into text), there may still be
|
||
// orphaned tool result messages in the history that must be cleaned up.
|
||
{
|
||
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,
|
||
};
|
||
if is_orphaned {
|
||
tracing::warn!(
|
||
tool_call_id = ?msg.tool_call_id,
|
||
message_id = %msg.id,
|
||
message_index = i,
|
||
"Removing orphaned tool result message — its parent assistant \
|
||
tool_calls message was removed or never persisted"
|
||
);
|
||
messages.remove(i);
|
||
removed += 1;
|
||
continue;
|
||
}
|
||
}
|
||
i += 1;
|
||
}
|
||
}
|
||
|
||
removed
|
||
}
|
||
|
||
// ============================================================================
|
||
// InboundMessage - Message from Channel to Bus (user input)
|
||
// ============================================================================
|
||
|
||
#[derive(Debug, Clone)]
|
||
pub struct InboundMessage {
|
||
pub channel: String,
|
||
pub sender_id: String,
|
||
pub chat_id: String,
|
||
pub content: String,
|
||
pub timestamp: i64,
|
||
pub media: Vec<MediaItem>,
|
||
/// Channel-specific data used internally by the channel (not forwarded).
|
||
pub metadata: HashMap<String, String>,
|
||
/// Data forwarded from inbound to outbound (copied to OutboundMessage.metadata by gateway).
|
||
pub forwarded_metadata: HashMap<String, String>,
|
||
/// 端到端追踪 ID（由 channel 在构造消息时生成，贯穿 bus→processor→agent_loop→provider→tool 全链路）。
|
||
/// 基础设施层元数据，不进入 domain 层。
|
||
pub trace_id: String,
|
||
}
|
||
|
||
impl InboundMessage {
|
||
pub fn session_key(&self) -> String {
|
||
format!("{}:{}", self.channel, self.chat_id)
|
||
}
|
||
}
|
||
|
||
// ============================================================================
|
||
// OutboundMessage - Message from Agent to Channel (bot response)
|
||
// ============================================================================
|
||
|
||
#[derive(Debug, Clone)]
|
||
pub struct OutboundMessage {
|
||
pub channel: String,
|
||
/// 消息发送目标 ID（如飞书 open_id、微信 chat_id）
|
||
/// 注意：这始终是原始入站消息的 chat_id，不会被修改为会话 ID
|
||
pub chat_id: String,
|
||
/// 内部会话 ID（对应 sessions.id）
|
||
/// 用于会话管理和消息持久化，与消息发送目标无关
|
||
pub session_id: Option<String>,
|
||
pub content: String,
|
||
pub reply_to: Option<String>,
|
||
pub media: Vec<MediaItem>,
|
||
pub metadata: HashMap<String, String>,
|
||
pub event_kind: OutboundEventKind,
|
||
pub role: String,
|
||
pub tool_call_id: Option<String>,
|
||
pub tool_name: Option<String>,
|
||
pub tool_arguments: Option<serde_json::Value>,
|
||
pub reasoning_content: Option<String>,
|
||
/// Carry the originating ChatMessage.id so the WS layer can use it
|
||
/// instead of generating a random UUID. Critical for stream delta → assistant_response
|
||
/// ID matching on the front-end.
|
||
pub message_id: Option<String>,
|
||
/// 端到端追踪 ID（从 InboundMessage 继承，用于 outbound dispatcher 日志关联）。
|
||
/// 非 agent 执行路径产生的消息（如 scheduler 通知）此字段为空。
|
||
pub trace_id: String,
|
||
}
|
||
|
||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||
pub enum OutboundEventKind {
|
||
AssistantResponse,
|
||
ToolCall,
|
||
ToolResult,
|
||
ToolPending,
|
||
SchedulerNotification,
|
||
ErrorNotification,
|
||
TaskStarted,
|
||
/// 流式文本增量
|
||
StreamDelta,
|
||
/// 流式结束信号
|
||
StreamEnd,
|
||
/// 智能体执行完全结束（不再有后续工具调用或 LLM 迭代）
|
||
ExecutionCompleted,
|
||
}
|
||
|
||
impl OutboundMessage {
|
||
pub fn is_stream_delta(&self) -> bool {
|
||
matches!(
|
||
self.event_kind,
|
||
OutboundEventKind::StreamDelta | OutboundEventKind::StreamEnd
|
||
)
|
||
}
|
||
|
||
/// 设置 trace_id（builder 模式，用于 agent 执行路径中从 InboundMessage 继承）。
|
||
pub fn with_trace_id(mut self, trace_id: impl Into<String>) -> Self {
|
||
self.trace_id = trace_id.into();
|
||
self
|
||
}
|
||
|
||
pub fn assistant(
|
||
channel: impl Into<String>,
|
||
chat_id: impl Into<String>,
|
||
session_id: Option<String>,
|
||
content: impl Into<String>,
|
||
reply_to: Option<String>,
|
||
metadata: HashMap<String, String>,
|
||
) -> Self {
|
||
Self {
|
||
channel: channel.into(),
|
||
chat_id: chat_id.into(),
|
||
session_id,
|
||
content: content.into(),
|
||
reply_to,
|
||
media: Vec::new(),
|
||
metadata,
|
||
event_kind: OutboundEventKind::AssistantResponse,
|
||
role: "assistant".to_string(),
|
||
tool_call_id: None,
|
||
tool_name: None,
|
||
tool_arguments: None,
|
||
reasoning_content: None,
|
||
message_id: None,
|
||
trace_id: String::new(),
|
||
}
|
||
}
|
||
|
||
pub fn scheduler_notification(
|
||
channel: impl Into<String>,
|
||
chat_id: impl Into<String>,
|
||
session_id: Option<String>,
|
||
content: impl Into<String>,
|
||
reply_to: Option<String>,
|
||
metadata: HashMap<String, String>,
|
||
) -> Self {
|
||
let mut message =
|
||
Self::assistant(channel, chat_id, session_id, content, reply_to, metadata);
|
||
message.event_kind = OutboundEventKind::SchedulerNotification;
|
||
message
|
||
}
|
||
|
||
pub fn error_notification(
|
||
channel: impl Into<String>,
|
||
chat_id: impl Into<String>,
|
||
session_id: Option<String>,
|
||
content: impl Into<String>,
|
||
reply_to: Option<String>,
|
||
metadata: HashMap<String, String>,
|
||
) -> Self {
|
||
let mut message =
|
||
Self::assistant(channel, chat_id, session_id, content, reply_to, metadata);
|
||
message.event_kind = OutboundEventKind::ErrorNotification;
|
||
message
|
||
}
|
||
|
||
pub fn tool_call(
|
||
channel: impl Into<String>,
|
||
chat_id: impl Into<String>,
|
||
session_id: Option<String>,
|
||
message_id: impl Into<String>,
|
||
tool_name: impl Into<String>,
|
||
tool_arguments: serde_json::Value,
|
||
reply_to: Option<String>,
|
||
metadata: HashMap<String, String>,
|
||
) -> Self {
|
||
let tool_name = tool_name.into();
|
||
let content = format_tool_call_content(&tool_name, &tool_arguments);
|
||
Self {
|
||
channel: channel.into(),
|
||
chat_id: chat_id.into(),
|
||
session_id,
|
||
content,
|
||
reply_to,
|
||
media: Vec::new(),
|
||
metadata,
|
||
event_kind: OutboundEventKind::ToolCall,
|
||
role: "assistant".to_string(),
|
||
tool_call_id: Some(message_id.into()),
|
||
tool_name: Some(tool_name),
|
||
tool_arguments: Some(tool_arguments),
|
||
reasoning_content: None,
|
||
message_id: None,
|
||
trace_id: String::new(),
|
||
}
|
||
}
|
||
|
||
pub fn tool_result(
|
||
channel: impl Into<String>,
|
||
chat_id: impl Into<String>,
|
||
session_id: Option<String>,
|
||
tool_call_id: impl Into<String>,
|
||
tool_name: impl Into<String>,
|
||
content: impl Into<String>,
|
||
reply_to: Option<String>,
|
||
metadata: HashMap<String, String>,
|
||
) -> Self {
|
||
let tool_name = tool_name.into();
|
||
let raw_content = content.into();
|
||
let content = format_tool_result_content(&tool_name, &raw_content);
|
||
Self {
|
||
channel: channel.into(),
|
||
chat_id: chat_id.into(),
|
||
session_id,
|
||
content,
|
||
reply_to,
|
||
media: Vec::new(),
|
||
metadata,
|
||
event_kind: OutboundEventKind::ToolResult,
|
||
role: "tool".to_string(),
|
||
tool_call_id: Some(tool_call_id.into()),
|
||
tool_name: Some(tool_name),
|
||
tool_arguments: None,
|
||
reasoning_content: None,
|
||
message_id: None,
|
||
trace_id: String::new(),
|
||
}
|
||
}
|
||
|
||
pub fn tool_pending(
|
||
channel: impl Into<String>,
|
||
chat_id: impl Into<String>,
|
||
session_id: Option<String>,
|
||
tool_call_id: impl Into<String>,
|
||
tool_name: impl Into<String>,
|
||
content: impl Into<String>,
|
||
reply_to: Option<String>,
|
||
metadata: HashMap<String, String>,
|
||
) -> Self {
|
||
let tool_name = tool_name.into();
|
||
let raw_content = content.into();
|
||
let content = format_tool_result_content(&tool_name, &raw_content);
|
||
Self {
|
||
channel: channel.into(),
|
||
chat_id: chat_id.into(),
|
||
session_id,
|
||
content,
|
||
reply_to,
|
||
media: Vec::new(),
|
||
metadata,
|
||
event_kind: OutboundEventKind::ToolPending,
|
||
role: "tool".to_string(),
|
||
tool_call_id: Some(tool_call_id.into()),
|
||
tool_name: Some(tool_name),
|
||
tool_arguments: None,
|
||
reasoning_content: None,
|
||
message_id: None,
|
||
trace_id: String::new(),
|
||
}
|
||
}
|
||
|
||
/// 构造流式文本增量消息
|
||
pub fn stream_delta(
|
||
channel: impl Into<String>,
|
||
chat_id: impl Into<String>,
|
||
session_id: Option<String>,
|
||
message_id: impl Into<String>,
|
||
delta: impl Into<String>,
|
||
reasoning_delta: Option<String>,
|
||
metadata: HashMap<String, String>,
|
||
) -> Self {
|
||
Self {
|
||
channel: channel.into(),
|
||
chat_id: chat_id.into(),
|
||
session_id,
|
||
content: delta.into(),
|
||
reply_to: None,
|
||
media: Vec::new(),
|
||
metadata,
|
||
event_kind: OutboundEventKind::StreamDelta,
|
||
role: "assistant".to_string(),
|
||
tool_call_id: Some(message_id.into()),
|
||
tool_name: None,
|
||
tool_arguments: None,
|
||
reasoning_content: reasoning_delta,
|
||
message_id: None,
|
||
trace_id: String::new(),
|
||
}
|
||
}
|
||
|
||
/// 构造流式结束信号
|
||
pub fn stream_end(
|
||
channel: impl Into<String>,
|
||
chat_id: impl Into<String>,
|
||
session_id: Option<String>,
|
||
message_id: impl Into<String>,
|
||
metadata: HashMap<String, String>,
|
||
) -> Self {
|
||
Self {
|
||
channel: channel.into(),
|
||
chat_id: chat_id.into(),
|
||
session_id,
|
||
content: String::new(),
|
||
reply_to: None,
|
||
media: Vec::new(),
|
||
metadata,
|
||
event_kind: OutboundEventKind::StreamEnd,
|
||
role: "assistant".to_string(),
|
||
tool_call_id: Some(message_id.into()),
|
||
tool_name: None,
|
||
tool_arguments: None,
|
||
reasoning_content: None,
|
||
message_id: None,
|
||
trace_id: String::new(),
|
||
}
|
||
}
|
||
|
||
/// 构造执行完成信号
|
||
pub fn execution_completed(
|
||
channel: impl Into<String>,
|
||
chat_id: impl Into<String>,
|
||
session_id: Option<String>,
|
||
metadata: HashMap<String, String>,
|
||
) -> Self {
|
||
Self {
|
||
channel: channel.into(),
|
||
chat_id: chat_id.into(),
|
||
session_id,
|
||
content: String::new(),
|
||
reply_to: None,
|
||
media: Vec::new(),
|
||
metadata,
|
||
event_kind: OutboundEventKind::ExecutionCompleted,
|
||
role: "assistant".to_string(),
|
||
tool_call_id: None,
|
||
tool_name: None,
|
||
tool_arguments: None,
|
||
reasoning_content: None,
|
||
message_id: None,
|
||
trace_id: String::new(),
|
||
}
|
||
}
|
||
|
||
pub fn from_chat_message(
|
||
channel: &str,
|
||
chat_id: &str,
|
||
session_id: Option<String>,
|
||
reply_to: Option<String>,
|
||
metadata: &HashMap<String, String>,
|
||
message: &ChatMessage,
|
||
) -> Vec<Self> {
|
||
match message.role.as_str() {
|
||
"assistant" => {
|
||
if let Some(tool_calls) = &message.tool_calls {
|
||
let mut outbound = Vec::new();
|
||
let has_content_or_reasoning =
|
||
!message.content.trim().is_empty() || message.reasoning_content.is_some();
|
||
if has_content_or_reasoning {
|
||
let mut resp = Self::assistant(
|
||
channel.to_string(),
|
||
chat_id.to_string(),
|
||
session_id.clone(),
|
||
message.content.clone(),
|
||
reply_to.clone(),
|
||
metadata.clone(),
|
||
);
|
||
resp.reasoning_content = message.reasoning_content.clone();
|
||
resp.message_id = Some(message.id.clone());
|
||
outbound.push(resp);
|
||
}
|
||
|
||
// AssistantResponse 已携带 reasoning 时，ToolCall 不再重复；
|
||
// 只有 AssistantResponse 没发时，ToolCall 才带 reasoning
|
||
let tc_reasoning = if has_content_or_reasoning {
|
||
None
|
||
} else {
|
||
message.reasoning_content.clone()
|
||
};
|
||
outbound.extend(tool_calls.iter().map(|tool_call| {
|
||
let mut tc = Self::tool_call(
|
||
channel.to_string(),
|
||
chat_id.to_string(),
|
||
session_id.clone(),
|
||
tool_call.id.clone(),
|
||
tool_call.name.clone(),
|
||
tool_call.arguments.clone(),
|
||
reply_to.clone(),
|
||
metadata.clone(),
|
||
);
|
||
tc.reasoning_content = tc_reasoning.clone();
|
||
tc
|
||
}));
|
||
outbound
|
||
} else {
|
||
let mut resp = Self::assistant(
|
||
channel.to_string(),
|
||
chat_id.to_string(),
|
||
session_id,
|
||
message.content.clone(),
|
||
reply_to,
|
||
metadata.clone(),
|
||
);
|
||
resp.reasoning_content = message.reasoning_content.clone();
|
||
resp.message_id = Some(message.id.clone());
|
||
vec![resp]
|
||
}
|
||
}
|
||
"tool" => match message
|
||
.tool_state
|
||
.as_ref()
|
||
.unwrap_or(&ToolMessageState::Completed)
|
||
{
|
||
ToolMessageState::Completed => vec![Self::tool_result(
|
||
channel.to_string(),
|
||
chat_id.to_string(),
|
||
session_id,
|
||
message.tool_call_id.clone().unwrap_or_default(),
|
||
message.tool_name.clone().unwrap_or_default(),
|
||
message.content.clone(),
|
||
reply_to,
|
||
metadata.clone(),
|
||
)],
|
||
ToolMessageState::PendingUserAction => vec![Self::tool_pending(
|
||
channel.to_string(),
|
||
chat_id.to_string(),
|
||
session_id,
|
||
message.tool_call_id.clone().unwrap_or_default(),
|
||
message.tool_name.clone().unwrap_or_default(),
|
||
message.content.clone(),
|
||
reply_to,
|
||
metadata.clone(),
|
||
)],
|
||
},
|
||
_ => Vec::new(),
|
||
}
|
||
}
|
||
}
|
||
|
||
pub(crate) fn format_tool_call_content(
|
||
tool_name: &str,
|
||
tool_arguments: &serde_json::Value,
|
||
) -> String {
|
||
match tool_arguments {
|
||
serde_json::Value::Object(map) if map.is_empty() => tool_name.to_string(),
|
||
other => format!("{}\nargs: {}", tool_name, format_tool_arguments_json(other)),
|
||
}
|
||
}
|
||
|
||
fn format_tool_result_content(tool_name: &str, content: &str) -> String {
|
||
format!("工具结果: {}\n\n{}", tool_name, content)
|
||
}
|
||
|
||
fn format_tool_argument_value(value: &serde_json::Value) -> String {
|
||
match value {
|
||
serde_json::Value::String(text) => text.clone(),
|
||
serde_json::Value::Null => "null".to_string(),
|
||
other => serde_json::to_string(other).unwrap_or_else(|_| other.to_string()),
|
||
}
|
||
}
|
||
|
||
fn format_tool_arguments_json(value: &serde_json::Value) -> String {
|
||
match value {
|
||
serde_json::Value::Object(map) => {
|
||
let mut entries: Vec<_> = map.iter().collect();
|
||
entries.sort_by(|(left, _), (right, _)| left.cmp(right));
|
||
let body = entries
|
||
.into_iter()
|
||
.map(|(key, value)| {
|
||
format!(
|
||
"{}:{}",
|
||
serde_json::to_string(key).unwrap_or_else(|_| format!("\"{}\"", key)),
|
||
serde_json::to_string(value).unwrap_or_else(|_| value.to_string()),
|
||
)
|
||
})
|
||
.collect::<Vec<_>>()
|
||
.join(",");
|
||
format!("{{{}}}", body)
|
||
}
|
||
other => format_tool_argument_value(other),
|
||
}
|
||
}
|
||
|
||
// ============================================================================
|
||
// Helpers
|
||
// ============================================================================
|
||
|
||
#[cfg(test)]
|
||
mod tests {
|
||
use super::{ChatMessage, OutboundEventKind, OutboundMessage, ToolMessageState};
|
||
use crate::domain::messages::ToolCall;
|
||
use serde_json::json;
|
||
use std::collections::HashMap;
|
||
|
||
const TEST_CHANNEL: &str = "test-channel";
|
||
|
||
#[test]
|
||
fn test_from_chat_message_expands_tool_calls() {
|
||
let message = ChatMessage::assistant_with_tool_calls(
|
||
"",
|
||
vec![
|
||
ToolCall {
|
||
id: "call-1".to_string(),
|
||
name: "calculator".to_string(),
|
||
arguments: json!({"expression": "1 + 1"}),
|
||
},
|
||
ToolCall {
|
||
id: "call-2".to_string(),
|
||
name: "read".to_string(),
|
||
arguments: json!({"path": "README.md"}),
|
||
},
|
||
],
|
||
);
|
||
|
||
let outbound = OutboundMessage::from_chat_message(
|
||
TEST_CHANNEL,
|
||
"chat-1",
|
||
None,
|
||
None,
|
||
&HashMap::new(),
|
||
&message,
|
||
);
|
||
|
||
assert_eq!(outbound.len(), 2);
|
||
assert_eq!(outbound[0].event_kind, OutboundEventKind::ToolCall);
|
||
assert_eq!(outbound[0].tool_name.as_deref(), Some("calculator"));
|
||
assert_eq!(
|
||
outbound[0].tool_arguments.as_ref().unwrap()["expression"],
|
||
"1 + 1"
|
||
);
|
||
assert_eq!(
|
||
outbound[0].content,
|
||
"calculator\nargs: {\"expression\":\"1 + 1\"}"
|
||
);
|
||
assert_eq!(outbound[1].tool_name.as_deref(), Some("read"));
|
||
assert_eq!(outbound[1].content, "read\nargs: {\"path\":\"README.md\"}");
|
||
}
|
||
|
||
#[test]
|
||
fn test_from_chat_message_keeps_assistant_content_when_tool_calls_exist() {
|
||
let message = ChatMessage::assistant_with_tool_calls(
|
||
"日报已整理完成。",
|
||
vec![ToolCall {
|
||
id: "call-1".to_string(),
|
||
name: "memory_manage".to_string(),
|
||
arguments: json!({"action": "put"}),
|
||
}],
|
||
);
|
||
|
||
let outbound = OutboundMessage::from_chat_message(
|
||
TEST_CHANNEL,
|
||
"chat-1",
|
||
None,
|
||
None,
|
||
&HashMap::new(),
|
||
&message,
|
||
);
|
||
|
||
assert_eq!(outbound.len(), 2);
|
||
assert_eq!(outbound[0].event_kind, OutboundEventKind::AssistantResponse);
|
||
assert_eq!(outbound[0].content, "日报已整理完成。");
|
||
assert_eq!(outbound[1].event_kind, OutboundEventKind::ToolCall);
|
||
assert_eq!(outbound[1].tool_name.as_deref(), Some("memory_manage"));
|
||
}
|
||
|
||
#[test]
|
||
fn test_from_chat_message_includes_tool_result() {
|
||
let message = ChatMessage::tool("call-9", "calculator", "2");
|
||
|
||
let outbound = OutboundMessage::from_chat_message(
|
||
TEST_CHANNEL,
|
||
"chat-1",
|
||
None,
|
||
None,
|
||
&HashMap::new(),
|
||
&message,
|
||
);
|
||
|
||
assert_eq!(outbound.len(), 1);
|
||
assert_eq!(outbound[0].event_kind, OutboundEventKind::ToolResult);
|
||
}
|
||
|
||
#[test]
|
||
fn test_from_chat_message_includes_tool_pending() {
|
||
let message = ChatMessage::tool_with_state(
|
||
"call-9",
|
||
"bash",
|
||
"等待你完成浏览器授权后再继续。",
|
||
ToolMessageState::PendingUserAction,
|
||
);
|
||
|
||
let outbound = OutboundMessage::from_chat_message(
|
||
TEST_CHANNEL,
|
||
"chat-1",
|
||
None,
|
||
None,
|
||
&HashMap::new(),
|
||
&message,
|
||
);
|
||
|
||
assert_eq!(outbound.len(), 1);
|
||
assert_eq!(outbound[0].event_kind, OutboundEventKind::ToolPending);
|
||
}
|
||
}
|