diff --git a/backend/package/yuxi/services/agent_run_service.py b/backend/package/yuxi/services/agent_run_service.py index 6148361b..740df31e 100644 --- a/backend/package/yuxi/services/agent_run_service.py +++ b/backend/package/yuxi/services/agent_run_service.py @@ -339,29 +339,19 @@ async def get_active_run_by_thread(*, thread_id: str, current_uid: str, db: Asyn from sqlalchemy import select from yuxi.storage.postgres.models_business import AgentRun - active_result = await db.execute( + # 线程内的 run 是串行的，最近一条 run 即代表线程当前状态。 + # 已被回复的 interrupted run 会被更晚创建的 resume run 取代，因此不会再被当作待处理中断返回。 + result = await db.execute( select(AgentRun) .where( AgentRun.thread_id == thread_id, AgentRun.uid == str(current_uid), AgentRun.run_type.in_(["chat", "resume"]), - AgentRun.status.in_(["pending", "running", "cancel_requested"]), ) .order_by(AgentRun.created_at.desc()) .limit(1) ) - run = active_result.scalar_one_or_none() - if not run: - interrupted_result = await db.execute( - select(AgentRun) - .where( - AgentRun.thread_id == thread_id, - AgentRun.uid == str(current_uid), - AgentRun.run_type.in_(["chat", "resume"]), - AgentRun.status == "interrupted", - ) - .order_by(AgentRun.created_at.desc()) - .limit(1) - ) - run = interrupted_result.scalar_one_or_none() - return {"run": run.to_dict() if run else None} + run = result.scalar_one_or_none() + if run and run.status in ("pending", "running", "cancel_requested", "interrupted"): + return {"run": run.to_dict()} + return {"run": None} diff --git a/backend/test/unit/agents/toolkits/buildin/test_ask_user_question.py b/backend/test/unit/agents/toolkits/buildin/test_ask_user_question.py new file mode 100644 index 00000000..84a2ed32 --- /dev/null +++ b/backend/test/unit/agents/toolkits/buildin/test_ask_user_question.py @@ -0,0 +1,85 @@ +"""测试内置 ask_user_question 工具的格式契约。""" + +import json + +import pytest + +from yuxi.agents.toolkits.buildin import tools + + +def test_ask_user_question_interrupt_payload_and_result_format(monkeypatch): + captured_payloads = [] + expected_answer = {"style": "simple"} + + def fake_interrupt(payload): + captured_payloads.append(payload) + return expected_answer + + monkeypatch.setattr(tools, "interrupt", fake_interrupt) + + result = tools.ask_user_question.func( + questions=[ + { + "question_id": "style", + "question": "选择界面风格", + "options": [ + {"label": "简洁 (Recommended)", "value": "simple"}, + {"label": "详细", "value": "detailed"}, + ], + "multi_select": False, + "allow_other": False, + } + ] + ) + + expected_questions = [ + { + "question_id": "style", + "question": "选择界面风格", + "options": [ + {"label": "简洁 (Recommended)", "value": "simple"}, + {"label": "详细", "value": "detailed"}, + ], + "multi_select": False, + "allow_other": False, + } + ] + + assert captured_payloads == [{"questions": expected_questions, "source": "ask_user_question"}] + assert result == {"questions": expected_questions, "answer": expected_answer} + + +def test_ask_user_question_accepts_json_string_questions(monkeypatch): + captured_payloads = [] + + monkeypatch.setattr(tools, "interrupt", lambda payload: captured_payloads.append(payload) or {"q-1": "A"}) + + result = tools.ask_user_question.func( + questions=json.dumps( + [ + { + "question": "选择一个选项", + "options": ["A", "B"], + "allow_other": False, + } + ], + ensure_ascii=False, + ) + ) + + assert captured_payloads[0]["source"] == "ask_user_question" + assert captured_payloads[0]["questions"] == [ + { + "question_id": "q-1", + "question": "选择一个选项", + "options": [{"label": "A", "value": "A"}, {"label": "B", "value": "B"}], + "multi_select": False, + "allow_other": False, + } + ] + assert result["answer"] == {"q-1": "A"} + + +def test_ask_user_question_rejects_empty_questions(): + with pytest.raises(ValueError, match="questions 至少需要包含一个有效问题"): + tools.ask_user_question.func(questions=[]) diff --git a/web/src/components/AgentChatComponent.vue b/web/src/components/AgentChatComponent.vue index cc8c831e..fe1f6629 100644 --- a/web/src/components/AgentChatComponent.vue +++ b/web/src/components/AgentChatComponent.vue @@ -101,8 +101,8 @@
@@ -954,18 +954,38 @@ const currentThreadConfigNotice = computed(() => { return threadConfigNoticeMap.value[currentChatId.value] || null }) +const currentApprovalModalVisible = computed( + () => + approvalState.showModal && + Boolean(approvalState.threadId) && + approvalState.threadId === currentChatId.value +) +const currentApprovalQuestions = computed(() => + currentApprovalModalVisible.value ? approvalState.questions : [] +) + const shouldSuppressRefsForApproval = () => - approvalState.showModal || + currentApprovalModalVisible.value || Boolean( approvalState.threadId && - chatState.currentThreadId === approvalState.threadId && + currentChatId.value === approvalState.threadId && isProcessing.value ) // 计算是否显示Refs组件的条件 const shouldShowRefs = computed(() => { + const convs = conversations.value + const lastConv = convs.length ? convs[convs.length - 1] : null return (conv) => { - return getLastMessage(conv) && conv.status !== 'streaming' && !shouldSuppressRefsForApproval() + if (!getLastMessage(conv) || conv.status === 'streaming' || shouldSuppressRefsForApproval()) { + return false + } + // 回复生成中（含 resume 续写的空窗期）抑制最后一个对话的 refs，避免过早出现操作栏。 + // 同时看 isReplyLoading：切换/重连时 isStreaming 可能已置 false，但「正在生成回复」仍在显示。 + if (conv === lastConv && (isProcessing.value || isReplyLoading.value)) { + return false + } + return true } }) @@ -1510,6 +1530,10 @@ const startChatMainResizeObserver = () => { } onMounted(() => { + if (typeof document !== 'undefined') { + document.addEventListener('visibilitychange', handlePageVisibilityChange) + } + nextTick(() => { const chatMainContainer = document.querySelector('.chat-main') if (chatMainContainer) { @@ -1532,6 +1556,9 @@ onDeactivated(() => { }) onUnmounted(() => { + if (typeof document !== 'undefined') { + document.removeEventListener('visibilitychange', handlePageVisibilityChange) + } scrollController.cleanup() stopChatMainResizeObserver() stopStreamingStateRefresh() @@ -1719,11 +1746,21 @@ const handleAttachmentRemove = async (attachment) => { } // ==================== 审批功能管理 ==================== -const { approvalState, processApprovalInStream } = useApproval({ +const { + approvalState, + processApprovalInStream, + restoreInterruptFromThreadState, + hideApprovalState +} = useApproval({ getThreadState, fetchThreadMessages }) +const restorePendingInterruptForThread = (threadId) => { + if (!threadId) return false + return restoreInterruptFromThreadState(threadId) +} + const { handleStreamChunk } = useAgentStreamHandler({ getThreadState, processApprovalInStream, @@ -1739,9 +1776,35 @@ const { startRunStream, resumeActiveRunForThread, stopRunStreamSubscription } = fetchAgentState, resetOnGoingConv, onScrollToBottom: () => scrollController.scrollToBottom(), - streamSmoother + streamSmoother, + onInterruptDetected: ({ threadId }) => { + restorePendingInterruptForThread(threadId) + }, + onTerminalDetected: ({ threadId, touchedThreadIds = [] }) => { + if (approvalState.threadId === threadId || touchedThreadIds.includes(approvalState.threadId)) { + hideApprovalState() + } + } }) +const resumeCurrentRunForVisiblePage = async () => { + if (typeof document !== 'undefined' && document.visibilityState !== 'visible') return + const threadId = currentChatId.value + if (!threadId) return + + try { + await resumeActiveRunForThread(threadId) + restorePendingInterruptForThread(threadId) + } catch (error) { + console.warn('Failed to resume current run after page became visible:', error) + } +} + +const handlePageVisibilityChange = () => { + if (typeof document !== 'undefined' && document.visibilityState !== 'visible') return + void resumeCurrentRunForVisiblePage() +} + // ==================== CHAT ACTIONS ==================== // 获取第一个非置顶的对话 const getFirstNonPinnedChat = (chatList) => { @@ -1809,6 +1872,7 @@ const selectChat = async (chatId) => { await handleAgentStateRefresh(chatId) syncThreadConfigSnapshot(chatId, { overwrite: false }) await resumeActiveRunForThread(chatId) + restorePendingInterruptForThread(chatId) } const selectThreadFromRoute = async (threadId) => { @@ -1885,6 +1949,10 @@ const handleSendMessage = async ({ image } = {}) => { const threadState = getThreadState(threadId) if (!threadState) return + threadState.pendingInterrupt = null + if (approvalState.threadId === threadId) { + hideApprovalState() + } const pendingAttachments = [...currentPendingThreadAttachments.value] const pendingAttachmentFileIds = pendingAttachments @@ -1966,6 +2034,10 @@ const handleSendOrStop = async (payload) => { if (isProcessing.value && threadState?.activeRunId) { try { await agentApi.cancelAgentRun(threadState.activeRunId) + threadState.pendingInterrupt = null + if (approvalState.threadId === threadId) { + hideApprovalState() + } message.info('已发送取消请求') } catch (error) { handleChatError(error, 'stop') @@ -1979,6 +2051,7 @@ const handleSendOrStop = async (payload) => { // ==================== 人工审批处理 ==================== const handleApprovalWithStream = async (answer) => { const threadId = approvalState.threadId + const parentRunId = approvalState.parentRunId if (!threadId) { message.error('无效的提问请求') approvalState.showModal = false @@ -1992,14 +2065,17 @@ const handleApprovalWithStream = async (answer) => { return } - if (!approvalState.parentRunId) { + if (!parentRunId) { message.error('无法找到需要恢复的运行任务') approvalState.showModal = false return } + const pendingInterrupt = threadState.pendingInterrupt + try { - approvalState.showModal = false + hideApprovalState() + threadState.pendingInterrupt = null threadState.isStreaming = true resetOnGoingConv(threadId) const resumeRequestId = createClientRequestId() @@ -2009,7 +2085,7 @@ const handleApprovalWithStream = async (answer) => { thread_id: threadId, meta: { request_id: resumeRequestId }, resume: answer, - parent_run_id: approvalState.parentRunId, + parent_run_id: parentRunId, resume_request_id: resumeRequestId }) const runId = runResp?.run_id @@ -2018,6 +2094,10 @@ const handleApprovalWithStream = async (answer) => { } await startRunStream(threadId, runId, '0-0') } catch (error) { + if (pendingInterrupt) { + threadState.pendingInterrupt = pendingInterrupt + restorePendingInterruptForThread(threadId) + } threadState.isStreaming = false threadState.replyLoadingVisible = false handleChatError(error, 'resume') @@ -2179,6 +2259,12 @@ const showMsgRefs = (msg) => { return false } + // 回复生成中（含 resume 续写的空窗期）不在最后一条消息上过早显示操作栏/来源。 + // isReplyLoading 兜底：切换/重连时 isStreaming 可能已置 false，但回复仍在生成。 + if (msg.isLast && (isProcessing.value || isReplyLoading.value)) { + return false + } + // 只有真正完成的消息才显示 refs if (msg.isLast && msg.status === 'finished') { return ['copy', 'sources'] @@ -2333,6 +2419,12 @@ watch( watch(currentChatId, (threadId, oldThreadId) => { if (threadId === oldThreadId) return + if (!threadId || approvalState.threadId !== threadId) { + hideApprovalState() + } + if (threadId) { + restorePendingInterruptForThread(threadId) + } emit('thread-change', threadId || '') }) diff --git a/web/src/components/ToolCallingResult/tools/AskUserQuestionTool.vue b/web/src/components/ToolCallingResult/tools/AskUserQuestionTool.vue index 4f2f8ae5..58bd0a42 100644 --- a/web/src/components/ToolCallingResult/tools/AskUserQuestionTool.vue +++ b/web/src/components/ToolCallingResult/tools/AskUserQuestionTool.vue @@ -1,11 +1,45 @@