ooodc/PicoBot
1
0
Fork 0
Code Pull Requests Activity

feat: 添加 user_message_id 字段以支持用户消息的关联和同步

Browse Source
This commit is contained in:
oudecheng 2026-06-24 15:13:36 +08:00
parent 7f262c9af2
commit c72853c79d
9 changed files with 61 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
src/command/adapters/websocket.rs
View File

@ -79,7 +79,7 @@ impl OutputAdapter for WebSocketOutputAdapter {
content: msg.content.clone(),
role: "assistant".to_string(),
attachments: Vec::new(), subagent_task_id: None, topic_id: None, timestamp: Some(crate::protocol::now_timestamp()),
reasoning_content: None,
reasoning_content: None, user_message_id: None,
},
MessageKind::Notification => {
// 根据元数据判断具体类型
@ -100,7 +100,7 @@ impl OutputAdapter for WebSocketOutputAdapter {
content: msg.content.clone(),
role: "assistant".to_string(),
attachments: Vec::new(), subagent_task_id: None, topic_id: None, timestamp: Some(crate::protocol::now_timestamp()),
reasoning_content: None,
reasoning_content: None, user_message_id: None,
},
}
} else if let Some(session_id) = response.metadata.get("session_id") {
@ -140,7 +140,7 @@ impl OutputAdapter for WebSocketOutputAdapter {
content: msg.content.clone(),
role: "assistant".to_string(),
attachments: Vec::new(), subagent_task_id: None, topic_id: None, timestamp: Some(crate::protocol::now_timestamp()),
reasoning_content: None,
reasoning_content: None, user_message_id: None,
},
}
} else if let Some(sessions_json) = response.metadata.get("sessions") {
@ -159,7 +159,7 @@ impl OutputAdapter for WebSocketOutputAdapter {
content: msg.content.clone(),
role: "assistant".to_string(),
attachments: Vec::new(), subagent_task_id: None, topic_id: None, timestamp: Some(crate::protocol::now_timestamp()),
reasoning_content: None,
reasoning_content: None, user_message_id: None,
},
}
} else if let Some(topics_json) = response.metadata.get("topics") {
@ -179,7 +179,7 @@ impl OutputAdapter for WebSocketOutputAdapter {
content: msg.content.clone(),
role: "assistant".to_string(),
attachments: Vec::new(), subagent_task_id: None, topic_id: None, timestamp: Some(crate::protocol::now_timestamp()),
reasoning_content: None,
reasoning_content: None, user_message_id: None,
},
}
} else {
@ -189,7 +189,7 @@ impl OutputAdapter for WebSocketOutputAdapter {
content: msg.content.clone(),
role: "assistant".to_string(),
attachments: Vec::new(), subagent_task_id: None, topic_id: None, timestamp: Some(crate::protocol::now_timestamp()),
reasoning_content: None,
reasoning_content: None, user_message_id: None,
}
}
}
@ -203,7 +203,7 @@ impl OutputAdapter for WebSocketOutputAdapter {
content: msg.content.clone(),
role: "assistant".to_string(),
attachments: Vec::new(), subagent_task_id: None, topic_id: None, timestamp: Some(crate::protocol::now_timestamp()),
reasoning_content: None,
reasoning_content: None, user_message_id: None,
},
};
outbounds.push(outbound);

4
src/gateway/execution.rs
View File

@ -258,7 +258,9 @@ impl AgentExecutionService {
};
let result = agent.process(history, Some(&system_prompt_context)).await?;
let metadata = HashMap::new();
let mut metadata = HashMap::new();
// 把用户消息的 UUID 回传给前端，前端用此更新本地消息 ID，使 todo 点击跳转能匹配
metadata.insert("user_message_id".to_string(), user_message.id.clone());
self.finalize_result_and_schedule_compaction(
request.session.clone(),

7
src/gateway/session.rs
View File

@ -200,7 +200,12 @@ impl BusToolCallEmitter {
priority: "medium".to_string(),
created_at: now + idx as i64,
updated_at: now,
created_by_message_id: Some(message.id.clone()),
created_by_message_id: item
.get("created_by_message_id")
.and_then(|v| v.as_str())
.filter(|s| !s.is_empty())
.map(|s| s.to_string())
.or_else(|| Some(message.id.clone())),
})
})
.collect();

4
src/gateway/ws.rs
View File

@ -857,6 +857,7 @@ fn chat_message_to_ws_outbound(msg: &crate::bus::ChatMessage) -> Vec<WsOutbound>
topic_id: None,
timestamp: Some(msg.timestamp / 1000),
reasoning_content: msg.reasoning_content.clone(),
user_message_id: None,
});
}
// AssistantResponse 已携带 reasoning 时，ToolCall 不再重复
@ -873,6 +874,7 @@ fn chat_message_to_ws_outbound(msg: &crate::bus::ChatMessage) -> Vec<WsOutbound>
topic_id: None,
timestamp: Some(msg.timestamp / 1000),
reasoning_content: tc_reasoning.clone(),
user_message_id: None,
});
}
outbound
@ -887,6 +889,7 @@ fn chat_message_to_ws_outbound(msg: &crate::bus::ChatMessage) -> Vec<WsOutbound>
topic_id: None,
timestamp: Some(msg.timestamp / 1000),
reasoning_content: msg.reasoning_content.clone(),
user_message_id: None,
}]
}
}
@ -926,6 +929,7 @@ fn chat_message_to_ws_outbound(msg: &crate::bus::ChatMessage) -> Vec<WsOutbound>
topic_id: None,
timestamp: Some(msg.timestamp / 1000),
reasoning_content: None,
user_message_id: None,
}],
_ => Vec::new(),
}

6
src/protocol/mod.rs
View File

@ -150,6 +150,8 @@ pub enum WsOutbound {
timestamp: Option<i64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
reasoning_content: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
user_message_id: Option<String>,
},
#[serde(rename = "tool_call")]
ToolCall {
@ -167,6 +169,8 @@ pub enum WsOutbound {
timestamp: Option<i64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
reasoning_content: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
user_message_id: Option<String>,
},
#[serde(rename = "tool_result")]
ToolResult {
@ -280,6 +284,8 @@ pub enum WsOutbound {
subagent_task_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
topic_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
user_message_id: Option<String>,
},
#[serde(rename = "stream_end")]
StreamEnd {

3
src/protocol/ws_adapter.rs
View File

@ -112,6 +112,7 @@ pub(crate) fn ws_outbound_from_outbound_message(message: &OutboundMessage) -> Ve
topic_id: message.metadata.get("topic_id").cloned(),
timestamp: Some(crate::protocol::now_timestamp()),
reasoning_content: message.reasoning_content.clone(),
user_message_id: message.metadata.get("user_message_id").cloned(),
}]
}
OutboundEventKind::ToolCall => vec![WsOutbound::ToolCall {
@ -131,6 +132,7 @@ pub(crate) fn ws_outbound_from_outbound_message(message: &OutboundMessage) -> Ve
topic_id: message.metadata.get("topic_id").cloned(),
timestamp: Some(crate::protocol::now_timestamp()),
reasoning_content: message.reasoning_content.clone(),
user_message_id: message.metadata.get("user_message_id").cloned(),
}],
OutboundEventKind::ToolResult => vec![WsOutbound::ToolResult {
id: message
@ -182,6 +184,7 @@ pub(crate) fn ws_outbound_from_outbound_message(message: &OutboundMessage) -> Ve
reasoning_delta: message.reasoning_content.clone(),
subagent_task_id: message.metadata.get("subagent_task_id").cloned(),
topic_id: message.metadata.get("topic_id").cloned(),
user_message_id: message.metadata.get("user_message_id").cloned(),
}],
OutboundEventKind::StreamEnd => vec![WsOutbound::StreamEnd {
id: message.tool_call_id.clone().unwrap_or_default(),

7
src/tools/task/runtime.rs
View File

@ -235,7 +235,12 @@ impl SubAgentEmitter {
priority: "medium".to_string(),
created_at: now + idx as i64,
updated_at: now,
created_by_message_id: Some(message.id.clone()),
created_by_message_id: item
.get("created_by_message_id")
.and_then(|v| v.as_str())
.filter(|s| !s.is_empty())
.map(|s| s.to_string())
.or_else(|| Some(message.id.clone())),
})
})
.collect();

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

@ -167,6 +167,9 @@ export function useChat(): UseChatReturn {
const [channels, setChannels] = useState<Channel[]>([])
const [selectedChannel, setSelectedChannel] = useState<string>('websocket')
// Track user message IDs already synced from backend to avoid duplicate updates
const syncedUserMessageIdsRef = useRef<Set<string>>(new Set())
// Message ID generator
const messageIdCounter = useRef(0)
const generateMessageId = () => {
@ -356,6 +359,23 @@ export function useChat(): UseChatReturn {
}
}
// Sync backend user message ID to the last local user message,
// so that created_by_message_id (backend UUID) can match DOM data-message-id
const applyUserMessageId = useCallback((userMessageId: string) => {
if (syncedUserMessageIdsRef.current.has(userMessageId)) return
syncedUserMessageIdsRef.current.add(userMessageId)
setMessages(prev => {
for (let i = prev.length - 1; i >= 0; i--) {
if (prev[i].role === 'user') {
const updated = [...prev]
updated[i] = { ...updated[i], id: userMessageId }
return updated
}
}
return prev
})
}, [])
const handleServerMessage = useCallback((message: WsOutbound) => {
console.log('Received message:', message)
@ -639,6 +659,7 @@ export function useChat(): UseChatReturn {
]
})
setIsLoading(false)
if (msg.user_message_id) applyUserMessageId(msg.user_message_id)
break
}
@ -678,6 +699,7 @@ export function useChat(): UseChatReturn {
if (currentTopic && !currentTopic.description) {
setTopicRefreshTrigger(n => n + 1)
}
if (msg.user_message_id) applyUserMessageId(msg.user_message_id)
break
}
@ -700,6 +722,7 @@ export function useChat(): UseChatReturn {
reasoningContent: msg.reasoning_content,
},
])
if (msg.user_message_id) applyUserMessageId(msg.user_message_id)
break
}

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

@ -46,6 +46,7 @@ export interface AssistantResponse {
topic_id?: string
timestamp?: number
reasoning_content?: string
user_message_id?: string
}
export interface ToolCall {
@ -60,6 +61,7 @@ export interface ToolCall {
topic_id?: string
timestamp?: number
reasoning_content?: string
user_message_id?: string
}
export interface ToolResult {
@ -266,6 +268,7 @@ export interface StreamDelta {
reasoning_delta?: string
subagent_task_id?: string
topic_id?: string
user_message_id?: string
}
export interface StreamEnd {